mirror of
https://github.com/fastapi-practices/fastapi-best-architecture.git
synced 2026-09-21 21:15:13 +00:00
Wu Clan
5762834744
* Update the ruff rules and format the code * Update the per-file-ignores * Update the ci * Update rules * Fix codes * Fix pagination * Update rules
12 lines
378 B
Python
12 lines
378 B
Python
from starlette.concurrency import run_in_threadpool
|
|
|
|
from backend.app.task.celery import celery_app
|
|
from backend.common.socketio.server import sio
|
|
|
|
|
|
@sio.event
|
|
async def task_worker_status(sid, data) -> None: # noqa: ANN001
|
|
"""任务 Worker 状态事件"""
|
|
worker = await run_in_threadpool(celery_app.control.ping)
|
|
await sio.emit('task_worker_status', worker, sid)
|