我有以下管道
事件模型(基于Gino 'db'对象):
class Event(db.Model):
__tablename__ = "events"
id = db.Column(db.BigInteger(), primary_key=True)
request_timestamp = db.Column(db.DateTime(), nullable=False)
service = db.Column(db.String(), nullable=False)
url = db.Column(db.String(), nullable=False)
status_code = db.Column(db.Integer(), nullable=False)
response_time = db.Column(db.DateTime(), nullable=False)
FastApi应用程序有一个POST视图调用芹菜任务传递事件对象的数据:
@router.post("/events/add")
async def add_event(event: EventModel):
event_data = {'request_timestamp': event.request_timestamp.replace(tzinfo=None),
'service': event.service,
'url': event.url,
'status_code': event.status_code,
'response_time': event.response_time.replace(tzinfo=None)
}
task = celery_app.send_task("monitoring_service.src.monitoring_service.worker.celery_worker.add_to_db",
kwargs=event_data)
return JSONResponse(content="Event recorded successfully", status_code=200)
和必须保存Event对象到数据库的celery worker:
@celery_app.task(acks_late=True)
async def add_to_db(request_timestamp, service, url, status_code, response_time):
event = await Event.create(
request_timestamp=datetime.strptime(request_timestamp, '%Y-%m-%dT%H:%M:%S.%f'),
service=service,
url=url,
status_code=status_code,
response_time=datetime.strptime(response_time, '%Y-%m-%dT%H:%M:%S.%f'),
)
return {"status": True}
现在我在对象保存过程中收到一个错误:
Traceback (most recent call last):
File "/usr/local/lib/python3.8/site-packages/celery/app/trace.py", line 479, in trace_task
mark_as_done(
File "/usr/local/lib/python3.8/site-packages/celery/backends/base.py", line 158, in mark_as_done
self.store_result(task_id, result, state, request=request)
File "/usr/local/lib/python3.8/site-packages/celery/backends/base.py", line 442, in store_result
self._store_result(task_id, result, state, traceback,
File "/usr/local/lib/python3.8/site-packages/celery/backends/database/__init__.py", line 51, in _inner
return fun(*args, **kwargs)
File "/usr/local/lib/python3.8/site-packages/celery/backends/database/__init__.py", line 130, in _store_result
session.commit()
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 1042, in commit
self.transaction.commit()
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 504, in commit
self._prepare_impl()
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 483, in _prepare_impl
self.session.flush()
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 2523, in flush
self._flush(objects)
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 2664, in _flush
transaction.rollback(_capture_exception=True)
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/util/langhelpers.py", line 68, in __exit__
compat.raise_(
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/util/compat.py", line 178, in raise_
raise exception
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 2624, in _flush
flush_context.execute()
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/orm/unitofwork.py", line 422, in execute
rec.execute(self)
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/orm/unitofwork.py", line 586, in execute
persistence.save_obj(
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/orm/persistence.py", line 230, in save_obj
_emit_update_statements(
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/orm/persistence.py", line 994, in _emit_update_statements
c = cached_connections[connection].execute(
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1014, in execute
return meth(self, multiparams, params)
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/sql/elements.py", line 298, in _execute_on_connection
return connection._execute_clauseelement(self, multiparams, params)
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1127, in _execute_clauseelement
ret = self._execute_context(
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1207, in _execute_context
self._handle_dbapi_exception(
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1511, in _handle_dbapi_exception
util.raise_(
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/util/compat.py", line 178, in raise_
raise exception
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1205, in _execute_context
context = constructor(dialect, self, conn, *args)
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/engine/default.py", line 858, in _init_compiled
param = dict(
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/engine/default.py", line 861, in <genexpr>
processors[key](compiled_params[key])
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/sql/sqltypes.py", line 1689, in process
value = dumps(value, protocol)
sqlalchemy.exc.SQLAlchemyError: (builtins.TypeError) cannot pickle 'coroutine' object
warn(RuntimeWarning(
[2021-02-19 10:19:39,010: ERROR/ForkPoolWorker-1] Task monitoring_service.src.monitoring_service.worker.celery_worker.add_to_db[994ae754-f216-4fa9-b502-5fbb0221011c] raised unexpected: SQLAlchemyError("(builtins.TypeError) cannot pickle 'coroutine' object")
Traceback (most recent call last):
File "/usr/local/lib/python3.8/site-packages/celery/app/trace.py", line 479, in trace_task
mark_as_done(
File "/usr/local/lib/python3.8/site-packages/celery/backends/base.py", line 158, in mark_as_done
self.store_result(task_id, result, state, request=request)
File "/usr/local/lib/python3.8/site-packages/celery/backends/base.py", line 442, in store_result
self._store_result(task_id, result, state, traceback,
File "/usr/local/lib/python3.8/site-packages/celery/backends/database/__init__.py", line 51, in _inner
return fun(*args, **kwargs)
File "/usr/local/lib/python3.8/site-packages/celery/backends/database/__init__.py", line 130, in _store_result
session.commit()
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 1042, in commit
self.transaction.commit()
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 504, in commit
self._prepare_impl()
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 483, in _prepare_impl
self.session.flush()
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 2523, in flush
self._flush(objects)
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 2664, in _flush
transaction.rollback(_capture_exception=True)
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/util/langhelpers.py", line 68, in __exit__
compat.raise_(
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/util/compat.py", line 178, in raise_
raise exception
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 2624, in _flush
flush_context.execute()
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/orm/unitofwork.py", line 422, in execute
rec.execute(self)
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/orm/unitofwork.py", line 586, in execute
persistence.save_obj(
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/orm/persistence.py", line 230, in save_obj
_emit_update_statements(
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/orm/persistence.py", line 994, in _emit_update_statements
c = cached_connections[connection].execute(
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1014, in execute
return meth(self, multiparams, params)
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/sql/elements.py", line 298, in _execute_on_connection
return connection._execute_clauseelement(self, multiparams, params)
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1127, in _execute_clauseelement
ret = self._execute_context(
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1207, in _execute_context
self._handle_dbapi_exception(
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1511, in _handle_dbapi_exception
util.raise_(
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/util/compat.py", line 178, in raise_
raise exception
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1205, in _execute_context
context = constructor(dialect, self, conn, *args)
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/engine/default.py", line 858, in _init_compiled
param = dict(
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/engine/default.py", line 861, in <genexpr>
processors[key](compiled_params[key])
File "/usr/local/lib/python3.8/site-packages/sqlalchemy/sql/sqltypes.py", line 1689, in process
value = dumps(value, protocol)
sqlalchemy.exc.SQLAlchemyError: (builtins.TypeError) cannot pickle 'coroutine' object
如果我在视图中调用event = await Event.create(...)，它是成功的，并将数据保存到db。如果我从芹菜工作器中删除'async/await'，则此错误不会出现，芹菜任务具有'SUCCESS'状态，但数据库为空。不能真正理解什么是不对的。
简单:目前，芹菜不能处理异步函数作为任务。您需要用asyncio.run(https://docs.python.org/3/library/asyncio-task.html#asyncio.run):
async def add_to_db(request_timestamp, service, url, status_code, response_time):
event = await Event.create(
request_timestamp=datetime.strptime(request_timestamp, '%Y-%m-%dT%H:%M:%S.%f'),
service=service,
url=url,
status_code=status_code,
response_time=datetime.strptime(response_time, '%Y-%m-%dT%H:%M:%S.%f'),
)
@celery_app.task(acks_late=True)
def add_to_db_task(request_timestamp, service, url, status_code, response_time):
asyncio.run(add_to_db(request_timestamp, service, url, status_code, response_time))
return {"status": True}