Merge pull request #85 from ScilifelabDataCentre/worker_config

SQ-879: Create a worker config

Author: RMCrean

docker/divbase_compose.tests.yaml

@@ -9,6 +9,7 @@ services:
       - internal
   worker-quick:
     environment:
+      - DIVBASE_ENV=test
       - S3_ENDPOINT_URL=http://host.docker.internal:9002
     ports: !override
       - "8111:8101" # metrics server port
@@ -19,6 +20,7 @@ services:
       - internal
   worker-long:
     environment:
+      - DIVBASE_ENV=test
       - S3_ENDPOINT_URL=http://host.docker.internal:9002
     ports: !override
       - "8112:8101" # metrics server port (fixed: container port is 8001, not 8002)

docker/divbase_compose.yaml

@@ -51,15 +51,16 @@ services:
       db-migrator:
         condition: service_completed_successfully
     environment: &worker-env
+      - DIVBASE_ENV=local_dev
       - CELERY_BROKER_URL=pyamqp://celery_user:badpassword@rabbitmq:5672//
       - CELERY_RESULT_BACKEND=db+postgresql://divbase_user:badpassword@postgres:5432/divbase_db
       - SYNC_DATABASE_URL=postgresql+psycopg2://divbase_user:badpassword@postgres:5432/divbase_db
       - S3_ENDPOINT_URL=http://host.docker.internal:9000
       - S3_PRESIGNING_URL=http://localhost:9000
       - S3_SERVICE_ACCOUNT_ACCESS_KEY=s3-service-account
       - S3_SERVICE_ACCOUNT_SECRET_KEY=badpassword
-      - ENABLE_WORKER_METRICS=False # Enable metrics collection and metrics server in the worker. Optional, as it comes with some overhead.
-      - ENABLE_WORKER_METRICS_PER_TASK=False # Enable per-task metrics collection in the worker. Optional, as it comes with some overhead.
+      - ENABLE_WORKER_METRICS=0 # Disable metrics collection and metrics server in the worker as it comes with some overhead.
+      - ENABLE_WORKER_METRICS_PER_TASK=0 # Disable per-task metrics collection in the worker as it comes with some overhead.
     develop: &watch-celery-src
       watch:
         - action: sync+restart

docs/development/celery_task_implementation.md

@@ -47,7 +47,7 @@ For a user-submitted task to be fully integrated in DivBase, it needs to be impl
 - Define the task arguments and their type hints just like any other Python function. There are no mandatory args for DivBase tasks, but there are some patterns that can be reused. For instance, if the task logic will interact with S3, add `bucket_name: str,` Tasks that interact with VCF dimensions table need `project_id: int,` for the `get_vcf_metadata_by_project()` helper function and `project_name: str,` for exceptions.
 - The logic inside the task can be anything, but typically include reading file(s) from S3 and acting on them.
   - Divbase tasks can include database transaction (see next bullet below for more details), but if the task only does a database CRUD it would be better to implement that as an async database CRUD in the API layer rather than a Celery task since that is likely faster and does not require a Celery worker.
-  - There are existing helper functions for S3 operations (e.g. `create_s3_file_manager()`, `_download_vcf_files()`, `_delete_job_files_from_worker()`) and VCF dimensions metadata table transactions (e.g. `get_vcf_metadata_by_project()`, `get_skipped_vcfs_by_project_worker()`,`create_or_update_vcf_metadata()`). Look at previous tasks to see if there are existing helper functions or logic that can be reused.
+  - There are existing helper functions for S3 operations (e.g. `_create_s3_file_manager()`, `_download_vcf_files()`, `_delete_job_files_from_worker()`) and VCF dimensions metadata table transactions (e.g. `get_vcf_metadata_by_project()`, `get_skipped_vcfs_by_project_worker()`,`create_or_update_vcf_metadata()`). Look at previous tasks to see if there are existing helper functions or logic that can be reused.
 - Database transactions can be called from the task using the `SyncSessionLocal()` sessionmaker instance from SQLAlchemy. There is no point in trying to use async database sessions in DivBase Celery tasks since tasks are run in dedicated worker containers/pods and inside a Celery event pool (see the Celery docs for `prefork` pools). Furthermore, since Celery does not provide a simple way to use the async event loops from SQLAlchemy. Experiments during DivBase development showed that it is technically possible to implement workarounds to use async db sessions in tasks executed by Celery workers, but that it can easily result in event pool issues and errors.
 - Raise errors in the task function (catching any errors proprageted from potential helper fuctions called by the task). This assures that `CeleryTaskMeta` is correctly updated with task status `FAILURE` and the error message. This assures that the task history CLI command correctly prints this information in the user's terminal. If errors are returned in any other way that with a `raise`, there is a risk that the status of the task becomes `SUCESS` even if it exited due to errors.
 - Tasks end with a `return` statement. This is written to the `results` field in `CeleryTaskMeta`. Celery will use `pickle` to serialize the results for storage in the `CeleryTaskMeta` table.
@@ -377,7 +377,7 @@ Example of a DivBase cron task definition:
 
 ```python
 @app.task(name="cron_tasks.cleanup_old_task_history")
-def cleanup_old_task_history_task(retention_days: int = TASK_RETENTION_DAYS):
+def cleanup_old_task_history_task(retention_days: int = worker_settings.cron.task_retention_days):
     """
     Periodic task to clean up old task history entries from both TaskHistoryDB and CeleryTaskMeta.
     Runs daily to remove entries older than retention_days.
@@ -436,7 +436,7 @@ app.conf.beat_schedule = {
         "schedule": crontab(
             hour=5, minute=0
         ),  # Run daily at 5 AM CET (timezone defined in app in tasks.py). Don't set to 2 AM or 3 AM due to daylight saving
-        "kwargs": {"retention_days": TASK_RETENTION_DAYS},
+        "kwargs": {"retention_days": worker_settings.cron.task_retention_days},
     },
     "cleanup-stuck-tasks-daily": {
         "task": "cron_tasks.cleanup_stuck_tasks",

packages/divbase-api/src/divbase_api/api_config.py

@@ -19,8 +19,8 @@
 
 
 @dataclass
-class APISettings:
-    """API configuration settings."""
+class GeneralSettings:
+    """General API configuration settings."""
 
     environment: str = os.getenv("DIVBASE_ENV", "NOT_SET")
     frontend_base_url: str = os.getenv("FRONTEND_BASE_URL", "NOT_SET")
@@ -127,10 +127,10 @@ def __post_init__(self):
 
 
 @dataclass
-class Settings:
+class APISettings:
     """Configuration settings for DivBase API."""
 
-    api: APISettings = field(default_factory=APISettings)
+    general: GeneralSettings = field(default_factory=GeneralSettings)
     database: DBSettings = field(default_factory=DBSettings)
     s3: S3Settings = field(default_factory=S3Settings)
     jwt: JWTSettings = field(default_factory=JWTSettings)
@@ -139,14 +139,14 @@ class Settings:
     def validate_api_settings(self) -> None:
         """
         Validate all required settings are actually set.
-        This is run on startup of the API which means that later on in the codebase
+        This is run on startup of the API (lifespan event) which means that later on in the codebase
         we don't have to check for any non set values, we can just assume they are set.
         """
         required_fields = {
-            "DIVBASE_ENV": self.api.environment,
-            "FRONTEND_BASE_URL": self.api.frontend_base_url,
-            "MKDOCS_SITE_URL": self.api.mkdocs_site_url,
-            "USER_SUPPORT_EMAIL": self.api.user_support_email,
+            "DIVBASE_ENV": self.general.environment,
+            "FRONTEND_BASE_URL": self.general.frontend_base_url,
+            "MKDOCS_SITE_URL": self.general.mkdocs_site_url,
+            "USER_SUPPORT_EMAIL": self.general.user_support_email,
             "ASYNC_DATABASE_URL": self.database.url,
             "SYNC_DATABASE_URL": self.database.sync_url,
             "JWT_SECRET_KEY": self.jwt.secret_key,
@@ -161,11 +161,14 @@ def validate_api_settings(self) -> None:
             if isinstance(setting, SecretStr):
                 if setting.get_secret_value() == "NOT_SET":
                     raise ValueError(f"A required environment variable was not set: {setting_name=}")
-                if self.api.environment not in LOCAL_DEV_ENVIRONMENTS and setting.get_secret_value() == "badpassword":
+                if (
+                    self.general.environment not in LOCAL_DEV_ENVIRONMENTS
+                    and setting.get_secret_value() == "badpassword"
+                ):
                     raise ValueError(
                         f"A secret environment variable was set to badpassword for a non local environment: {setting_name=}"
                     )
 
 
 # This instance can be imported and used throughout the application.
-settings = Settings()
+api_settings = APISettings()

packages/divbase-api/src/divbase_api/crud/announcements.py

@@ -13,7 +13,7 @@
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from divbase_api.api_config import settings
+from divbase_api.api_config import api_settings
 from divbase_api.models.announcements import AnnouncementDB, AnnouncementTarget
 from divbase_api.services.validate_cli_versions import cli_update_available
 from divbase_lib.api_schemas.announcements import AnnouncementResponse
@@ -48,9 +48,9 @@ def new_cli_version_announcement(user_cli_version: str) -> AnnouncementResponse
         heading="A new version of DivBase CLI is available",
         message=(
             f"You are using an outdated version of the DivBase CLI '{user_cli_version}'. "
-            f"Please consider upgrading to the latest version '{settings.api.latest_cli_version}' for new features, bug fixes, and improved security.\n"
+            f"Please consider upgrading to the latest version '{api_settings.general.latest_cli_version}' for new features, bug fixes, and improved security.\n"
             "If you're not sure how to do that, you can find instructions on how to upgrade here: "
-            f"{settings.api.mkdocs_site_url}/user-guides/installation"
+            f"{api_settings.general.mkdocs_site_url}/user-guides/installation"
         ),
         level="info",
     )

packages/divbase-api/src/divbase_api/crud/s3.py

@@ -2,11 +2,26 @@
 Crud operations for the S3 endpoints
 """
 
-from divbase_api.api_config import settings
+from divbase_api.api_config import api_settings
+from divbase_api.services.pre_signed_urls import S3PreSignedService
 from divbase_api.services.s3_client import S3FileManager
 from divbase_lib.api_schemas.s3 import FileChecksumResponse
 
 
+def get_s3_file_manager() -> S3FileManager:
+    """Dependency to get an S3FileManager instance. Dependency injected into FastAPI endpoints."""
+    return S3FileManager(
+        url=api_settings.s3.endpoint_url,
+        access_key=api_settings.s3.access_key.get_secret_value(),
+        secret_key=api_settings.s3.secret_key.get_secret_value(),
+    )
+
+
+def get_pre_signed_service() -> S3PreSignedService:
+    """Dependency to get pre-signed S3 service. Dependency injected into FastAPI endpoints."""
+    return S3PreSignedService()
+
+
 def get_s3_checksums(bucket_name: str, files_to_check: list[str]) -> list[FileChecksumResponse]:
     """
     Given a list of object names, return a list of their names and checksums for those that exist in S3.
@@ -16,11 +31,7 @@ def get_s3_checksums(bucket_name: str, files_to_check: list[str]) -> list[FileCh
     If we consider a list_objects_v2 approach we have to think carefully about pagination there,
     other we make a lot of calls to S3 anyway.
     """
-    s3_file_manager = S3FileManager(
-        url=settings.s3.endpoint_url,
-        access_key=settings.s3.access_key.get_secret_value(),
-        secret_key=settings.s3.secret_key.get_secret_value(),
-    )
+    s3_file_manager = get_s3_file_manager()
 
     response = []
     for file_name in files_to_check:

packages/divbase-api/src/divbase_api/db.py

@@ -12,7 +12,7 @@
 from sqlalchemy import text
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
 
-from divbase_api.api_config import settings
+from divbase_api.api_config import api_settings
 from divbase_api.crud.users import create_user, get_all_users
 from divbase_api.schemas.users import UserCreate
 
@@ -22,8 +22,8 @@
 # NOTE: The creation of the AsyncSessionLocal should only occur once, when the module is loaded.
 # AKA: Do not refactor to have these in functions.
 engine = create_async_engine(
-    url=settings.database.url.get_secret_value(),
-    echo=settings.database.echo_db_output,
+    url=api_settings.database.url.get_secret_value(),
+    echo=api_settings.database.echo_db_output,
     pool_pre_ping=True,
 )
 AsyncSessionLocal = async_sessionmaker(
@@ -90,8 +90,8 @@ async def create_first_admin_user() -> None:
     1. no admin users already exists in the db.
     2. FIRST_ADMIN_EMAIL and FIRST_ADMIN_PASSWORD env vars are set.
     """
-    admin_email = settings.api.first_admin_email
-    admin_password = settings.api.first_admin_password
+    admin_email = api_settings.general.first_admin_email
+    admin_password = api_settings.general.first_admin_password
 
     if admin_email == "NOT_SET" or admin_password.get_secret_value() == "NOT_SET":
         logger.info(

packages/divbase-api/src/divbase_api/deps.py

@@ -10,7 +10,7 @@
 - get_current_user_from_cookie for frontend based routes.
 
 The dependencies that depend on this can use e.g. get_current_user as a sub-dependency,
-so all checks in the sub-dependancy function are ran when you use this dependency.
+so all checks in the sub-dependency function are ran when you use this dependency.
 (see here: https://fastapi.tiangolo.com/yo/advanced/security/oauth2-scopes/#dependency-tree-and-scopes)
 """
 
@@ -157,7 +157,7 @@ async def get_current_admin_user(current_user: Annotated[UserDB, Depends(get_cur
     Verify current user has admin access.
 
     Note: This function works by using a sub-dependency to get the current user,
-    so all the checks in the sub-dependancy function are ran when you use this dependency.
+    so all the checks in the sub-dependency function are ran when you use this dependency.
     (see here: https://fastapi.tiangolo.com/yo/advanced/security/oauth2-scopes/#dependency-tree-and-scopes)
     """
     if not current_user.is_admin:

packages/divbase-api/src/divbase_api/divbase_api.py

@@ -14,7 +14,7 @@
 
 from divbase_api import __version__ as divbase_version
 from divbase_api.admin_panel import register_admin_panel
-from divbase_api.api_config import LOCAL_DEV_ENVIRONMENTS, settings
+from divbase_api.api_config import LOCAL_DEV_ENVIRONMENTS, api_settings
 from divbase_api.db import (
     check_db_migrations_up_to_date,
     create_first_admin_user,
@@ -36,7 +36,7 @@
 from divbase_api.routes.task_history import task_history_router
 from divbase_api.routes.vcf_dimensions import vcf_dimensions_router
 
-logging.basicConfig(level=settings.api.log_level, handlers=[logging.StreamHandler(sys.stderr)])
+logging.basicConfig(level=api_settings.general.log_level, handlers=[logging.StreamHandler(sys.stderr)])
 
 logger = logging.getLogger(__name__)
 
@@ -47,7 +47,7 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     # startup
     logger.info("Starting up DivBase API...")
 
-    settings.validate_api_settings()
+    api_settings.validate_api_settings()
     logger.info("All API settings are correctly set.")
 
     if not await health_check_db():
@@ -94,7 +94,7 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
 app.include_router(s3_router, prefix="/api/v1/s3", tags=["s3"])
 app.include_router(task_history_router, prefix="/api/v1/task-history", tags=["task-history"])
 app.include_router(vcf_dimensions_router, prefix="/api/v1/vcf-dimensions", tags=["vcf-dimensions"])
-if settings.api.environment in LOCAL_DEV_ENVIRONMENTS:
+if api_settings.general.environment in LOCAL_DEV_ENVIRONMENTS:
     # not needed in deployed enviroments, so no need to expose it.
     app.include_router(admin_router, prefix="/api/v1/admin", tags=["admin"])
 

packages/divbase-api/src/divbase_api/frontend_routes/auth.py

@@ -9,7 +9,7 @@
 from pydantic import SecretStr, ValidationError
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from divbase_api.api_config import settings
+from divbase_api.api_config import api_settings
 from divbase_api.api_constants import KNOWN_JOB_ROLES, SWEDISH_UNIVERSITIES
 from divbase_api.crud.auth import (
     authenticate_user,
@@ -215,7 +215,7 @@ def registration_failed_response(error_message: str):
     return templates.TemplateResponse(
         request=request,
         name="auth_pages/register_success.html",
-        context={"name": user_data.name, "email": user_data.email, "from_email": settings.email.from_email},
+        context={"name": user_data.name, "email": user_data.email, "from_email": api_settings.email.from_email},
     )
 
 
@@ -374,7 +374,7 @@ async def post_forgot_password_form(
     """Handle forgot password form submission to send a password reset email."""
     RESET_LINK_SENT_MSG = (
         f"If your account exists, and your email is verified, a password reset email has been sent to {email}. Please check your inbox."
-        + f"The email will be sent from {settings.email.from_email}."
+        + f"The email will be sent from {api_settings.email.from_email}."
     )
 
     user = await get_user_by_email(db=db, email=email)