add files

Author: giancarloromeo

packages/celery-library/requirements/Makefile

@@ -0,0 +1,6 @@
+#
+# Targets to pip-compile requirements
+#
+include ../../../requirements/base.Makefile
+
+# Add here any extra explicit dependency: e.g. _migration.txt: _base.txt

packages/celery-library/requirements/_base.in

@@ -0,0 +1,10 @@
+#
+# Specifies third-party dependencies for 'aws-library'
+#
+--constraint ../../../requirements/constraints.txt
+--requirement ../../../packages/common-library/requirements/_base.in
+--requirement ../../../packages/models-library/requirements/_base.in
+--requirement ../../../packages/service-library/requirements/_base.in
+--requirement ../../../packages/settings-library/requirements/_base.in
+
+celery

packages/celery-library/requirements/_test.in

@@ -0,0 +1,27 @@
+#
+# Specifies dependencies required to run 'models-library'
+#
+--constraint ../../../requirements/constraints.txt
+
+# Adds base AS CONSTRAINT specs, not requirement.
+#  - Resulting _text.txt is a frozen list of EXTRA packages for testing, besides _base.txt
+#
+--constraint _base.txt
+
+# testing
+coverage
+faker
+fastapi
+httpx
+pint
+pytest
+pytest-asyncio
+pytest-benchmark
+pytest-cov
+pytest-icdiff
+pytest-instafail
+pytest-mock
+pytest-runner
+pytest-sugar
+python-dotenv
+pyyaml

packages/celery-library/requirements/_tools.in

@@ -0,0 +1,5 @@
+--constraint ../../../requirements/constraints.txt
+--constraint _base.txt
+--constraint _test.txt
+
+--requirement ../../../requirements/devenv.txt

packages/celery-library/requirements/ci.txt

@@ -0,0 +1,22 @@
+# Shortcut to install all packages for the contigous integration (CI) of 'models-library'
+#
+# - As ci.txt but w/ tests
+#
+# Usage:
+#   pip install -r requirements/ci.txt
+#
+
+# installs base + tests requirements
+--requirement _base.txt
+--requirement _test.txt
+--requirement _tools.txt
+
+# installs this repo's packages
+simcore-common-library @ ../common-library
+simcore-models-library @ ../models-library/
+pytest-simcore @ ../pytest-simcore
+simcore-service-library @ ../service-library/
+simcore-settings-library @ ../settings-library/
+
+# current module
+simcore-celery-library @ .

packages/celery-library/requirements/dev.txt

@@ -0,0 +1,22 @@
+# Shortcut to install all packages needed to develop 'models-library'
+#
+# - As ci.txt but with current and repo packages in develop (edit) mode
+#
+# Usage:
+#   pip install -r requirements/dev.txt
+#
+
+# installs base + tests requirements
+--requirement _base.txt
+--requirement _test.txt
+--requirement _tools.txt
+
+# installs this repo's packages
+--editable ../common-library/
+--editable ../models-library/
+--editable ../pytest-simcore/
+--editable ../service-library/
+--editable ../settings-library/
+
+# current module
+--editable .

packages/celery-library/src/celery-library/__init__.py

@@ -0,0 +1,56 @@
+import logging
+from asyncio import AbstractEventLoop
+
+from fastapi import FastAPI
+from servicelib.redis._client import RedisClientSDK
+from settings_library.redis import RedisDatabase
+
+from ..._meta import APP_NAME
+from ...core.settings import get_application_settings
+from ._celery_types import register_celery_types
+from ._common import create_app
+from .backends._redis import RedisTaskInfoStore
+from .client import CeleryTaskClient
+
+_logger = logging.getLogger(__name__)
+
+
+def setup_celery_client(app: FastAPI) -> None:
+    async def on_startup() -> None:
+        application_settings = get_application_settings(app)
+        celery_settings = application_settings.STORAGE_CELERY
+        assert celery_settings  # nosec
+        celery_app = create_app(celery_settings)
+        redis_client_sdk = RedisClientSDK(
+            celery_settings.CELERY_REDIS_RESULT_BACKEND.build_redis_dsn(
+                RedisDatabase.CELERY_TASKS
+            ),
+            client_name=f"{APP_NAME}.celery_tasks",
+        )
+
+        app.state.celery_client = CeleryTaskClient(
+            celery_app,
+            celery_settings,
+            RedisTaskInfoStore(redis_client_sdk),
+        )
+
+        register_celery_types()
+
+    app.add_event_handler("startup", on_startup)
+
+
+def get_celery_client(app: FastAPI) -> CeleryTaskClient:
+    assert hasattr(app.state, "celery_client")  # nosec
+    celery_client = app.state.celery_client
+    assert isinstance(celery_client, CeleryTaskClient)
+    return celery_client
+
+
+def get_event_loop(app: FastAPI) -> AbstractEventLoop:
+    event_loop = app.state.event_loop
+    assert isinstance(event_loop, AbstractEventLoop)
+    return event_loop
+
+
+def set_event_loop(app: FastAPI, event_loop: AbstractEventLoop) -> None:
+    app.state.event_loop = event_loop

packages/celery-library/src/celery-library/_celery_types.py

@@ -0,0 +1,61 @@
+from functools import partial
+from pathlib import Path
+from typing import Any
+
+from kombu.utils.json import register_type  # type: ignore[import-untyped]
+from models_library.api_schemas_storage.storage_schemas import (
+    FileUploadCompletionBody,
+    FoldersBody,
+)
+from pydantic import BaseModel
+
+from ...models import FileMetaData
+
+
+def _path_encoder(obj):
+    if isinstance(obj, Path):
+        return {"__path__": True, "path": str(obj)}
+    return obj
+
+
+# Define how Path objects are deserialized
+def _path_decoder(obj):
+    if "__path__" in obj:
+        return Path(obj["path"])
+    return obj
+
+
+def _class_full_name(clz: type) -> str:
+    return ".".join([clz.__module__, clz.__qualname__])
+
+
+def _pydantic_model_encoder(obj: BaseModel, *args, **kwargs) -> dict[str, Any]:
+    return obj.model_dump(*args, **kwargs, mode="json")
+
+
+def _pydantic_model_decoder(clz: type[BaseModel], data: dict[str, Any]) -> BaseModel:
+    return clz(**data)
+
+
+def _register_pydantic_types(*models: type[BaseModel]) -> None:
+    for model in models:
+        register_type(
+            model,
+            _class_full_name(model),
+            encoder=_pydantic_model_encoder,
+            decoder=partial(_pydantic_model_decoder, model),
+        )
+
+
+def register_celery_types() -> None:
+    register_type(
+        Path,
+        _class_full_name(Path),
+        _path_encoder,
+        _path_decoder,
+    )
+    register_type(set, _class_full_name(set), encoder=list, decoder=set)
+
+    _register_pydantic_types(FileUploadCompletionBody)
+    _register_pydantic_types(FileMetaData)
+    _register_pydantic_types(FoldersBody)

packages/celery-library/src/celery-library/_common.py

@@ -0,0 +1,37 @@
+import logging
+import ssl
+from typing import Any
+
+from celery import Celery  # type: ignore[import-untyped]
+from settings_library.celery import CelerySettings
+from settings_library.redis import RedisDatabase
+
+_logger = logging.getLogger(__name__)
+
+
+def _celery_configure(celery_settings: CelerySettings) -> dict[str, Any]:
+    base_config = {
+        "broker_connection_retry_on_startup": True,
+        "result_expires": celery_settings.CELERY_RESULT_EXPIRES,
+        "result_extended": True,
+        "result_serializer": "json",
+        "task_default_queue": "default",
+        "task_send_sent_event": True,
+        "task_track_started": True,
+        "worker_send_task_events": True,
+    }
+    if celery_settings.CELERY_REDIS_RESULT_BACKEND.REDIS_SECURE:
+        base_config["redis_backend_use_ssl"] = {"ssl_cert_reqs": ssl.CERT_NONE}
+    return base_config
+
+
+def create_app(celery_settings: CelerySettings) -> Celery:
+    assert celery_settings
+
+    return Celery(
+        broker=celery_settings.CELERY_RABBIT_BROKER.dsn,
+        backend=celery_settings.CELERY_REDIS_RESULT_BACKEND.build_redis_dsn(
+            RedisDatabase.CELERY_TASKS,
+        ),
+        **_celery_configure(celery_settings),
+    )