from functools import lru_cache from pathlib import Path from urllib.parse import quote_plus from pydantic import Field from pydantic_settings import BaseSettings, SettingsConfigDict ENV_FILE = Path(__file__).resolve().parent.parent / ".env" class Settings(BaseSettings): model_config = SettingsConfigDict( env_file=str(ENV_FILE), env_file_encoding="utf-8", extra="ignore", ) app_name: str = Field(default="mini-program-api", validation_alias="APP_NAME") app_version: str = Field(default="0.1.0", validation_alias="APP_VERSION") app_description: str = Field( default="Backend service for mini program", validation_alias="APP_DESCRIPTION", ) host: str = Field(default="0.0.0.0", validation_alias="HOST") port: int = Field(default=8001, validation_alias="PORT") db_type: str = Field(default="mysql", validation_alias="DB_TYPE") db_host: str = Field(default="127.0.0.1", validation_alias="DB_HOST") db_port: int = Field(default=3306, validation_alias="DB_PORT") db_user: str = Field(default="root", validation_alias="DB_USER") db_password: str = Field(default="", validation_alias="DB_PASSWORD") db_name: str = Field(default="talkingq", validation_alias="DB_NAME") db_auto_init_tables: bool = Field(default=False, validation_alias="DB_AUTO_INIT_TABLES") db_path: str = Field(default="./data.db", validation_alias="DB_PATH") wechat_app_id: str = Field(default="", validation_alias="WECHAT_APP_ID") wechat_app_secret: str = Field(default="", validation_alias="WECHAT_APP_SECRET") wechat_api_base_url: str = Field( default="https://api.weixin.qq.com", validation_alias="WECHAT_API_BASE_URL", ) wechat_http_timeout_seconds: float = Field( default=5.0, validation_alias="WECHAT_HTTP_TIMEOUT_SECONDS", ) cos_secret_id: str = Field(default="", validation_alias="COS_SECRET_ID") cos_secret_key: str = Field(default="", validation_alias="COS_SECRET_KEY") cos_region: str = Field(default="", validation_alias="COS_REGION") cos_bucket_message: str = Field(default="", validation_alias="COS_BUCKET_MESSAGE") cos_bucket_ava: str = Field(default="", validation_alias="COS_BUCKET_AVA") cos_public_base_url: str = Field(default="", validation_alias="COS_PUBLIC_BASE_URL") cos_avatar_prefix: str = Field(default="avatars/", validation_alias="COS_AVATAR_PREFIX") cos_avatar_url_expire_seconds: int = Field( default=86400, validation_alias="COS_AVATAR_URL_EXPIRE_SECONDS", ) cos_avatar_max_bytes: int = Field( default=2 * 1024 * 1024, validation_alias="COS_AVATAR_MAX_BYTES", ) jwt_secret: str = Field( default="dev_only_change_jwt_secret", validation_alias="JWT_SECRET", ) jwt_algorithm: str = Field(default="HS256", validation_alias="JWT_ALGORITHM") jwt_access_token_expire_minutes: int = Field( default=60, validation_alias="JWT_ACCESS_TOKEN_EXPIRE_MINUTES", ) log_level: str = Field(default="INFO", validation_alias="LOG_LEVEL") log_json: bool = Field(default=False, validation_alias="LOG_JSON") slow_sql_ms: int = Field(default=200, validation_alias="SLOW_SQL_MS") @property def mysql_dsn(self) -> str: user = quote_plus(self.db_user) password = quote_plus(self.db_password) auth = f"{user}:{password}" if password else user return ( f"mysql+pymysql://{auth}@{self.db_host}:{self.db_port}/" f"{self.db_name}?charset=utf8mb4" ) @property def sqlite_dsn(self) -> str: db_path = Path(self.db_path) db_path.parent.mkdir(parents=True, exist_ok=True) return f"sqlite:///{db_path.absolute()}" @property def database_dsn(self) -> str: if self.db_type == "sqlite": return self.sqlite_dsn return self.mysql_dsn @property def uses_shared_talkingq_db(self) -> bool: return self.db_type == "mysql" and self.db_name == "talkingq" @lru_cache(maxsize=1) def get_settings() -> Settings: return Settings() settings = get_settings()