import enum from typing import Annotated, Any from pydantic import MySQLDsn, BeforeValidator, AnyUrl from pydantic_core import MultiHostUrl from pydantic_settings import BaseSettings, SettingsConfigDict from pydantic import computed_field def parse_cors(v: Any) -> list[str] | str: if isinstance(v, str) and not v.startswith("["): return [i.strip() for i in v.split(",")] elif isinstance(v, list | str): return v raise ValueError(v) class Environment(str, enum.Enum): LOCAL = "local" STAGING = "staging" PRODUCTION = "production" class Settings(BaseSettings): model_config = SettingsConfigDict( env_file=".env", env_ignore_empty=True, extra="ignore" ) ENVIRONMENT: Environment = Environment.LOCAL LOG_LEVEL: str = "INFO" STATIC_TOKEN: str = "" TIDB_HOST: str = "127.0.0.1" TIDB_PORT: int = 4000 TIDB_USER: str = "root" TIDB_PASSWORD: str = "" TIDB_DATABASE: str TIDB_SSL: bool = True DB_ECHO: bool = False TiDBCLOUD_PUBLIC_KEY: str = "" TiDBCLOUD_PRIVATE_KEY: str = "" TiDBCLOUD_PROJECT_ID: str = "" TiDBCLOUD_CLUSTER_ID: str = "" OPENAI_API_KEY: str = "" TASK_QUEUE_WORKERS: int = 20 # in seconds TASK_QUEUE_TIMEOUT: int = 600 SESSION_POOL_SIZE: int = 40 BACKEND_CORS_ORIGINS: Annotated[list[AnyUrl] | str, BeforeValidator(parse_cors)] = ( [] ) @computed_field @property def SQLALCHEMY_DATABASE_URI(self) -> MySQLDsn: return MultiHostUrl.build( scheme="mysql+pymysql", username=self.TIDB_USER, password=self.TIDB_PASSWORD, host=self.TIDB_HOST, port=self.TIDB_PORT, path=self.TIDB_DATABASE, query=( "ssl_verify_cert=true&ssl_verify_identity=true" if self.TIDB_SSL else None ), ) @computed_field # type: ignore[misc] @property def SQLALCHEMY_ASYNC_DATABASE_URI(self) -> MySQLDsn: return MultiHostUrl.build( scheme="mysql+asyncmy", username=self.TIDB_USER, password=self.TIDB_PASSWORD, host=self.TIDB_HOST, port=self.TIDB_PORT, path=self.TIDB_DATABASE, ) settings = Settings() # type: ignore