alien_py_cloud/alien_gateway/config.py

from pydantic_settings import BaseSettings, SettingsConfigDict
from typing import Any, Dict, List
from dotenv import load_dotenv
import os
load_dotenv()

class Settings(BaseSettings):
    # 基础配置
    PROJECT_NAME: str = "Alien Cloud Python"
    API_V1_STR: str = "/api/v1"
    
    # 鉴权配置 (原 alien-gateway 职责)
    SECRET_KEY: str = os.getenv("SECRET_KEY")
    ALGORITHM: str = os.getenv("ALGORITHM")
    ACCESS_TOKEN_EXPIRE_MINUTES: int = os.getenv("ACCESS_TOKEN_EXPIRE_MINUTES")  # 7天
    
    # 数据库配置
    DB_USER: str = os.getenv("DB_USER")
    DB_PASSWORD: str = os.getenv("DB_PASSWORD")
    DB_HOST: str = os.getenv("DB_HOST")
    DB_PORT: int = os.getenv("DB_PORT")
    DB_NAME: str = os.getenv("DB_NAME")

    # Redis Sentinel 高可用配置
    # 例: REDIS_SENTINELS=192.168.2.251:36379,192.168.2.252:36379,192.168.2.253:36379
    REDIS_SENTINELS: str = os.getenv("REDIS_SENTINELS", "")
    REDIS_MASTER_NAME: str = os.getenv("REDIS_MASTER_NAME", "mymaster")
    REDIS_PASSWORD: str = os.getenv("REDIS_PASSWORD", "")
    REDIS_DB: int = int(os.getenv("REDIS_DB", "0"))
    REDIS_SENTINEL_USERNAME: str = os.getenv("REDIS_SENTINEL_USERNAME", "")
    REDIS_SENTINEL_PASSWORD: str = os.getenv("REDIS_SENTINEL_PASSWORD", "")
    REDIS_SOCKET_TIMEOUT: float = float(os.getenv("REDIS_SOCKET_TIMEOUT", "0.5"))
    REDIS_CONNECT_TIMEOUT: float = float(os.getenv("REDIS_CONNECT_TIMEOUT", "1.0"))
    # 下游服务地址
    STORE_BASE_URL: str = os.getenv("STORE_BASE_URL")  # alien_store 服务地址

    # 阿里云短信配置
    ALIYUN_SMS_SIGN_NAME_CONTRACT: str = os.getenv("ALIYUN_SMS_SIGN_NAME_CONTRACT")
    ALIYUN_SMS_TEMPLATE_CODE_CONTRACT: str = os.getenv("ALIYUN_SMS_TEMPLATE_CODE_CONTRACT")
    ALIYUN_ACCESS_KEY_ID: str = os.getenv("ALIYUN_ACCESS_KEY_ID")
    ALIYUN_ACCESS_KEY_SECRET: str = os.getenv("ALIYUN_ACCESS_KEY_SECRET")



    @property
    def SQLALCHEMY_DATABASE_URI(self) -> str:
        return f"mysql+pymysql://{self.DB_USER}:{self.DB_PASSWORD}@{self.DB_HOST}:{self.DB_PORT}/{self.DB_NAME}"

    @property
    def REDIS_SENTINEL_NODES(self) -> List[tuple[str, int]]:
        nodes: List[tuple[str, int]] = []
        for item in self.REDIS_SENTINELS.split(","):
            entry = item.strip()
            if not entry:
                continue
            if ":" not in entry:
                raise ValueError(f"Invalid REDIS_SENTINELS entry: {entry}")
            host, port = entry.split(":", 1)
            nodes.append((host.strip(), int(port.strip())))
        return nodes

    @property
    def REDIS_SENTINEL_KWARGS(self) -> Dict[str, Any]:
        kwargs: Dict[str, Any] = {}
        if self.REDIS_SENTINEL_USERNAME:
            kwargs["username"] = self.REDIS_SENTINEL_USERNAME
        if self.REDIS_SENTINEL_PASSWORD:
            kwargs["password"] = self.REDIS_SENTINEL_PASSWORD
        return kwargs

    @property
    def REDIS_SENTINEL_URL(self) -> str:
        # Celery sentinel transport 需要 "sentinel://host:port;sentinel://host:port"
        return ";".join(
            f"sentinel://{host}:{port}" for host, port in self.REDIS_SENTINEL_NODES
        )

    @property
    def REDIS_SENTINEL_TRANSPORT_OPTIONS(self) -> Dict[str, Any]:
        return {
            "master_name": self.REDIS_MASTER_NAME,
            "sentinel_kwargs": self.REDIS_SENTINEL_KWARGS,
            "password": self.REDIS_PASSWORD,
            "db": self.REDIS_DB,
        }

    model_config = SettingsConfigDict(
        case_sensitive=True,
        env_file=".env",
    )

settings = Settings()