feat(celery): add basic framework for celery

backend/cli/worker.py

@@ -0,0 +1,59 @@
+# -*- coding: utf-8 -*-
+
+import logging
+import sys
+
+import click
+from celery.__main__ import main as celery_main
+from rich.console import Console
+
+console = Console()
+log = logging.getLogger(__name__)
+
+
+@click.command()
+@click.option(
+ "--identifier", help="Identifier worker in case multiple operate in parallel"
+)
+@click.option(
+ "--queue",
+ type=str,
+ help="Selected queue to deal with",
+ default=None,
+)
+@click.option("--concurrency", type=int, help="Number of concurrent workers", default=1)
+@click.option(
+ "--loglevel", type=str, help="error, debug, warning, info", default="info"
+)
+def worker(identifier, queue, concurrency, loglevel):
+ """Run the celery worker"""
+ # dynamically load all of the the tasks of our stack
+ queues = {"celery", "general"}
+ if not queue:
+ queue = ",".join(list(queues))
+
+ log.info(f"Running on queues: {queues}")
+
+ sys.argv = [
+ "celery",
+ "-A",
+ "backend.tasks.celery",
+ "worker",
+ "--loglevel={}".format(loglevel.upper()),
+ # Fair scheduler
+ "-Ofair",
+ # make use of multiprocessing since libs are not thread-safe
+ "--pool=prefork",
+ # Concurrency of multiple workers at once
+ "--concurrency={}".format(concurrency),
+ # Send task-related events that can be captured by monitors like celery
+ # events, celerymon, and others.
+ "--task-events",
+ # Queues to process
+ "-Q",
+ queue,
+ ]
+
+ if identifier:
+ sys.argv.extend(["-n", identifier])
+ sys.exit(celery_main())

backend/config.py

@@ -18,6 +18,23 @@ class SQLAlchemyEngineOptions(BaseSettings):
  max_overflow: Optional[int] = 64
 
 
+class CelerySettings(BaseSettings):
+ """Celery Configurations
+
+ Allows to specify the BROKER_URL and RESULT_BACKEND
+ """
+
+ broker_url: str = "redis:https://localhost:6379"
+ result_backend: str = "redis:https://localhost:6379"
+
+ beat_schedule: dict = {
+ # "check": {
+ # "task": "backend.tasks.regular",
+ # "schedule": 60 * 60 * 24, # ever day once
+ # },
+ }
+
+
 class Settings(BaseSettings):
  """General Settings"""
 
@@ -29,6 +46,9 @@ class Settings(BaseSettings):
  SQLALCHEMY_TRACK_MODIFICATIONS: bool = False
  SQLALCHEMY_ECHO: bool = False
 
+ # Celery
+ CELERY_CONF: CelerySettings = CelerySettings()
+
  # Admins
  ADMIN_USER_IDS: List[str] = []
 

entrypoint.sh

@@ -1,5 +1,8 @@
 #!/bin/bash
 
+# fail script when command fails and print the commands and arguments
+set -xe
+
 export APP_RUN=${APP_RUN:=api}
 export LOG_LEVEL=${APP_LOGLEVEL:=critical}
 

main.py

@@ -19,6 +19,7 @@ def main():
  "backend.cli.document_body",
  "backend.cli.access_token",
  "backend.cli.asset",
+ "backend.cli.worker",
 ]
 
 for cli in clis:

requirements.txt

@@ -42,3 +42,10 @@ stripe>=7.9.0,<7.10
 
 sentry-sdk[fastapi]
 xmltodict==0.13.0
+
+pycountry
+
+# Celery
+redis>=4.4.0rc4
+celery==5.2.2
+flower==1.2.0