Update the celery conf

Author: wu-clan

backend/app/task/celery.py

@@ -19,31 +19,23 @@ def init_celery() -> celery.Celery:
 
     app = celery.Celery(
         'fba_celery',
-        enable_utc=False,
-        timezone=settings.DATETIME_TIMEZONE,
-        beat_schedule=LOCAL_BEAT_SCHEDULE,
-        broker_url=(
-            f'redis://:{settings.REDIS_PASSWORD}@{settings.REDIS_HOST}:'
-            f'{settings.REDIS_PORT}/{settings.CELERY_BROKER_REDIS_DATABASE}'
-        )
+        broker=f'redis://:{settings.REDIS_PASSWORD}@{settings.REDIS_HOST}:{settings.REDIS_PORT}/{settings.CELERY_BROKER_REDIS_DATABASE}'
         if settings.CELERY_BROKER == 'redis'
-        else (
-            f'amqp://{settings.CELERY_RABBITMQ_USERNAME}:{settings.CELERY_RABBITMQ_PASSWORD}@'
-            f'{settings.CELERY_RABBITMQ_HOST}:{settings.CELERY_RABBITMQ_PORT}'
-        ),
+        else f'amqp://{settings.CELERY_RABBITMQ_USERNAME}:{settings.CELERY_RABBITMQ_PASSWORD}@{settings.CELERY_RABBITMQ_HOST}:{settings.CELERY_RABBITMQ_PORT}',
         broker_connection_retry_on_startup=True,
-        result_backend=(
-            f'redis://:{settings.REDIS_PASSWORD}@{settings.REDIS_HOST}:'
-            f'{settings.REDIS_PORT}/{settings.CELERY_BACKEND_REDIS_DATABASE}'
-        ),
-        result_backend_transport_options={
-            'global_keyprefix': settings.CELERY_BACKEND_REDIS_PREFIX,
-            'retry_policy': {
-                'timeout': settings.CELERY_BACKEND_REDIS_TIMEOUT,
-            },
-        },
+        backend=f'db+{settings.DATABASE_TYPE + "+pymysql" if settings.DATABASE_TYPE == "mysql" else settings.DATABASE_TYPE}'  # noqa: E501
+        f'://{settings.DATABASE_USER}:{settings.DATABASE_PASSWORD}@{settings.DATABASE_HOST}:{settings.DATABASE_PORT}/{settings.DATABASE_SCHEMA}',
+        database_engine_options={'echo': settings.DATABASE_ECHO},
+        result_extended=True,
+        result_expires=0,
+        beat_schedule=LOCAL_BEAT_SCHEDULE,
+        beat_scheduler='app.task.utils.schedulers:DatabaseScheduler',
         task_cls='app.task.tasks.base:TaskBase',
         task_track_started=True,
+        enable_utc=False,
+        timezone=settings.DATETIME_TIMEZONE,
+        # TODO: Update this work if celery version >= 6.0.0
+        worker_pool=celery_aio_pool.pool.AsyncIOPool,
     )
 
     # 自动发现任务

pyproject.toml

@@ -42,6 +42,7 @@ dependencies = [
     "pwdlib>=0.2.1",
     "pydantic>=2.11.0",
     "pydantic-settings>=2.0.0",
+    "pymysql>=1.1.1",
     "python-jose>=3.3.0",
     "python-socketio>=5.12.0",
     "redis[hiredis]>=5.2.0",

requirements.txt

@@ -201,6 +201,8 @@ pydantic-settings==2.8.1
     # via fastapi-best-architecture
 pygments==2.19.1
     # via rich
+pymysql==1.1.1
+    # via fastapi-best-architecture
 pytest==8.3.5
     # via pytest-sugar
 pytest-sugar==1.0.0