Revert "🔥 Remove Celery and Flower, they are currently not used nor recommended (fastapi#694)"

This reverts commit 1e256bc.

Author: frma-tolife

.env

@@ -33,6 +33,10 @@ POSTGRES_PASSWORD=-7vcO6peCJz0y0AcYUl8eM9m-LhC4NPsG9WLNe75hrs
 
 SENTRY_DSN=
 
+# Flower
+FLOWER_BASIC_AUTH=admin:changethis
+
 # Configure these with your own Docker registry images
 DOCKER_IMAGE_BACKEND=backend
+DOCKER_IMAGE_CELERYWORKER=celery
 DOCKER_IMAGE_FRONTEND=frontend

.github/workflows/deploy-production.yml

@@ -23,6 +23,7 @@ jobs:
       EMAILS_FROM_EMAIL: ${{ secrets.EMAILS_FROM_EMAIL }}
       POSTGRES_PASSWORD: ${{ secrets.POSTGRES_PASSWORD }}
       SENTRY_DSN: ${{ secrets.SENTRY_DSN }}
+      FLOWER_BASIC_AUTH: ${{ secrets.FLOWER_BASIC_AUTH }}
     steps:
       - name: Checkout
         uses: actions/checkout@v4

.github/workflows/deploy-staging.yml

@@ -23,6 +23,7 @@ jobs:
       EMAILS_FROM_EMAIL: ${{ secrets.EMAILS_FROM_EMAIL }}
       POSTGRES_PASSWORD: ${{ secrets.POSTGRES_PASSWORD }}
       SENTRY_DSN: ${{ secrets.SENTRY_DSN }}
+      FLOWER_BASIC_AUTH: ${{ secrets.FLOWER_BASIC_AUTH }}
     steps:
       - name: Checkout
         uses: actions/checkout@v4

README.md

@@ -210,6 +210,7 @@ The input variables, with their default values (some auto generated) are:
 - `emails_from_email`: (default: `"[email protected]"`) The email account to send emails from, you can set it later in .env.
 - `postgres_password`: (default: `"changethis"`) The password for the PostgreSQL database, stored in .env, you can generate one with the method above.
 - `sentry_dsn`: (default: "") The DSN for Sentry, if you are using it, you can set it later in .env.
+- `flower_basic_auth`: (default: `"admin:changethis"`) The basic auth for Flower, you can set it later in .env.
 
 ## Backend Development
 

backend/README.md

@@ -23,6 +23,8 @@ Automatic interactive documentation with Swagger UI (from the OpenAPI backend):
 
 Adminer, database web administration: http://localhost:8080
 
+Flower, administration of Celery tasks: http://localhost:5555
+
 Traefik UI, to see how the routes are being handled by the proxy: http://localhost:8090
 
 **Note**: The first time you start your stack, it might take a minute for it to be ready. While the backend waits for the database to be ready and configures everything. You can check the logs to monitor it.
@@ -63,6 +65,8 @@ Make sure your editor is using the correct Python virtual environment.
 
 Modify or add SQLModel models for data and SQL tables in `./backend/app/models.py`, API endpoints in `./backend/app/api/`, CRUD (Create, Read, Update, Delete) utils in `./backend/app/crud.py`.
 
+Add and modify tasks to the Celery worker in `./backend/app/worker.py`.
+
 ### Enabling Open User Registration
 
 By default the backend has user registration disabled, but there's already a route to register users. If you want to allow users to register themselves, you can set the environment variable `USERS_OPEN_REGISTRATION` to `True` in the `.env` file.
@@ -163,7 +167,19 @@ docker compose exec backend bash /app/tests-start.sh -x
 
 #### Test Coverage
 
-When the tests are run, a file `htmlcov/index.html` is generated, you can open it in your browser to see the coverage of the tests.
+Because the test scripts forward arguments to `pytest`, you can enable test coverage HTML report generation by passing `--cov-report=html`.
+
+To run the local tests with coverage HTML reports:
+
+```Bash
+DOMAIN=backend sh ./scripts/test-local.sh --cov-report=html
+```
+
+To run the tests in a running stack with coverage HTML reports:
+
+```bash
+docker compose exec backend bash /app/tests-start.sh --cov-report=html
+```
 
 ### Migrations
 

backend/app/api/routes/utils.py

@@ -2,12 +2,26 @@
 from pydantic.networks import EmailStr
 
 from app.api.deps import get_current_active_superuser
+from app.core.celery_app import celery_app
 from app.models import Message
 from app.utils import generate_test_email, send_email
 
 router = APIRouter()
 
 
+@router.post(
+    "/test-celery/",
+    dependencies=[Depends(get_current_active_superuser)],
+    status_code=201,
+)
+def test_celery(body: Message) -> Message:
+    """
+    Test Celery worker.
+    """
+    celery_app.send_task("app.worker.test_celery", args=[body.message])
+    return Message(message="Word received")
+
+
 @router.post(
     "/test-email/",
     dependencies=[Depends(get_current_active_superuser)],

backend/app/core/celery_app.py

@@ -0,0 +1,5 @@
+from celery import Celery
+
+celery_app = Celery("worker", broker="amqp://guest@queue//")
+
+celery_app.conf.task_routes = {"app.worker.test_celery": "main-queue"}

backend/app/tests/api/routes/test_celery.py

@@ -0,0 +1,16 @@
+from fastapi.testclient import TestClient
+
+from app.core.config import settings
+
+
+def test_celery_worker_test(
+    client: TestClient, superuser_token_headers: dict[str, str]
+) -> None:
+    data = {"message": "test"}
+    r = client.post(
+        f"{settings.API_V1_STR}/utils/test-celery/",
+        json=data,
+        headers=superuser_token_headers,
+    )
+    response = r.json()
+    assert response["message"] == "Word received"

backend/app/tests/scripts/test_celery_pre_start.py

@@ -0,0 +1,33 @@
+from unittest.mock import MagicMock
+
+from pytest_mock import MockerFixture
+from sqlmodel import select
+
+from app.celeryworker_pre_start import init, logger
+
+
+def test_init_successful_connection(mocker: MockerFixture) -> None:
+    engine_mock = MagicMock()
+
+    session_mock = MagicMock()
+    exec_mock = MagicMock(return_value=True)
+    session_mock.configure_mock(**{"exec.return_value": exec_mock})
+    mocker.patch("sqlmodel.Session", return_value=session_mock)
+
+    mocker.patch.object(logger, "info")
+    mocker.patch.object(logger, "error")
+    mocker.patch.object(logger, "warn")
+
+    try:
+        init(engine_mock)
+        connection_successful = True
+    except Exception:
+        connection_successful = False
+
+    assert (
+        connection_successful
+    ), "The database connection should be successful and not raise an exception."
+
+    assert session_mock.exec.called_once_with(
+        select(1)
+    ), "The session should execute a select statement once."

backend/app/worker.py

@@ -0,0 +1,12 @@
+import sentry_sdk
+
+from app.core.celery_app import celery_app
+from app.core.config import settings
+
+if settings.SENTRY_DSN:
+    sentry_sdk.init(dsn=str(settings.SENTRY_DSN))
+
+
+@celery_app.task(acks_late=True)
+def test_celery(word: str) -> str:
+    return f"test task return {word}"