template-fastapi/{{project_slug}}/config.py.jinja

"""Application configuration class is defined here."""

from collections import OrderedDict
from dataclasses import asdict, dataclass, field, fields
from pathlib import Path
from types import NoneType, UnionType
from typing import Any, Literal, TextIO, Type, Union, get_origin

import yaml

from {{project_slug}}.db.config import DBConfig, MultipleDBsConfig
from {{project_slug}}.observability.config import (
    ExporterConfig,
    FileLogger,
    JaegerConfig,
    LoggingConfig,
    ObservabilityConfig,
    PrometheusConfig,
)
from {{project_slug}}.utils.secrets import SecretStr, representSecretStrYAML


@dataclass
class CORSConfig:
    allow_origins: list[str]
    allow_methods: list[str]
    allow_headers: list[str]
    allow_credentials: bool


@dataclass
class UvicornConfig:
    host: str
    port: int
    reload: bool = False
    forwarded_allow_ips: list[str] = field(default_factory=list)


@dataclass
class AppConfig:
    uvicorn: UvicornConfig
    debug: bool
    cors: CORSConfig


@dataclass
class {{ProjectName}}Config:
    app: AppConfig
    db: MultipleDBsConfig
    observability: ObservabilityConfig

    def to_order_dict(self) -> OrderedDict:
        """OrderDict transformer."""

        def to_ordered_dict_recursive(obj) -> OrderedDict:
            """Recursive OrderDict transformer."""

            if isinstance(obj, (dict, OrderedDict)):
                return OrderedDict((k, to_ordered_dict_recursive(v)) for k, v in obj.items())
            if isinstance(obj, list):
                return [to_ordered_dict_recursive(item) for item in obj]
            if hasattr(obj, "__dataclass_fields__"):
                return OrderedDict(
                    (field, to_ordered_dict_recursive(getattr(obj, field))) for field in obj.__dataclass_fields__
                )
            return obj

        return OrderedDict([(section, to_ordered_dict_recursive(getattr(self, section))) for section in asdict(self)])

    def dump(self, file: str | Path | TextIO) -> None:
        """Export current configuration to a file"""

        class OrderedDumper(yaml.SafeDumper):
            def represent_dict_preserve_order(self, data):
                return self.represent_dict(data.items())

            def increase_indent(self, flow=False, indentless=False):
                return super().increase_indent(flow, False)

        OrderedDumper.add_representer(OrderedDict, OrderedDumper.represent_dict_preserve_order)
        OrderedDumper.add_representer(SecretStr, representSecretStrYAML)

        if isinstance(file, (str, Path)):
            with open(str(file), "w", encoding="utf-8") as file_w:
                yaml.dump(self.to_order_dict(), file_w, Dumper=OrderedDumper)
        else:
            yaml.dump(self.to_order_dict(), file, Dumper=OrderedDumper)

    @classmethod
    def get_example(cls) -> "{{ProjectName}}Config":
        """Generate an example of configuration."""

        res = cls(
            app=AppConfig(
                uvicorn=UvicornConfig(host="0.0.0.0", port=8080, reload=True, forwarded_allow_ips=["127.0.0.1"]),
                debug=True,
                cors=CORSConfig(["*"], ["*"], ["*"], True),
            ),
            db=MultipleDBsConfig(
                master=DBConfig(
                    host="localhost",
                    port=5432,
                    database="{{project_slug}}_db",
                    user="postgres",
                    password="postgres",
                    pool_size=15,
                ),
                replicas=[
                    DBConfig(
                        host="localhost",
                        port=5433,
                        user="readonly",
                        password="readonly",
                        database="{{project_slug}}_db",
                        pool_size=8,
                    )
                ],
            ),
            observability=ObservabilityConfig(
                logging=LoggingConfig(
                    stderr_level="INFO",
                    root_logger_level="INFO",
                    exporter=ExporterConfig(endpoint="http://127.0.0.1:4317", level="INFO", tls_insecure=True),
                    files=[FileLogger(filename="logs/info.log", level="INFO")],
                ),
                prometheus=PrometheusConfig(host="0.0.0.0", port=9090),
                jaeger=JaegerConfig(endpoint="http://127.0.0.1:4318/v1/traces"),
            ),
        )
        return res

    @classmethod
    def load(cls, file: str | Path | TextIO) -> "{{ProjectName}}Config":
        """Import config from the given filename or raise `ValueError` on error."""

        try:
            if isinstance(file, (str, Path)):
                with open(file, "r", encoding="utf-8") as file_r:
                    data = yaml.safe_load(file_r)
            else:
                data: dict = yaml.safe_load(file)

            return {{ProjectName}}Config._initialize_from_dict({{ProjectName}}Config, data)
        except TypeError as exc:
            raise ValueError(f"Seems like config file is invalid: {file}") from exc
        except Exception as exc:
            raise ValueError(f"Could not read app config file: {file}") from exc

    @staticmethod
    def _initialize_from_dict(t: Type, data: Any) -> Any:
        """Try to initialize given type field-by-field recursively with data from dictionary substituting {} and None
        if no value provided.
        """
        if get_origin(t) is Union or get_origin(t) is UnionType:  # both actually required
            for inner_type in t.__args__:
                if inner_type is NoneType and data is None:
                    return None
                try:
                    return {{ProjectName}}Config._initialize_from_dict(inner_type, data)
                except Exception:  # pylint: disable=broad-except
                    pass
            raise ValueError(f"Cannot instanciate type '{t}' from {data}")

        if hasattr(t, "__origin__") and t.__origin__ is dict:
            return data

        if not isinstance(data, dict):
            if hasattr(t, "__origin__") and t.__origin__ is Literal and data in t.__args__:
                return data
            return t(data)

        init_dict = {}
        for fld in fields(t):
            inner_data = data.get(fld.name)
            if inner_data is None:
                if isinstance(fld.type, UnionType) and NoneType in fld.type.__args__:
                    init_dict[fld.name] = None
                    continue
                inner_data = {}
            else:
                init_dict[fld.name] = {{ProjectName}}Config._initialize_from_dict(fld.type, inner_data)
        return t(**init_dict)

    @classmethod
    def from_file(cls, config_path: str) -> "{{ProjectName}}Config":
        """Load configuration from the given path."""

        if not config_path or not Path(config_path).is_file():
            raise ValueError(f"Requested config is not a valid file: {config_path}")

        return cls.load(config_path)