chore(Celery): add basic Celery worker with broker only (#20)

* chore(Celery): add basic Celery worker with broker only

* chore(Celery): saving progress. Not able to schedule tasks

* fix(Celery): add celery app for use by django

* fix(Celery): register tasks

* fix(Docker): add celery workers to docker-compose

* chore(Celery): add django-celery-results backend

to store results using Django ORM

* fix(Celery): get app config the correct way

* fix(Docker): start connecting docker Celery workers to Valkey

not yet operational

* fix(Celery): get celery & django to work in docker-compose

* docs(Celery): document how to run Celery in development environment

includes changes to support the configuration and deployment
of Celery worker and its dependencies, Valkey and Postgres.

* fix(GHA): add valkey to CI services

* fix(GHA): add valkey to CI services

* fix(GHA): add valkey-cli ping to CI services

* fix(GHA): use right port for valkey

* fix(Views): remove debug task code

* test(Celery): start adding celery task tests

not yet working!

* fix(pyproject): rollback django upgrade

* fix(docker): updated docker runtime and env vars

based on feedback from #20

* fix(Dockerfile): include dependencies for psutil

psutil was introduced by pytest-celery

* fix(Backend): PRWLR-4013 fix celery settings structure

* fix(Celery): update celery app to work with new settings structure

* fix(Views): remove debug task code

* fix(Config): remove debug code

* fix(Celery): update celery app name when running worker

---------

Co-authored-by: Víctor Fernández Poyatos <victor@prowler.com>

src/backend/config/celery.py

@@ -0,0 +1,8 @@
+from celery import Celery
+
+
+celery_app = Celery("tasks")
+
+celery_app.config_from_object("django.conf:settings", namespace="CELERY")
+
+celery_app.autodiscover_tasks(["tasks"])

src/backend/config/django/base.py

@@ -1,5 +1,6 @@
 from config.custom_logging import LOGGING  # noqa
 from config.env import BASE_DIR, env  # noqa
+from config.settings.celery import *  # noqa
 
 SECRET_KEY = env("SECRET_KEY", default="secret")
 DEBUG = env.bool("DJANGO_DEBUG", default=False)
@@ -20,6 +21,7 @@ INSTALLED_APPS = [
     "drf_spectacular",
     "django_guid",
     "rest_framework_json_api",
+    "django_celery_results",
 ]
 
 MIDDLEWARE = [

src/backend/config/settings/__init__.py

@@ -0,0 +1,3 @@
+from config.celery import celery_app
+
+__all__ = ("celery_app",)

src/backend/config/settings/celery.py

@@ -0,0 +1,10 @@
+from config.env import env
+
+VALKEY_HOST = env("VALKEY_HOST", default="valkey-db")
+VALKEY_PORT = env("VALKEY_PORT", default="6479")
+VALKEY_DB = env("VALKEY_DB", default="0")
+
+CELERY_BROKER_URL = f"redis://{VALKEY_HOST}:{VALKEY_PORT}/{VALKEY_DB}"
+CELERY_RESULT_BACKEND = "django-db"
+
+CELERY_BROKER_CONNECTION_RETRY_ON_STARTUP = True

src/backend/conftest.py

@@ -23,3 +23,20 @@ def get_tenant():
 @pytest.fixture
 def tenant_header(get_tenant):
     return {"X-Tenant-ID": str(get_tenant.id)}
+
+
+# Put this in your conftest.py
+@pytest.fixture(scope="session")
+def celery_config():
+    return {
+        "broker_url": "redis://",
+        "result_backend": "redis://",
+        "broker_connection_retry_on_startup": True,
+    }
+
+
+@pytest.fixture(scope="session")
+def celery_includes():
+    return [
+        "tasks",
+    ]

src/backend/tasks/tasks.py

@@ -0,0 +1,16 @@
+from celery import shared_task
+
+from celery.utils.log import get_task_logger
+
+# TODO: add celery metadata to CustomLogger
+logger = get_task_logger(__name__)
+
+
+@shared_task
+def debug_task(x, y):
+    try:
+        logger.info("Hello world")
+    except Exception as e:
+        logger.error(e)
+
+    return x + y

src/backend/tasks/tests/test_tasks.py

@@ -0,0 +1,3 @@
+# @pytest.mark.celery
+# def test_debug_task(celery_worker):
+#     assert debug_task.delay(1, 2).get() == 4