Merge pull request #65 from febus982/celery

Celery

Author: febus982

.coveragerc

@@ -1,7 +1,7 @@
 [run]
 branch = True
-source = http_app, grpc_app, domains, gateways
-omit = grpc_app/generated/*
+source = http_app, grpc_app, domains, gateways, celery_worker, common
+omit = grpc_app/generated/*, common/config.py
 concurrency = multiprocessing
 parallel = true
 

Dockerfile

@@ -49,6 +49,10 @@ COPY --chown=nonroot:nonroot . .
 # when signals are propagated, we disable it in dev image default CMD
 CMD ["uvicorn", "http_app:create_app", "--host", "0.0.0.0", "--port", "8000", "--factory", "--reload"]
 
+# Installs requirements to run production celery application
+FROM base_builder as celery_builder
+RUN poetry install --no-root
+
 # Installs requirements to run production http application
 FROM base_builder as http_builder
 RUN poetry install --no-root --with http
@@ -66,8 +70,7 @@ COPY --chown=nonroot:nonroot poetry.lock .
 COPY --chown=nonroot:nonroot alembic ./alembic
 COPY --chown=nonroot:nonroot domains ./domains
 COPY --chown=nonroot:nonroot gateways ./gateways
-COPY --chown=nonroot:nonroot config.py .
-COPY --chown=nonroot:nonroot di_container.py .
+COPY --chown=nonroot:nonroot common ./common
 COPY --chown=nonroot:nonroot alembic.ini .
 COPY --chown=nonroot:nonroot Makefile .
 
@@ -84,3 +87,10 @@ COPY --from=grpc_builder /poetryvenvs /poetryvenvs
 COPY --chown=nonroot:nonroot grpc_app ./grpc_app
 # Run CMD using array syntax, so it's uses `exec` and runs as PID1
 CMD ["opentelemetry-instrument", "python3", "-m", "grpc_app"]
+
+# Copy the celery python package and requirements from relevant builder
+FROM base_app as celery_app
+COPY --from=celery_builder /poetryvenvs /poetryvenvs
+COPY --chown=nonroot:nonroot celery_worker ./celery_worker
+# Run CMD using array syntax, so it's uses `exec` and runs as PID1
+CMD ["opentelemetry-instrument", "celery", "-A", "celery_worker:app", "worker", "-l", "INFO"]

alembic/env.py

@@ -4,19 +4,12 @@
 from sqlalchemy.ext.asyncio import AsyncEngine
 
 from alembic import context
-from config import AppConfig, init_logger
-from gateways.storage.SQLAlchemy import init_tables
+from common.bootstrap import application_init
+from common.config import AppConfig
 
 USE_TWOPHASE = False
 
 
-def init_container(app_config: AppConfig):
-    # Workaround to avoid circular import
-    from di_container import Container
-
-    return Container(config=app_config)
-
-
 # this is the Alembic Config object, which provides
 # access to the values within the .ini file in use.
 config = context.config
@@ -31,13 +24,9 @@ def init_container(app_config: AppConfig):
 # in the sample .ini file.
 # db_names = config.get_main_option("databases")
 
-# TODO: Something better organised than this
-app_config = AppConfig()
-init_logger(app_config)
+di_container = application_init(AppConfig()).di_container
 logger = logging.getLogger("alembic.env")
-di_container = init_container(app_config)
 sa_manager = di_container.SQLAlchemyBindManager()
-init_tables()
 
 target_metadata = sa_manager.get_bind_mappers_metadata()
 db_names = target_metadata.keys()

architecture.png

@@ -2,11 +2,13 @@
 allowmixing
 skinparam componentStyle uml1
 
-package http_app {
+package celery_worker {}
+
+package http_app #fff {
     package routes #DDDDDD
 }
 
-package grpc_app {
+package grpc_app #fff {
     package servicers #DDDDDD
 }
 
@@ -17,6 +19,11 @@ package domains #DDDDDD {
 
         package entities {
             class BookModel
+            class BookEvent
+        }
+
+        package tasks {
+            class BookTask
         }
 
         package data_access_interface {
@@ -37,6 +44,9 @@ routes --> BookService
 routes --> Book
 servicers --> BookService
 servicers --> Book
+celery_worker ---> tasks
+tasks <-u-> BookService
+tasks -u-> Book
 
 'links internal to books domain
 BookService -l-> Book

architecture.puml

@@ -0,0 +1,9 @@
+"""
+This is a tiny layer that takes care  of initialising the shared
+application layers (storage, logs) when running standalone workers
+without having to initialise the HTTP framework (or other ones)
+"""
+from common.bootstrap import application_init
+from common.config import AppConfig
+
+app = application_init(AppConfig()).celery_app

celery_worker/__init__.py

@@ -0,0 +1,32 @@
+from celery import Celery
+from dependency_injector.containers import DynamicContainer
+from dependency_injector.providers import Object
+from pydantic import BaseModel, ConfigDict
+
+from domains import init_celery, init_domains
+from gateways.storage import init_storage
+
+from .config import AppConfig, init_logger
+from .di_container import Container
+
+
+class InitReference(BaseModel):
+    celery_app: Celery
+    di_container: DynamicContainer
+
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+
+def application_init(app_config: AppConfig) -> InitReference:
+    container = Container(
+        config=Object(app_config),
+    )
+    init_logger(app_config)
+    init_domains(app_config)
+    init_storage()
+    celery = init_celery(app_config)
+
+    return InitReference(
+        celery_app=celery,
+        di_container=container,
+    )

tests/domains/books/local/__init__.py renamed to common/__init__.py

@@ -4,14 +4,53 @@
 
 import structlog
 from opentelemetry import trace
-from pydantic_settings import BaseSettings
+from pydantic import BaseModel
+from pydantic_settings import BaseSettings, SettingsConfigDict
 from sqlalchemy_bind_manager import SQLAlchemyAsyncConfig
 from structlog.typing import Processor
 
 TYPE_ENVIRONMENT = Literal["local", "test", "staging", "production"]
 
 
+class CeleryConfig(BaseModel):
+    # https://docs.celeryq.dev/en/stable/userguide/configuration.html#configuration
+
+    timezone: str = "UTC"
+
+    # Broker config
+    broker_url: str = "redis://redis:6379/0"
+    broker_connection_retry_on_startup: bool = True
+
+    # Results backend config
+    result_backend: str = "redis://redis:6379/1"
+    redis_socket_keepalive: bool = True
+
+    # Enable to ignore the results by default and not produce tombstones
+    task_ignore_result: bool = False
+
+    # We want to use the default python logger configured using structlog
+    worker_hijack_root_logger: bool = False
+
+    # Events enabled for monitoring
+    worker_send_task_events: bool = True
+    task_send_sent_event: bool = True
+
+    beat_schedule: dict = {
+        "recurrent_example": {
+            "task": "domains.books.tasks.book_created",
+            "schedule": 5.0,
+            "args": ("a-random-book-id",),
+        },
+    }
+
+
 class AppConfig(BaseSettings):
+    model_config = SettingsConfigDict(env_nested_delimiter="__")
+
+    APP_NAME: str = "bootstrap"
+    CELERY: CeleryConfig = CeleryConfig()
+    DEBUG: bool = False
+    ENVIRONMENT: TYPE_ENVIRONMENT = "local"
     SQLALCHEMY_CONFIG: Dict[str, SQLAlchemyAsyncConfig] = dict(
         default=SQLAlchemyAsyncConfig(
             engine_url=f"sqlite+aiosqlite:///{os.path.dirname(os.path.abspath(__file__))}/sqlite.db",
@@ -24,8 +63,6 @@ class AppConfig(BaseSettings):
             ),
         ),
     )
-    ENVIRONMENT: TYPE_ENVIRONMENT = "local"
-    DEBUG: bool = False
 
 
 def init_logger(config: AppConfig) -> None:

common/bootstrap.py

@@ -3,7 +3,7 @@
 from sqlalchemy_bind_manager import SQLAlchemyBindManager
 from sqlalchemy_bind_manager._repository import SQLAlchemyAsyncRepository
 
-from config import AppConfig
+from common.config import AppConfig
 from domains.books._data_access_interfaces import (
     BookEventGatewayInterface,
     BookRepositoryInterface,