37d282c0a2
- 添加 pytest 配置和测试依赖到 requirements.txt - 创建测试包结构和 fixtures (conftest.py) - 添加数据库模块的 CRUD 操作测试 (test_database.py) - 添加 Provider 插件系统测试 (test_providers.py) - 添加调度器模块测试 (test_scheduler.py) - 添加 API 路由测试 (test_api.py) - 添加回归测试覆盖边界条件和错误处理 (test_regressions.py) - 添加健康检查端点用于容器监控 - 修复调度器中的日历计算逻辑和任务执行参数处理 - 更新数据库函数以返回操作结果状态
80 lines
2.2 KiB
Python
80 lines
2.2 KiB
Python
"""配置加载模块 -- 从 config.yaml / 环境变量读取配置"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import yaml
|
|
from pydantic import BaseModel, Field
|
|
|
|
|
|
_CONFIG_PATH = os.getenv("CONFIG_PATH", "config.yaml")
|
|
|
|
|
|
class ServerConfig(BaseModel):
|
|
host: str = "0.0.0.0"
|
|
port: int = 8080
|
|
proxy_api_key: str = "sk-plan-manage-change-me"
|
|
|
|
|
|
class DatabaseConfig(BaseModel):
|
|
path: str = "./data/plan_manage.db"
|
|
|
|
|
|
class StorageConfig(BaseModel):
|
|
path: str = "./data/files"
|
|
|
|
|
|
class SchedulerConfig(BaseModel):
|
|
"""调度器配置"""
|
|
task_processing_limit: int = 5 # 每次处理的最大任务数
|
|
loop_interval_seconds: int = 30 # 主循环间隔（秒）
|
|
|
|
|
|
class QuotaRuleSeed(BaseModel):
|
|
"""config.yaml 中单条 QuotaRule 种子"""
|
|
rule_name: str
|
|
quota_total: int
|
|
quota_unit: str = "requests"
|
|
refresh_type: str = "calendar_cycle"
|
|
interval_hours: float | None = None
|
|
calendar_unit: str | None = None
|
|
calendar_anchor: dict[str, Any] | None = None
|
|
|
|
|
|
class PlanSeed(BaseModel):
|
|
"""config.yaml 中单个 Plan 种子配置"""
|
|
name: str
|
|
provider: str
|
|
api_key: str = ""
|
|
api_base: str = ""
|
|
plan_type: str = "coding"
|
|
supported_models: list[str] = Field(default_factory=list)
|
|
extra_headers: dict[str, str] = Field(default_factory=dict)
|
|
extra_config: dict[str, Any] = Field(default_factory=dict)
|
|
quota_rules: list[QuotaRuleSeed] = Field(default_factory=list)
|
|
|
|
|
|
class AppConfig(BaseModel):
|
|
server: ServerConfig = Field(default_factory=ServerConfig)
|
|
database: DatabaseConfig = Field(default_factory=DatabaseConfig)
|
|
storage: StorageConfig = Field(default_factory=StorageConfig)
|
|
scheduler: SchedulerConfig = Field(default_factory=SchedulerConfig)
|
|
plans: list[PlanSeed] = Field(default_factory=list)
|
|
|
|
|
|
def load_config(path: str | None = None) -> AppConfig:
|
|
"""加载配置文件，不存在则返回默认值"""
|
|
cfg_path = Path(path or _CONFIG_PATH)
|
|
if cfg_path.exists():
|
|
with open(cfg_path, "r", encoding="utf-8") as f:
|
|
raw = yaml.safe_load(f) or {}
|
|
return AppConfig(**raw)
|
|
return AppConfig()
|
|
|
|
|
|
# 全局单例
|
|
settings: AppConfig = load_config()
|