我在使用异步ODM时遇到了多个问题,主要是在Celery工作器内。一开始我无法使用Celery工作器信号来初始化我的数据库模型,我正在使用Beanie进行数据库连接。
第一个实现方式
from asyncer import syncify
from asgiref.sync import async_to_sync
client = AsyncIOMotorClient(
DATABASE_URL, uuidRepresentation="standard" )
db = client[DB_NAME]
async def db_session():
await init_beanie(
database=db,
document_models=[Project, User],
)
@worker_ready.connect
def startup_celery_ecosystem(**kwargs):
logger.info('Startup celery worker process')
async_to_sync(db_session)()
logger.info('FINISHED : Startup celery worker process')
async def get_users():
users = User.find()
users_list = await users.to_list()
return users_list
@celery_app.task
def pool_db():
async_to_sync(get_users)()
#syncify(get_users)() same error User class is not initialized yet (init_beanie should have already initialized all the models )
使用这种实现方式,我无法通过User和Project类访问我的数据库,并且会产生一个错误,就好像User和Project还没有被实例化一样。
解决方法是在模块级别调用db_session(),这可以解决数据库模型实例化的问题,但现在当查询数据库时,我的celery任务会出现以下错误:
运行时错误:事件循环已关闭
第二个实现
from asyncer import syncify
from asgiref.sync import async_to_sync client = AsyncIOMotorClient(
DATABASE_URL, uuidRepresentation="standard" )
db = client[DB_NAME]
async def db_session():
await init_beanie(
database=db,
document_models=[Project, User],
)
# now init_beanie at module level
async_to_sync(db_session)()
async def get_users():
users = User.find()
users_list = await users.to_list()
return users_list
@celery_app.task
def pool_db():
# this raises the following Runtime error RuntimeError('Event loop is closed')
async_to_sync(get_users)()
#syncify(get_users)() same error
我对asyncio的实现方式以及asyncer和asgiref如何允许在同步线程中运行异步代码并感到困惑,希望能得到帮助。