我有一个实例方法,我想作为演员/工人同时运行。跑步时process_request.send(service, url)
,我得到TypeError: Object of type 'Foo' is not JSON serializable
services.py
class Foo:
def __init__(self, name):
self.name = name
def get_data(self, url):
return requests.get(url)
class Bar:
def __init__(self, name):
self.name = name
def get_data(self, url):
return requests.get(url)
tasks.py
import dramatiq
@dramatiq.actor
def process_request(cls_instance, url):
data = cls_instance(url)
save_data(data, cls_instance.name)
main.py
import services
from tasks import process_request
if __name__ == '__main__':
if args[0] == 'foo':
service = services.Foo('foo')
else:
service = services.Bar('bar')
urls = [...]
for url in urls:
process_request.send(service, url)