如何解决使用 Celery 和 Gino 保存对象
我有以下管道
事件模型(基于 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 视图,调用 celery 任务来传递事件对象的数据:
@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,)
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,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,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",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",protocol)
sqlalchemy.exc.SQLAlchemyError: (builtins.TypeError) cannot pickle 'coroutine' object
如果我在视图中调用 event = await Event.create(...)
,它会成功并将数据保存到数据库。
如果我从 Celery worker 中删除 'async/await',则不会出现此错误,Celery 任务具有 'SUCCESS' 状态但数据库为空。
无法真正理解什么是不对的。
解决方法
简单:目前,Celery 无法将异步函数作为任务处理。你需要用 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,)
@celery_app.task(acks_late=True)
def add_to_db_task(request_timestamp,response_time):
asyncio.run(add_to_db(request_timestamp,response_time))
return {"status": True}
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。