Раис Юсупалиев
114 lines
3.3 KiB
Python
114 lines
3.3 KiB
Python
"""Application configuration split by architecture component."""
|
|
|
|
from decimal import Decimal
|
|
from functools import lru_cache
|
|
from pathlib import Path
|
|
import sys
|
|
from typing import Any
|
|
|
|
from pydantic import BaseModel, Field
|
|
from pydantic_settings import (
|
|
BaseSettings,
|
|
PydanticBaseSettingsSource,
|
|
SettingsConfigDict,
|
|
YamlConfigSettingsSource,
|
|
)
|
|
|
|
DEFAULT_CONFIG_FILE = "/config.yaml"
|
|
TEST_CONFIG_FILE = "config.test.yaml"
|
|
|
|
|
|
class ControllerConfig(BaseModel):
|
|
api_prefix: str = "/api/v1"
|
|
request_id_header: str = "X-Request-ID"
|
|
|
|
|
|
class ServiceConfig(BaseModel):
|
|
provider_timeout_seconds: float = 10.0
|
|
max_parallel_providers: int = 8
|
|
|
|
|
|
class BusinessLogicConfig(BaseModel):
|
|
weight_round_scale: int = 2
|
|
provider_price_multiplier: Decimal = Field(..., gt=0)
|
|
|
|
|
|
class RepositoryConfig(BaseModel):
|
|
redis_dsn: str = "redis://localhost:6379/0"
|
|
price_cache_ttl_seconds: int = 900
|
|
|
|
|
|
class AdapterConfig(BaseModel):
|
|
cdek_base_url: str = "https://api.cdek.ru/v2"
|
|
cdek_client_id: str = ""
|
|
cdek_client_secret: str = ""
|
|
cdek_retry_attempts: int = Field(default=2, ge=0)
|
|
cdek_retry_backoff_seconds: float = Field(default=0.2, ge=0)
|
|
cdek_timeout_seconds: float = Field(default=10.0, gt=0)
|
|
cdek_cache_ttl_seconds: int = Field(default=900, gt=0)
|
|
|
|
|
|
class Settings(BaseSettings):
|
|
model_config = SettingsConfigDict(
|
|
extra="ignore",
|
|
)
|
|
|
|
controller: ControllerConfig = Field(default_factory=ControllerConfig)
|
|
service: ServiceConfig = Field(default_factory=ServiceConfig)
|
|
business_logic: BusinessLogicConfig = Field(default_factory=BusinessLogicConfig)
|
|
repository: RepositoryConfig = Field(default_factory=RepositoryConfig)
|
|
adapter: AdapterConfig = Field(default_factory=AdapterConfig)
|
|
|
|
@classmethod
|
|
def settings_customise_sources(
|
|
cls,
|
|
settings_cls: type[BaseSettings],
|
|
init_settings: PydanticBaseSettingsSource,
|
|
env_settings: PydanticBaseSettingsSource,
|
|
dotenv_settings: PydanticBaseSettingsSource,
|
|
file_secret_settings: PydanticBaseSettingsSource,
|
|
) -> tuple[PydanticBaseSettingsSource, ...]:
|
|
config_file = _resolve_runtime_config_file()
|
|
yaml_source = YamlConfigSettingsSource(
|
|
settings_cls,
|
|
yaml_file=config_file,
|
|
)
|
|
return (
|
|
init_settings,
|
|
yaml_source,
|
|
)
|
|
|
|
|
|
class _RequiredYamlSections(BaseModel):
|
|
controller: dict[str, Any]
|
|
service: dict[str, Any]
|
|
business_logic: dict[str, Any]
|
|
repository: dict[str, Any]
|
|
adapter: dict[str, Any]
|
|
|
|
|
|
def _resolve_runtime_config_file() -> str:
|
|
test_config_path = Path(TEST_CONFIG_FILE)
|
|
if "pytest" in sys.modules and test_config_path.is_file():
|
|
return TEST_CONFIG_FILE
|
|
return DEFAULT_CONFIG_FILE
|
|
|
|
|
|
def _validate_required_yaml_sections() -> None:
|
|
config_file = _resolve_runtime_config_file()
|
|
config_path = Path(config_file)
|
|
if not config_path.is_file():
|
|
return
|
|
|
|
yaml_source = YamlConfigSettingsSource(Settings, yaml_file=config_file)
|
|
yaml_data = yaml_source.yaml_data
|
|
if not isinstance(yaml_data, dict):
|
|
yaml_data = {}
|
|
_RequiredYamlSections.model_validate(yaml_data)
|
|
|
|
|
|
@lru_cache(maxsize=1)
|
|
def get_settings() -> Settings:
|
|
_validate_required_yaml_sections()
|
|
return Settings()
|