Replace Celery with Dramatiq

Dockerfile

@@ -51,8 +51,8 @@ RUN make dev-dependencies
 # 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
+# Installs requirements to run production dramatiq application
+FROM base_builder AS dramatiq_builder
 RUN poetry install --no-root
 
 # Installs requirements to run production http application
@@ -68,7 +68,7 @@ COPY --chown=nonroot:nonroot poetry.lock .
 COPY --chown=nonroot:nonroot src/alembic ./alembic
 COPY --chown=nonroot:nonroot src/domains ./domains
 COPY --chown=nonroot:nonroot src/gateways ./gateways
-COPY --chown=nonroot:nonroot src/common ./bootstrap
+COPY --chown=nonroot:nonroot src/common ./common
 COPY --chown=nonroot:nonroot src/alembic.ini .
 COPY --chown=nonroot:nonroot Makefile .
 
@@ -79,9 +79,10 @@ COPY --chown=nonroot:nonroot src/http_app ./http_app
 # Run CMD using array syntax, so it's uses `exec` and runs as PID1
 CMD ["opentelemetry-instrument", "uvicorn", "http_app:create_app", "--host", "0.0.0.0", "--port", "8000", "--factory"]
 
-# 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 src/celery_worker ./celery_worker
+# Copy the dramatiq python package and requirements from relevant builder
+FROM base_app AS dramatiq_app
+COPY --from=dramatiq_builder /poetryvenvs /poetryvenvs
+COPY --chown=nonroot:nonroot src/dramatiq_worker ./dramatiq_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"]
+# TODO: Review processes/threads
+CMD ["opentelemetry-instrument", "dramatiq", "-p", "1", "-t", "1", "dramatiq_worker"]

README.md

@@ -21,7 +21,7 @@ and [SOLID principles](https://en.wikipedia.org/wiki/SOLID).
 This template provides out of the box some commonly used functionalities:
 
 * API Documentation using [FastAPI](https://fastapi.tiangolo.com/)
-* Async tasks execution using [Celery](https://docs.celeryq.dev/en/stable/index.html)
+* Async tasks execution using [Dramatiq](https://dramatiq.io/index.html)
 * Repository pattern for databases using [SQLAlchemy](https://www.sqlalchemy.org/) and [SQLAlchemy bind manager](https://febus982.github.io/sqlalchemy-bind-manager/stable/)
 * Database migrations using [Alembic](https://alembic.sqlalchemy.org/en/latest/) (configured supporting both sync and async SQLAlchemy engines)
 * [TODO] Producer and consumer to emit and consume events using [CloudEvents](https://cloudevents.io/) format on [Confluent Kafka](https://docs.confluent.io/kafka-clients/python/current/overview.html)
@@ -44,7 +44,7 @@ Using Docker:
 * `make containers`: Build containers
 * `docker compose run --rm dev make migrate`: Run database migrations
 * `docker compose up dev`: Run HTTP application with hot reload
-* `docker compose up celery-worker`: Run the celery worker
+* `docker compose up dramatiq-worker`: Run the dramatiq worker
 * `docker compose run --rm test`: Run test suite
 
 Locally:

docker-compose.yaml

@@ -8,34 +8,16 @@ services:
   redis:
     image: redis
 
-  celery-worker:
+  dramatiq-worker:
     build:
       dockerfile: Dockerfile
       context: .
-      target: celery_app
+      target: dramatiq_app
     environment:
-      OTEL_SERVICE_NAME: "bootstrap-fastapi-worker"
-      OTEL_EXPORTER_OTLP_ENDPOINT: "http://otel-collector:4317"
-      CELERY__broker_url: "redis://redis:6379/0"
-      CELERY__result_backend: "redis://redis:6379/1"
-    working_dir: "/app/src"
-    volumes:
-      - '.:/app'
-    depends_on:
-      - redis
-      - otel-collector
-      - celery-beat
-
-  celery-beat:
-    build:
-      dockerfile: Dockerfile
-      context: .
-      target: celery_app
-    environment:
-      OTEL_SERVICE_NAME: "bootstrap-fastapi-worker"
+      WATCHFILES_FORCE_POLLING: true
+      OTEL_SERVICE_NAME: "bootstrap-fastapi-dramatiq-worker"
       OTEL_EXPORTER_OTLP_ENDPOINT: "http://otel-collector:4317"
-      CELERY__broker_url: "redis://redis:6379/0"
-      CELERY__result_backend: "redis://redis:6379/1"
+      DRAMATIQ__REDIS_URL: "redis://redis:6379/0"
     working_dir: "/app/src"
     volumes:
       - '.:/app'
@@ -44,12 +26,14 @@ services:
       - otel-collector
     command:
       - opentelemetry-instrument
-      - celery
-      - -A
-      - celery_worker:app
-      - beat
-      - -l
-      - INFO
+      - dramatiq
+      - --watch
+      - .
+      - -p
+      - "1"
+      - -t
+      - "1"
+      - dramatiq_worker
 
   dev:
     build:
@@ -60,8 +44,6 @@ services:
       WATCHFILES_FORCE_POLLING: true
       OTEL_SERVICE_NAME: "bootstrap-fastapi-dev"
       OTEL_EXPORTER_OTLP_ENDPOINT: "http://otel-collector:4317"
-      CELERY__broker_url: "redis://redis:6379/0"
-      CELERY__result_backend: "redis://redis:6379/1"
     ports:
       - '8000:8000'
     working_dir: "/app/src"
@@ -93,8 +75,6 @@ services:
     environment:
       OTEL_SERVICE_NAME: "bootstrap-fastapi-http"
       OTEL_EXPORTER_OTLP_ENDPOINT: "http://otel-collector:4317"
-      CELERY__broker_url: "redis://redis:6379/0"
-      CELERY__result_backend: "redis://redis:6379/1"
     ports:
       - '8001:8000'
     volumes:

docs/architecture.md

@@ -11,7 +11,7 @@ and the persistence layer.
 This is a high level list of the packages in this application template:
 
 * `alembic` (database migration manager)
-* `celery_worker` (async tasks runner)
+* `dramatiq_worker` (async tasks runner)
 * `common` (some common boilerplate initialisation shared by all applications )
 * `http_app` (http presentation layer)
 * `gateways` (database connection manager, repository implementation, event emitter, etc.)
@@ -31,7 +31,7 @@ This is a high level representation of the nested layers in the application:
 flowchart TD
     subgraph "Framework & Drivers + Interface Adapters" 
         alembic
-        celery_worker
+        dramatiq_worker
         http_app
         gateways
         subgraph domains.books["Use Cases"]
@@ -52,7 +52,7 @@ flowchart TD
     end
 
     alembic ~~~ domains.books
-    celery_worker ~~~ domains.books
+    dramatiq_worker ~~~ domains.books
     http_app ~~~ domains.books
     gateways ~~~ domains.books
 
@@ -69,7 +69,7 @@ A more detailed view showing the class dependencies and the absence of cyclical
 
 ```mermaid
 flowchart TD
-    celery_worker
+    dramatiq_worker
     http_app
     subgraph gateways
       SQLAlchemyRepository
@@ -99,8 +99,8 @@ flowchart TD
       end
     end
 
-    celery_worker-->domain_boundary
-    celery_worker-->dto
+    dramatiq_worker-->domain_boundary
+    dramatiq_worker-->dto
     http_app-->domain_boundary
     http_app-->dto
     domain_boundary-->domain_logic

docs/dockerfile.md

@@ -61,22 +61,22 @@ flowchart TD
       http_builder-->http_app
     end
 
-    subgraph Celery
-      celery_builder["Celery builder
+    subgraph Dramatiq
+      dramatiq_builder["Dramatiq builder
           ============
           Installs requirements for
-          Celery worker in a virtualenv"]
-      celery_app["HTTP app
+          Dramatiq worker in a virtualenv"]
+      dramatiq_app["HTTP app
         ========
-        Copies Celery worker app,
+        Copies Dramatiq worker app,
         shared logic
         and requirements
         from previous containers"]
-      celery_builder-->celery_app
+      dramatiq_builder-->dramatiq_app
     end
 
     base_builder-->http_builder
-    base_builder-->celery_builder
+    base_builder-->dramatiq_builder
     base_app-->http_app
-    base_app-->celery_app
+    base_app-->dramatiq_app
 ```

docs/packages/bootstrap.md

@@ -1,12 +1,12 @@
 # Bootstrap
 
 The `bootstrap` package contains logic that is shared among the external layer
-(i.e. `http_app`, `celery_worker`, etc.).
+(i.e. `http_app`, `dramatiq_worker`, etc.).
 
 It contains the following submodules and packages (and related responsibilities):
 
 * `bootstrap.bootstrap`: The application initialisation logic (database, logging,
-  celery tasks) necessary to run the domain logic. It uses `bootstrap.config` and
+  dramatiq tasks) necessary to run the domain logic. It uses `bootstrap.config` and
   `bootstrap.di_container` subpackages. It does not contain the specific HTTP
   framework initialisation (or other frameworks such as GRPC).
 * `bootstrap.config`: The application config models, based on `BaseSettings`

docs/packages/domains.md

@@ -43,6 +43,6 @@ and can be used to bootstrap the application:
   to bootstrap the SQLAlchemy imperative mapping in `bootstrap.storage.SQLAlchemy`
   package.
 * The `domains.book._service` contains the `BookService` implementation.
-* The `domains.book._tasks` contains the implementation of celery tasks
+* The `domains.book._tasks` contains the implementation of dramatiq tasks
   for operations that can be queued without waiting for a result (e.g.
   send an email, invalidate cache).

docs/packages/dramatiq_worker.md

@@ -0,0 +1,3 @@
+# Dramatiq worker
+
+The `dramatiq_worker` package is a small entrypoint to run Dramatiq workers.