wren
65 lines
1.8 KiB
Python
65 lines
1.8 KiB
Python
"""Celery 应用入口。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from celery import Celery
|
|
from celery.schedules import crontab
|
|
from kombu import Queue
|
|
|
|
from fastapi_admin.config import (
|
|
LEAUDIT_TASK_SOFT_TIME_LIMIT,
|
|
LEAUDIT_TASK_TIME_LIMIT,
|
|
LEAUDIT_STUCK_SCAN_CRON_MINUTES,
|
|
LEAUDIT_WORKER_QUEUE_NORMAL,
|
|
LEAUDIT_WORKER_QUEUE_URGENT,
|
|
REDIS_DB,
|
|
REDIS_HOST,
|
|
REDIS_PASSWORD,
|
|
REDIS_PORT,
|
|
)
|
|
|
|
|
|
def _build_redis_url() -> str:
|
|
"""拼接 Redis broker / backend 连接串。"""
|
|
auth = f":{REDIS_PASSWORD}@" if REDIS_PASSWORD else ""
|
|
return f"redis://{auth}{REDIS_HOST}:{REDIS_PORT}/{REDIS_DB}"
|
|
|
|
|
|
celery_app = Celery(
|
|
"leaudit_platform",
|
|
broker=_build_redis_url(),
|
|
backend=_build_redis_url(),
|
|
)
|
|
|
|
celery_app.conf.update(
|
|
task_default_queue=LEAUDIT_WORKER_QUEUE_NORMAL,
|
|
imports=("fastapi_modules.fastapi_leaudit.leaudit_bridge.tasks",),
|
|
task_queues=(
|
|
Queue(LEAUDIT_WORKER_QUEUE_URGENT),
|
|
Queue(LEAUDIT_WORKER_QUEUE_NORMAL),
|
|
),
|
|
task_track_started=True,
|
|
task_acks_late=True,
|
|
worker_prefetch_multiplier=1,
|
|
broker_connection_retry_on_startup=True,
|
|
task_soft_time_limit=LEAUDIT_TASK_SOFT_TIME_LIMIT,
|
|
task_time_limit=LEAUDIT_TASK_TIME_LIMIT,
|
|
beat_schedule={
|
|
"leaudit-scan-stuck-documents": {
|
|
"task": "leaudit.scan_stuck_documents",
|
|
"schedule": crontab(minute=f"*/{max(1, int(LEAUDIT_STUCK_SCAN_CRON_MINUTES))}"),
|
|
"options": {"queue": LEAUDIT_WORKER_QUEUE_NORMAL},
|
|
}
|
|
},
|
|
)
|
|
|
|
celery_app.autodiscover_tasks(
|
|
[
|
|
"fastapi_modules.fastapi_leaudit.leaudit_bridge",
|
|
],
|
|
force=True,
|
|
)
|
|
|
|
# 显式导入任务模块，避免 worker 在某些启动方式下漏注册 bridge tasks。
|
|
from fastapi_modules.fastapi_leaudit.leaudit_bridge import tasks as _leaudit_bridge_tasks # noqa: F401,E402
|