Files
Mutual_Fund/config/settings.py

144 lines
4.9 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""应用配置:全部从 .env 读取,代码不硬编码任何连接/模型值。
.env 路径由本文件位置推导(绝对路径),与启动工作目录无关;每个子配置各自声明
env_file,保证嵌套配置也读文件(否则默认实例会吞掉环境变量);任一必需键缺失,
启动即报错(fail-fast),避免带着错误配置静默运行。
"""
from pathlib import Path
from typing import Optional
from pydantic_settings import BaseSettings, SettingsConfigDict
from sqlalchemy.engine import URL
# 项目根 = settings.py 的上两级;无论从哪个 cwd 启动都能定位 .env
_ENV_FILE: Path = Path(__file__).resolve().parent.parent / ".env"
class MysqlCfg(BaseSettings):
host: str
port: int
user: str
password: str
database: str
pool_size: int
max_overflow: int
pool_recycle: int
connect_timeout: int
pool_timeout: int
model_config = SettingsConfigDict(env_prefix="MYSQL_", env_file=_ENV_FILE, extra="ignore")
@property
def url(self) -> URL:
return URL.create(
"mysql+aiomysql",
username=self.user,
password=self.password,
host=self.host,
port=self.port,
database=self.database,
query={"charset": "utf8mb4"},
)
class RedisCfg(BaseSettings):
host: str
port: int
db: int
password: Optional[str] = None # 可缺省(空串/未设置表示无密码)
socket_connect_timeout: int
socket_timeout: int
health_check_interval: int
max_connections: int
model_config = SettingsConfigDict(env_prefix="REDIS_", env_file=_ENV_FILE, extra="ignore")
class Neo4jCfg(BaseSettings):
uri: str
user: str
password: str
max_connection_pool_size: int
connection_acquisition_timeout: int
connection_timeout: int
max_transaction_retry_time: int
model_config = SettingsConfigDict(env_prefix="NEO4J_", env_file=_ENV_FILE, extra="ignore")
class MilvusCfg(BaseSettings):
uri: str
token: Optional[str] = None # 可缺省(无需鉴权时留空)
user: Optional[str] = None
password: Optional[str] = None
db: Optional[str] = None
connect_timeout: int # 建连/通道就绪超时(构造是急切连接,必须短)
timeout: int # 数据操作超时,调用处可覆盖
model_config = SettingsConfigDict(env_prefix="MILVUS_", env_file=_ENV_FILE, extra="ignore")
@property
def db_name(self) -> Optional[str]:
"""Compatibility name used by the Milvus client wrapper."""
return self.db
class LLMCfg(BaseSettings):
"""大模型配置:本地 Ollama / OpenAI 兼容 API 双模式见 tool/llm.py。"""
mode: str = "auto" # auto=本地优先、API 兜底(按下方参数是否填写自动判定);ollama / api=强制单一后端
# —— 本地 Ollama(base + chat_model 填写即启用)——
ollama_base: str = ""
ollama_chat_model: str = ""
ollama_embed_model: str = ""
# —— OpenAI 兼容 API(api_key + base + chat_model 填写即启用;base_url 以 /v1 结尾)——
api_base: str = ""
api_key: str = ""
api_chat_model: str = ""
api_embed_model: str = ""
embed_dimensions: int # 向量维度:既作为 embeddings 请求的 dimensions 参数,也是 Milvus 建表/入库校验的维度;须与模型输出一致(MRL 模型如 qwen3-embedding 可任选 32~4096)
# —— 通用超参 ——
temperature: float
max_tokens: int
timeout: int
max_retries: int
retry_backoff_sec: float # 指数退避基数
fallback_chat_model: str # 备用模型:主模型失败后自动切换(空则不启用)
model_config = SettingsConfigDict(env_prefix="LLM_", env_file=_ENV_FILE, extra="ignore")
class JwtCfg(BaseSettings):
"""JWT 鉴权配置(service/auth.py)。"""
secret: str # 签名密钥,生产必须换强随机值
algorithm: str # 签名算法,如 HS256
expire_seconds: int # Token 有效期(秒)
model_config = SettingsConfigDict(env_prefix="JWT_", env_file=_ENV_FILE, extra="ignore")
class DBMaintenanceCfg(BaseSettings):
"""数据库启动策略:连接重试与严格模式(config/database/__init__.py)。"""
strict_startup: bool = False # true=任一核心库启动失败即阻止应用启动(生产建议开启)
conn_retries: int = 3 # 各库 init 最大尝试次数
retry_backoff_sec: float = 1.0 # 指数退避基数:1s/2s/4s
model_config = SettingsConfigDict(env_prefix="DB_", env_file=_ENV_FILE, extra="ignore")
class Settings(BaseSettings):
app_env: str
db: DBMaintenanceCfg = DBMaintenanceCfg()
jwt: JwtCfg = JwtCfg()
mysql: MysqlCfg = MysqlCfg()
redis: RedisCfg = RedisCfg()
neo4j: Neo4jCfg = Neo4jCfg()
milvus: MilvusCfg = MilvusCfg()
llm: LLMCfg = LLMCfg()
model_config = SettingsConfigDict(env_file=_ENV_FILE, extra="ignore")
settings = Settings()