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

import yaml

from {{project_slug}}.db.config import DBConfig, MultipleDBsConfig
from {{project_slug}}.utils.observability import LoggingConfig, FileLogger, ExporterConfig
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 AppConfig:
    host: str
    port: int
    debug: bool
    cors: CORSConfig


@dataclass
class PrometheusConfig:
    host: str
    port: int
    urls_mapping: dict[str, str] = field(default_factory=dict)


@dataclass
class JaegerConfig:
    endpoint: str


@dataclass
class ObservabilityConfig:
    logging: LoggingConfig
    prometheus: PrometheusConfig | None = None
    jaeger: JaegerConfig | None = None


@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(host="0.0.0.0", port=8080, debug=False, 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(
                    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, urls_mapping={"/api/debug/.*": "/api/debug/*"}),
                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 isinstance(t, UnionType):
            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)