from __future__ import annotations from pathlib import Path from typing import ClassVar, Literal from urllib.parse import quote from pydantic import ( AnyHttpUrl, BaseModel, Field, SecretStr, computed_field, field_validator, model_validator, ) from pydantic_settings import BaseSettings, SettingsConfigDict def _resolve_project_root() -> Path: current = Path(__file__).resolve() for parent in current.parents: if ( (parent / "pyproject.toml").is_file() and (parent / "backend").is_dir() and (parent / "infra").is_dir() ): return parent for parent in current.parents: if parent.name == "backend": return parent.parent return Path.cwd().resolve() class RuntimeSettings(BaseModel): environment: Literal["dev", "test", "prod"] = "dev" service_name: str = "app" debug: bool = True log_level: str = "INFO" log_json: bool = True log_rotation: Literal["time", "size", "none"] = "time" log_rotation_when: str = "midnight" log_rotation_interval: int = 1 log_rotation_backup_count: int = 14 log_rotation_max_bytes: int = 10_000_000 log_dir: str = "logs" log_error_dir: str = "logs/errors" log_file_name: str = "" log_error_file_name: str = "" log_sensitive_fields: list[str] = Field( default_factory=lambda: [ "password", "secret", "token", "api_key", "authorization", "cookie", "client_ip", "user_id", ] ) sql_log_queries: bool = False trusted_proxy_ips: list[str] = Field(default_factory=list) @field_validator("log_dir", mode="before") @classmethod def lock_log_dir(cls, _: object) -> str: return "logs" @field_validator("log_error_dir", mode="before") @classmethod def lock_log_error_dir(cls, _: object) -> str: return "logs/errors" @model_validator(mode="after") def ensure_service_scoped_log_file_names(self) -> "RuntimeSettings": service = "".join( char if char.isalnum() or char in {"-", "_"} else "-" for char in self.service_name ).strip("-_") service_name = service or "app" if not self.log_file_name.strip(): self.log_file_name = f"{service_name}.log" if not self.log_error_file_name.strip(): self.log_error_file_name = f"{service_name}.error.log" return self class CorsSettings(BaseModel): allow_origins: list[str] = Field( default_factory=lambda: [ "http://localhost", "http://localhost:3000", ] ) allow_credentials: bool = True allow_methods: list[str] = Field(default_factory=lambda: ["*"]) allow_headers: list[str] = Field(default_factory=lambda: ["*"]) class RedisSettings(BaseModel): host: str = "redis" port: int = 6379 password: str | None = None db: int = 0 socket_connect_timeout: float = 1.0 socket_timeout: float = 1.0 max_connections: int = 10 @computed_field @property def url(self) -> str: if self.password: password = quote(self.password, safe="") return f"redis://:{password}@{self.host}:{self.port}/{self.db}" return f"redis://{self.host}:{self.port}/{self.db}" class SupabaseSettings(BaseModel): public_url: AnyHttpUrl anon_key: str = "CHANGE_ME" service_role_key: str = "CHANGE_ME" jwt_secret: SecretStr | None = Field(default=None, exclude=True) jwt_algorithm: Literal["HS256"] = "HS256" jwt_issuer: str | None = None @model_validator(mode="after") def compute_defaults(self) -> "SupabaseSettings": base = str(self.public_url).rstrip("/") if self.jwt_issuer is None: self.jwt_issuer = f"{base}/auth/v1" return self @computed_field @property def url(self) -> str: return str(self.public_url) class StorageSettings(BaseModel): provider: Literal["supabase"] = "supabase" signed_url_ttl_seconds: int = Field(default=600, ge=60, le=3600) retention_days: int = Field(default=30, ge=1, le=3650) class AttachmentSettings(BaseModel): bucket: str = Field(default="eryao-attachments", min_length=3, max_length=63) max_size_mb: int = Field(default=20, ge=1, le=200) class AvatarSettings(BaseModel): bucket: str = Field(default="avatars", min_length=3, max_length=63) max_size_mb: int = Field(default=2, ge=1, le=10) attachment: AttachmentSettings = Field(default_factory=AttachmentSettings) avatar: AvatarSettings = Field(default_factory=AvatarSettings) class LlmSettings(BaseModel): provider_keys: dict[str, str] = Field(default_factory=dict) class DatabaseSettings(BaseModel): host: str = "localhost" port: int = 5432 name: str = "postgres" user: str = "postgres" password: str = "CHANGE_ME" @computed_field @property def url(self) -> str: password = quote(self.password, safe="") return ( f"postgresql+asyncpg://{self.user}:{password}" f"@{self.host}:{self.port}/{self.name}" ) class SensitiveWordSettings(BaseModel): use_aliyun: bool = True fallback_to_local: bool = True class TestSettings(BaseModel): phone: str = "" password: str = "" class TaskiqSettings(BaseModel): broker_url: str | None = None result_backend_url: str | None = None def _resolve_env_file() -> str: current = Path(__file__).resolve() for parent in [current, *current.parents]: candidate = parent / ".env" if candidate.is_file(): return str(candidate) return ".env" PROJECT_ROOT = _resolve_project_root() class Settings(BaseSettings): runtime: RuntimeSettings = RuntimeSettings() cors: CorsSettings = CorsSettings() redis: RedisSettings = RedisSettings() supabase: SupabaseSettings = Field( default_factory=lambda: SupabaseSettings(public_url="http://localhost:8001") ) storage: StorageSettings = StorageSettings() llm: LlmSettings = LlmSettings() database: DatabaseSettings = DatabaseSettings() sensitive_word: SensitiveWordSettings = Field(default_factory=SensitiveWordSettings) test: TestSettings = Field(default_factory=TestSettings) taskiq: TaskiqSettings = Field(default_factory=TaskiqSettings) @computed_field @property def database_url(self) -> str: return self.database.url @computed_field @property def taskiq_broker_url(self) -> str: return self.taskiq.broker_url or self.redis.url @computed_field @property def taskiq_result_backend_url(self) -> str: return self.taskiq.result_backend_url or self.redis.url model_config: ClassVar[SettingsConfigDict] = SettingsConfigDict( env_file=_resolve_env_file(), env_prefix="ERYAO_", env_nested_delimiter="__", case_sensitive=False, extra="ignore", ) config = Settings() # type: ignore[reportCallIssue]