progress

Author: chaen

diracx-core/src/diracx/core/config/sources.py

@@ -8,6 +8,7 @@
 import asyncio
 import logging
 import os
+from abc import ABCMeta, abstractmethod
 from datetime import datetime, timezone
 from pathlib import Path
 from tempfile import TemporaryDirectory
@@ -76,7 +77,12 @@ def __init_subclass__(cls) -> None:
 
     @classmethod
     def create(cls):
-        return cls.create_from_url(backend_url=os.environ["DIRACX_CONFIG_BACKEND_URL"])
+        # Avoid circular import
+        from diracx.core.settings import FactorySettings
+
+        return cls.create_from_url(
+            backend_url=FactorySettings().diracx_config_backend_url
+        )
 
     @classmethod
     def create_from_url(

diracx-core/src/diracx/core/settings.py

@@ -14,10 +14,12 @@
 
 import contextlib
 import json
+import os
 from collections.abc import AsyncIterator
 from pathlib import Path
 from typing import Annotated, Any, Self, TypeVar, cast
 
+import dotenv
 from cryptography.fernet import Fernet
 from joserfc.jwk import KeySet, KeySetSerialization
 from pydantic import (
@@ -29,15 +31,18 @@
     SecretStr,
     TypeAdapter,
     UrlConstraints,
-    create_model,
+    field_validator,
     model_validator,
 )
 from pydantic_settings import BaseSettings, SettingsConfigDict
 from signurlarity.aio.client import AsyncClient
 from signurlarity.exceptions import SignurlarityError
 
+from .config.sources import ConfigSourceUrl
+from .extensions import DiracEntryPoint, select_from_extension
 from .properties import SecurityProperty
 from .s3 import s3_bucket_exists
+from .utils import dotenv_files_from_environment
 
 T = TypeVar("T")
 
@@ -353,81 +358,100 @@ def s3_client(self) -> AsyncClient:
         return self._client
 
 
-# def make_factory_settings(model_cls: type[StaticFactorySettings]):
-#     new_fields = {}
-#     for var, val in os.environ.items():
-#         if var.startswith("XDG_"):
-#             new_fields[var] = (str, val)
-#     return create_model("FactorySettings", __base__=model_cls, **new_fields)
-
-
-# FactorySettings = make_factory_settings(StaticFactorySettings)
-
-
-from pydantic import Field
-
-# Could come from code, YAML, DB, plugin system, etc.
-FIELD_SPECS = {
-    "state_key": {
-        "type": str,
-        "default": ...,
-        "env": "DIRACX_SERVICE_AUTH_STATE_KEY",
-        "description": "Fernet key used to encrypt/decrypt OAuth2 state.",
-    },
-    "token_issuer": {
-        "type": str,
-        "default": ...,
-        "env": "DIRACX_SERVICE_AUTH_TOKEN_ISSUER",
-        "description": "JWT issuer value ('iss' claim).",
-    },
-    "access_token_expire_minutes": {
-        "type": int,
-        "default": 20,
-        "env": "DIRACX_SERVICE_AUTH_ACCESS_TOKEN_EXPIRE_MINUTES",
-        "description": "Access token lifetime in minutes.",
-    },
-}
+def _parse_env_bool(value: str) -> bool:
+    """Parse a boolean environment variable value."""
+    return TypeAdapter(bool).validate_python(value)
 
-from .extensions import DiracEntryPoint, select_from_extension
 
+class FactorySettings(ServiceSettingsBase):
+    """Factory settings.
 
-def _build_factory_settings_model() -> type[ServiceSettingsBase]:
-    class _BaseFactorySettings(ServiceSettingsBase):
-        """Factory settings.
-
-        Settings which do not fit into dedicated classes,
-        or are dynamically generated
-        """
-
-        diracx_legacy_exchange_hashed_api_key: str = ""
-        enabled_services: dict[str, bool]
-
-    fields: dict[str, tuple[Any, Any]] = {}
-
-    for entry_point in select_from_extension(group=DiracEntryPoint.SERVICES):
-        if "well-known" in entry_point.name:
-            continue
-        fields[f"diracx_service_{entry_point.name}_enabled"] = (
-            bool,
-            Field(
-                default=True,
-                description=f"Enable the {entry_point.name.upper()} router",
-            ),
-        )
-
-    new_mod = create_model(
-        "FactorySettings",
-        __doc__=_BaseFactorySettings.__doc__,
-        __base__=_BaseFactorySettings,
-        # __config__=SettingsConfigDict(
-        #     frozen=True,
-        #     extra="ignore",
-        #     use_attribute_docstrings=True,
-        # ),
-        **fields,
-    )
+    Settings which do not fit into dedicated classes,
+    or are dynamically generated.
+    """
 
-    return new_mod
+    model_config = SettingsConfigDict(use_attribute_docstrings=True)
+
+    diracx_config_backend_url: ConfigSourceUrl | None = None
+    """The URL of the configuration backend.
+    """
+
+    diracx_legacy_exchange_hashed_api_key: str = ""
+    """The hashed API key for the legacy exchange endpoint.
+    """
 
+    diracx_tasks_redis_url: str = "redis://localhost"
+    """The url for the redis server to manage tasks"""
 
-FactorySettings = _build_factory_settings_model()
+    enabled_services: dict[str, bool] = Field(default_factory=dict)
+    """The following environment variables dictates which routers are enabled."""
+
+    opensearch_dbs: dict[str, str] = Field(default_factory=dict)
+    """The following environment variables configure the OpenSearch database connections."""
+
+    sql_dbs: dict[str, str] = Field(default_factory=dict)
+    """The following environment variables configure the SQL database connections."""
+
+    @model_validator(mode="before")
+    @classmethod
+    def load_dotenv_files(cls, data: Any) -> Any:
+        """Load dotenv files before reading settings from environment."""
+        for env_file in dotenv_files_from_environment("DIRACX_SERVICE_DOTENV"):
+            if not dotenv.load_dotenv(env_file):
+                raise NotImplementedError(f"Could not load dotenv file {env_file}")
+        return data
+
+    @field_validator("enabled_services", mode="before")
+    @classmethod
+    def build_enabled_services(cls, value: Any) -> dict[str, bool]:
+        """Build enabled services from the installed service entry points."""
+        enabled_services: dict[str, bool] = {
+            entry_point.name: True
+            for entry_point in select_from_extension(group=DiracEntryPoint.SERVICES)
+            if "well-known" not in entry_point.name
+        }
+
+        for service_name in enabled_services:
+            env_name = f"DIRACX_SERVICE_{service_name.upper()}_ENABLED"
+            if env_value := os.environ.get(env_name):
+                enabled_services[service_name] = _parse_env_bool(env_value)
+
+        if isinstance(value, dict):
+            enabled_services.update(value)
+        return enabled_services
+
+    @field_validator("opensearch_dbs", mode="before")
+    @classmethod
+    def build_opensearch_dbs(cls, value: Any) -> dict[str, str]:
+        """Build OpenSearch database URLs from the installed entry points."""
+        opensearch_dbs: dict[str, str] = {
+            entry_point.name: ""
+            for entry_point in select_from_extension(group=DiracEntryPoint.OS_DB)
+        }
+
+        for db_name in opensearch_dbs:
+            env_name = f"DIRACX_OS_DB_{db_name.upper()}"
+            if env_value := os.environ.get(env_name):
+                opensearch_dbs[db_name] = env_value
+
+        if isinstance(value, dict):
+            opensearch_dbs.update(value)
+        return opensearch_dbs
+
+    @field_validator("sql_dbs", mode="before")
+    @classmethod
+    def build_sql_dbs(cls, value: Any) -> dict[str, str]:
+        """Build SQL database URLs from the installed entry points."""
+        sql_dbs: dict[str, str] = {
+            entry_point.name: ""
+            for entry_point in select_from_extension(group=DiracEntryPoint.SQL_DB)
+        }
+
+        for db_name in sql_dbs:
+            env_name = f"DIRACX_DB_URL_{db_name.upper()}"
+            if env_value := os.environ.get(env_name):
+                sql_dbs[db_name] = env_value
+
+        if isinstance(value, dict):
+            sql_dbs.update(value)
+        return sql_dbs

diracx-db/src/diracx/db/os/utils.py

@@ -3,7 +3,6 @@
 import contextlib
 import json
 import logging
-import os
 from abc import ABCMeta, abstractmethod
 from collections.abc import AsyncIterator
 from contextvars import ContextVar
@@ -14,6 +13,7 @@
 
 from diracx.core.exceptions import InvalidQueryError
 from diracx.core.extensions import DiracEntryPoint, select_from_extension
+from diracx.core.settings import FactorySettings
 from diracx.db.exceptions import DBUnavailableError
 
 logger = logging.getLogger(__name__)
@@ -38,7 +38,8 @@ class BaseOSDB(metaclass=ABCMeta):
     This method returns a dictionary of database names to connection parameters.
     The available databases are determined by the `diracx.dbs.os` entrypoint in
     the `pyproject.toml` file and the connection parameters are taken from the
-    environment variables prefixed with `DIRACX_OS_DB_{DB_NAME}`.
+    `opensearch_dbs` field in FactorySettings, which reads from environment variables
+    prefixed with `DIRACX_OS_DB_{DB_NAME}`.
 
     If extensions to DiracX are being used, there can be multiple implementations
     of the same database. To list the available implementations use
@@ -104,19 +105,26 @@ def available_implementations(cls, db_name: str) -> list[type[BaseOSDB]]:
     def available_urls(cls) -> dict[str, dict[str, Any]]:
         """Return a dict of available OpenSearch database urls.
 
-        The list of available URLs is determined by environment variables
+        The list of available URLs is determined by the opensearch_dbs field
+        in FactorySettings, which reads from environment variables
         prefixed with ``DIRACX_OS_DB_{DB_NAME}``.
         """
+        factory_settings = FactorySettings()
+        opensearch_dbs = factory_settings.opensearch_dbs
+
         conn_kwargs: dict[str, dict[str, Any]] = {}
         for entry_point in select_from_extension(group=DiracEntryPoint.OS_DB):
             db_name = entry_point.name
-            var_name = f"DIRACX_OS_DB_{entry_point.name.upper()}"
-            if var_name in os.environ:
-                try:
-                    conn_kwargs[db_name] = json.loads(os.environ[var_name])
-                except Exception:
-                    logger.error("Error loading connection parameters for %s", db_name)
-                    raise
+            # Get the field value from the OpenSearchDBSettings model
+            if field_value := opensearch_dbs.get(db_name):
+                if field_value:
+                    try:
+                        conn_kwargs[db_name] = json.loads(field_value)
+                    except Exception:
+                        logger.error(
+                            "Error loading connection parameters for %s", db_name
+                        )
+                        raise
         return conn_kwargs
 
     @classmethod

diracx-db/src/diracx/db/sql/utils/base.py

@@ -2,7 +2,6 @@
 
 import contextlib
 import logging
-import os
 import re
 from abc import ABCMeta
 from collections.abc import AsyncIterator
@@ -53,8 +52,9 @@ class BaseSQLDB(metaclass=ABCMeta):
     The available databases are discovered by calling `BaseSQLDB.available_urls`.
     This method returns a mapping of database names to connection URLs. The
     available databases are determined by the `diracx.dbs.sql` entrypoint in the
-    `pyproject.toml` file and the connection URLs are taken from the environment
-    variables of the form `DIRACX_DB_URL_<db-name>`.
+    `pyproject.toml` file and the connection URLs are taken from the
+    `sql_dbs` field in FactorySettings, which reads from environment variables
+    of the form `DIRACX_DB_URL_<db-name>`.
 
     If extensions to DiracX are being used, there can be multiple implementations
     of the same database. To list the available implementations use
@@ -125,16 +125,21 @@ def available_implementations(cls, db_name: str) -> list[type["BaseSQLDB"]]:
     def available_urls(cls) -> dict[str, str]:
         """Return a dict of available database urls.
 
-        The list of available URLs is determined by environment variables
+        The list of available URLs is determined by the sql_dbs field
+        in FactorySettings, which reads from environment variables
         prefixed with ``DIRACX_DB_URL_{DB_NAME}``.
         """
+        from diracx.core.settings import FactorySettings
+
+        factory_settings = FactorySettings()
+        sql_dbs = factory_settings.sql_dbs
+
         db_urls: dict[str, str] = {}
         for entry_point in select_from_extension(group=DiracEntryPoint.SQL_DB):
             db_name = entry_point.name
-            var_name = f"DIRACX_DB_URL_{entry_point.name.upper()}"
-            if var_name in os.environ:
+            # Get the field value from the SqlDBSettings model
+            if db_url := sql_dbs.get(db_name):
                 try:
-                    db_url = os.environ[var_name]
                     if db_url == "sqlite+aiosqlite:///:memory:":
                         db_urls[db_name] = db_url
                     # pydantic does not allow for underscore in scheme

diracx-logic/src/diracx/logic/__main__.py

@@ -93,14 +93,15 @@ async def delete_jwk(args):
 async def cleanup_authdb(args):
     """Maintain AuthDB partitions and remove expired flows."""
     logger.info("Maintaining AuthDB partitions and removing expired flows")
-    import os
 
-    from diracx.core.settings import AuthSettings
+    from diracx.core.settings import AuthSettings, FactorySettings
     from diracx.db.sql import AuthDB
     from diracx.logic.auth.management import cleanup_expired_data
 
     settings = AuthSettings()
-    db_url = os.environ["DIRACX_DB_URL_AUTHDB"]
+    factory_settings = FactorySettings()
+    db_url = factory_settings.sql_dbs.AuthDB
+
     db = AuthDB(db_url)
     async with db.engine_context():
         async with db:

diracx-routers/src/diracx/routers/auth/token.py

@@ -3,7 +3,6 @@
 from __future__ import annotations
 
 import logging
-import os
 from http import HTTPStatus
 from typing import Annotated, Literal
 
@@ -21,7 +20,7 @@
     RefreshTokenPayload,
     TokenResponse,
 )
-from diracx.core.settings import AuthSettings
+from diracx.core.settings import AuthSettings, FactorySettings
 from diracx.db.sql import AuthDB
 from diracx.logic.auth import create_token
 from diracx.logic.auth import get_oidc_token as get_oidc_token_bl
@@ -182,6 +181,7 @@ async def perform_legacy_exchange(
     auth_db: AuthDB,
     available_properties: AvailableSecurityProperties,
     settings: AuthSettings,
+    factory_settings: FactorySettings,
     config: Config,
     all_access_policies: Annotated[
         dict[str, BaseAccessPolicy], Depends(BaseAccessPolicy.all_used_access_policies)
@@ -193,9 +193,7 @@ async def perform_legacy_exchange(
     This route is disabled if DIRACX_LEGACY_EXCHANGE_HASHED_API_KEY is not set
     in the environment.
     """
-    if not (
-        expected_api_key := os.environ.get("DIRACX_LEGACY_EXCHANGE_HASHED_API_KEY")
-    ):
+    if not (expected_api_key := factory_settings.diracx_legacy_exchange_hashed_api_key):
         raise HTTPException(
             status_code=HTTPStatus.SERVICE_UNAVAILABLE,
             detail="Legacy exchange is not enabled",