feat: BI-5983 add dl-settings

lib/dl_auth_api_lib/dl_auth_api_lib/app.py

@@ -21,7 +21,7 @@
 from dl_auth_api_lib.oauth.yandex import YandexOAuthClient
 from dl_auth_api_lib.settings import (
     AuthAPISettings,
-    register_auth_client,
+    BaseOAuthClient,
 )
 from dl_auth_api_lib.views import google as google_views
 from dl_auth_api_lib.views import snowflake as snowflake_views
@@ -90,5 +90,5 @@ def create_app(self, app_version: str | None = None) -> web.Application:
         return app
 
 
-register_auth_client("yandex", YandexOAuthClient)
-register_auth_client("google", GoogleOAuthClient)
+BaseOAuthClient.register("yandex", YandexOAuthClient)
+BaseOAuthClient.register("google", GoogleOAuthClient)

lib/dl_auth_api_lib/dl_auth_api_lib/oauth/google.py

@@ -15,7 +15,7 @@
 
 
 class GoogleOAuthClient(BaseOAuthClient):
-    auth_type: str = "google"
+    type: str = pydantic.Field(alias="auth_type", default="google")
 
     client_id: str
     client_secret: str

lib/dl_auth_api_lib/dl_auth_api_lib/oauth/yandex.py

@@ -16,7 +16,7 @@
 
 
 class YandexOAuthClient(BaseOAuthClient):
-    auth_type: str = "yandex"
+    type: str = pydantic.Field(alias="auth_type", default="yandex")
 
     client_id: str
     client_secret: str

lib/dl_auth_api_lib/dl_auth_api_lib/settings.py

@@ -1,67 +1,13 @@
-import os
-from typing import (
-    Annotated,
-    Any,
-    Type,
-)
-
 import pydantic
-import pydantic_settings
 
-from dl_auth_api_lib.utils.pydantic import make_dict_factory
+import dl_settings
 
 
-class BaseOAuthClient(pydantic.BaseModel):
+class BaseOAuthClient(dl_settings.TypedBaseModel):
+    type: str = pydantic.Field(alias="auth_type")
     conn_type: str
-    auth_type: str
-
-    @classmethod
-    def factory(cls, data: Any) -> "BaseOAuthClient":
-        if isinstance(data, BaseOAuthClient):
-            return data
-
-        assert isinstance(data, dict), "OAuthClient settings must be a dict"
-        assert "auth_type" in data, "No auth_type in client"
-        assert data["auth_type"] in _REGISTRY, f"No such OAuth type: {data['auth_type']}"
-
-        config_class = _REGISTRY[data["auth_type"]]
-        return config_class.model_validate(data)
-
-
-_REGISTRY: dict[str, type[BaseOAuthClient]] = {}
 
 
-def register_auth_client(
-    name: str,
-    auth_type: type[BaseOAuthClient],
-) -> None:
-    _REGISTRY[name] = auth_type
-
-
-class AuthAPISettings(pydantic_settings.BaseSettings):
-    model_config = pydantic_settings.SettingsConfigDict(env_nested_delimiter="__")
-
-    auth_clients: Annotated[
-        dict[str, pydantic.SerializeAsAny[BaseOAuthClient]],
-        pydantic.BeforeValidator(make_dict_factory(BaseOAuthClient.factory)),
-    ] = pydantic.Field(default=dict())
-
+class AuthAPISettings(dl_settings.BaseRootSettings):
+    auth_clients: dl_settings.TypedDictAnnotation[BaseOAuthClient] = pydantic.Field(default=dict())
     sentry_dsn: str | None = pydantic.Field(default=None)
-
-    @classmethod
-    def settings_customise_sources(
-        cls,
-        settings_cls: Type[pydantic_settings.BaseSettings],
-        init_settings: pydantic_settings.PydanticBaseSettingsSource,
-        env_settings: pydantic_settings.PydanticBaseSettingsSource,
-        dotenv_settings: pydantic_settings.PydanticBaseSettingsSource,
-        file_secret_settings: pydantic_settings.PydanticBaseSettingsSource,
-    ) -> tuple[pydantic_settings.PydanticBaseSettingsSource, ...]:
-        return (
-            env_settings,
-            pydantic_settings.YamlConfigSettingsSource(
-                settings_cls,
-                yaml_file=os.environ.get("CONFIG_PATH", None),
-            ),
-            init_settings,
-        )

lib/dl_auth_api_lib/dl_auth_api_lib_tests/conftest.py

@@ -2,9 +2,10 @@
 import logging
 import os
 
-import aiohttp.pytest_plugin
+from aiohttp.pytest_plugin import aiohttp_client
 from aiohttp.typedefs import Middleware
 import pytest
+import pytest_asyncio
 
 from dl_api_commons.base_models import (
     NoAuthData,
@@ -20,12 +21,8 @@
 LOGGER = logging.getLogger(__name__)
 
 
-pytest_plugins = ("aiohttp.pytest_plugin",)
-
-try:
-    del aiohttp.pytest_plugin.loop
-except AttributeError:
-    pass
+# Fixtures
+aiohttp_client = aiohttp_client
 
 
 @pytest.fixture(autouse=True)
@@ -98,7 +95,7 @@ def oauth_app(loop, aiohttp_client, oauth_app_settings):
     return loop.run_until_complete(aiohttp_client(app))
 
 
-@pytest.fixture(scope="function")
+@pytest_asyncio.fixture(scope="function")
 async def oauth_app_client(oauth_app) -> DLCommonAPIClient:
     async with get_default_aiohttp_session() as session:
         yield DLCommonAPIClient(

lib/dl_auth_api_lib/pyproject.toml

@@ -20,6 +20,7 @@ typing-extensions = ">=4.9.0"
 dl-api-commons = {path = "../dl_api_commons"}
 dl-core = {path = "../dl_core"}
 dl-constants = {path = "../dl_constants"}
+dl-settings = {path = "../dl_settings"}
 
 [tool.poetry.group.tests.dependencies]
 pytest = ">=7.2.2"
@@ -48,6 +49,7 @@ target_path = "ext"
 labels = ["ext_public"]
 
 [tool.mypy]
+exclude = ["dl_auth_api_lib_tests/"]
 warn_unused_configs = true
 disallow_untyped_defs = true
 check_untyped_defs = true

lib/dl_settings/dl_settings/__init__.py

@@ -0,0 +1,18 @@
+from .base import (
+    BaseRootSettings,
+    BaseSettings,
+    TypedAnnotation,
+    TypedBaseModel,
+    TypedDictAnnotation,
+    TypedListAnnotation,
+)
+
+
+__all__ = [
+    "BaseSettings",
+    "BaseRootSettings",
+    "TypedBaseModel",
+    "TypedAnnotation",
+    "TypedListAnnotation",
+    "TypedDictAnnotation",
+]

lib/dl_settings/dl_settings/base/__init__.py

@@ -0,0 +1,20 @@
+from .settings import (
+    BaseRootSettings,
+    BaseSettings,
+)
+from .typed import (
+    TypedAnnotation,
+    TypedBaseModel,
+    TypedDictAnnotation,
+    TypedListAnnotation,
+)
+
+
+__all__ = [
+    "BaseSettings",
+    "BaseRootSettings",
+    "TypedBaseModel",
+    "TypedAnnotation",
+    "TypedListAnnotation",
+    "TypedDictAnnotation",
+]

lib/dl_settings/dl_settings/base/settings.py

@@ -0,0 +1,56 @@
+import os
+import typing
+
+import pydantic_settings
+
+
+class BaseSettings(pydantic_settings.BaseSettings):
+    ...
+
+
+class BaseRootSettings(BaseSettings):
+    model_config = pydantic_settings.SettingsConfigDict(
+        env_nested_delimiter="__",
+    )
+
+    @classmethod
+    def _get_yaml_source_paths(cls) -> list[str]:
+        config_paths = os.environ.get("CONFIG_PATH", None)
+
+        if not config_paths:
+            return []
+
+        return [path.strip() for path in config_paths.split(",")]
+
+    @classmethod
+    def _get_yaml_sources(cls) -> list[pydantic_settings.YamlConfigSettingsSource]:
+        return [
+            pydantic_settings.YamlConfigSettingsSource(
+                cls,
+                yaml_file=yaml_file,
+            )
+            for yaml_file in cls._get_yaml_source_paths()
+        ]
+
+    @classmethod
+    def settings_customise_sources(
+        cls,
+        settings_cls: typing.Type[pydantic_settings.BaseSettings],
+        init_settings: pydantic_settings.PydanticBaseSettingsSource,
+        env_settings: pydantic_settings.PydanticBaseSettingsSource,
+        dotenv_settings: pydantic_settings.PydanticBaseSettingsSource,
+        file_secret_settings: pydantic_settings.PydanticBaseSettingsSource,
+    ) -> tuple[pydantic_settings.PydanticBaseSettingsSource, ...]:
+        return (
+            env_settings,
+            *cls._get_yaml_sources(),
+            dotenv_settings,
+            file_secret_settings,
+            init_settings,
+        )
+
+
+__all__ = [
+    "BaseSettings",
+    "BaseRootSettings",
+]

lib/dl_settings/dl_settings/base/typed.py

@@ -0,0 +1,108 @@
+import typing
+
+import pydantic
+import pydantic._internal._model_construction as pydantic_model_construction
+import pydantic.fields
+
+import dl_settings.base.settings as base_settings
+
+
+class TypedMeta(pydantic_model_construction.ModelMetaclass):
+    def __init__(cls, name: str, bases: tuple[type, ...], attrs: dict[str, typing.Any]):
+        cls._classes: dict[str, type[base_settings.BaseSettings]] = {}
+
+
+class TypedBaseModel(base_settings.BaseSettings, metaclass=TypedMeta):
+    type: str = pydantic.Field(alias="type")
+
+    @classmethod
+    def __get_type_field_name(cls) -> str:
+        alias = cls.model_fields["type"].alias
+
+        if alias is None:
+            raise ValueError("Field 'type' must have alias")
+
+        return alias
+
+    @classmethod
+    def register(cls, name: str, class_: typing.Type) -> None:
+        if name in cls._classes:
+            raise ValueError(f"Class with name '{name}' already registered")
+
+        if not issubclass(class_, cls):
+            raise ValueError(f"Class '{class_}' must be subclass of '{cls}'")
+
+        cls._classes[name] = class_
+
+    @classmethod
+    def factory(cls, data: typing.Any) -> base_settings.BaseSettings:
+        if isinstance(data, base_settings.BaseSettings):
+            return data
+
+        if not isinstance(data, dict):
+            raise ValueError("Data must be dict")
+
+        class_name = data[cls.__get_type_field_name()]
+        if class_name not in cls._classes:
+            raise ValueError(f"Unknown type: {class_name}")
+
+        class_ = cls._classes[class_name]
+
+        return class_.model_validate(data)
+
+    @classmethod
+    def list_factory(cls, data: list[typing.Any]) -> typing.List[base_settings.BaseSettings]:
+        if not isinstance(data, list):
+            raise ValueError("Data must be sequence for list factory")
+
+        return [cls.factory(item) for item in data]
+
+    @classmethod
+    def dict_factory(cls, data: dict[str, typing.Any]) -> typing.Dict[str, base_settings.BaseSettings]:
+        if not isinstance(data, dict):
+            raise ValueError("Data must be mapping for dict factory")
+
+        import logging
+
+        logging.error(data)
+
+        return {key: cls.factory(value) for key, value in data.items()}
+
+
+TypedBaseModelT = typing.TypeVar("TypedBaseModelT", bound=TypedBaseModel)
+
+
+if typing.TYPE_CHECKING:
+    TypedAnnotation = typing.Annotated[TypedBaseModelT, ...]
+    TypedListAnnotation = typing.Annotated[list[TypedBaseModelT], ...]
+    TypedDictAnnotation = typing.Annotated[dict[str, TypedBaseModelT], ...]
+else:
+
+    class TypedAnnotation:
+        def __class_getitem__(cls, base_class: TypedBaseModelT) -> typing.Any:
+            return typing.Annotated[
+                pydantic.SerializeAsAny[base_class],
+                pydantic.BeforeValidator(base_class.factory),
+            ]
+
+    class TypedListAnnotation:
+        def __class_getitem__(cls, base_class: TypedBaseModelT) -> typing.Any:
+            return typing.Annotated[
+                list[pydantic.SerializeAsAny[base_class]],
+                pydantic.BeforeValidator(base_class.list_factory),
+            ]
+
+    class TypedDictAnnotation:
+        def __class_getitem__(cls, base_class: TypedBaseModelT) -> typing.Any:
+            return typing.Annotated[
+                dict[str, pydantic.SerializeAsAny[base_class]],
+                pydantic.BeforeValidator(base_class.dict_factory),
+            ]
+
+
+__all__ = [
+    "TypedBaseModel",
+    "TypedAnnotation",
+    "TypedListAnnotation",
+    "TypedDictAnnotation",
+]