import secrets from typing import Optional from typing import Dict from typing import Any from pydantic import BaseSettings from pydantic import validator from pydantic import AnyUrl class MysqlDsn(AnyUrl): allowed_schemes = {'mysql'} user_required = True class RedisDsn(AnyUrl): allowed_schemes = {'redis'} @classmethod def validate_parts(cls, parts: Dict[str, str]) -> Dict[str, str]: defaults = { 'domain': 'localhost' if not (parts['ipv4'] or parts['ipv6']) else '', 'port': '6379' } for key, value in defaults.items(): if not parts[key]: parts[key] = value return super().validate_parts(parts) class Settings(BaseSettings): # JWT SECRET_KEY: str = secrets.token_urlsafe(32) ALGORITHM: str = 'HS256' ACCESS_TOKEN_EXPIRE_MINUTES: int = 60 * 24 * 3 # 数据库 MYSQL_ADDRESS: str MYSQL_USER: str MYSQL_PASSWORD: str MYSQL_DB: str MYSQL_DATABASE_URL: Optional[MysqlDsn] = None @validator("MYSQL_DATABASE_URL", pre=True) def assemble_db_connection(cls, v: Optional[str], values: Dict[str, Any]) -> Any: if isinstance(v, str): return v return MysqlDsn.build( scheme="mysql", user=values.get("MYSQL_USER"), password=values.get("MYSQL_PASSWORD"), host=values.get("MYSQL_ADDRESS"), path=f"/{values.get('MYSQL_DB') or ''}", ) # redis, ip + 端口 REDIS_ADDRESS: str REDIS_URL: Optional[RedisDsn] = None @validator("REDIS_URL", pre=True) def assemble_redis_connection(cls, v: Optional[str], values: Dict[str, Any]) -> Any: if isinstance(v, str): return v return RedisDsn.build( scheme="redis", host=values.get("REDIS_ADDRESS") ) # rabbitMQ RABBITMQ_ADDRESS: str ... # 短信服务 # 邮件服务 # 支付宝信息 # 存储服务信息 # pip freeze > requirements.txt # settings = Settings(_env_file='../.env.development', _env_file_encoding='utf-8') settings = Settings(_env_file='.env.development', _env_file_encoding='utf-8')