Config API¶
SchedulerConfig¶
SchedulerConfig
¶
Bases: BaseSettings
调度器主配置类。
支持从以下来源加载配置(优先级从高到低): 1. 直接传入的参数 2. 环境变量(前缀 chronflow_) 3. .env 文件 4. 默认值
示例
使用默认配置¶
config = SchedulerConfig()
自定义配置¶
config = SchedulerConfig(max_workers=20, log_level="DEBUG")
从环境变量(chronflow_MAX_WORKERS=20)¶
config = SchedulerConfig()
validate_timezone(v)
classmethod
¶
验证时区字符串是否有效。
参数
v: 时区字符串,如 'UTC', 'Asia/Shanghai'
返回值
验证通过的时区字符串
抛出
ValueError: 时区字符串无效
parse_backend(value)
classmethod
¶
允许以字符串或映射形式声明队列后端。
validate_backend_exists(value)
classmethod
¶
验证后端是否已注册。
此验证在配置加载时立即执行,避免运行时错误。
from_file(path)
classmethod
¶
从配置文件加载配置。
支持的文件格式: - JSON (.json) - TOML (.toml) - YAML (.yaml, .yml)
参数
path: 配置文件路径
返回值
SchedulerConfig 实例
抛出
FileNotFoundError: 配置文件不存在 ValueError: 不支持的文件格式 ImportError: 缺少必要的解析库(如 PyYAML)
示例
config = SchedulerConfig.from_file("config.toml")
create_backend()
¶
根据配置生成队列后端实例。
TaskMetrics¶
TaskMetrics
¶
Bases: BaseModel
任务执行指标类。
用于统计和记录任务的执行情况,包括成功率、执行时间等关键指标。
属性
total_runs: 总运行次数 successful_runs: 成功运行次数 failed_runs: 失败运行次数 total_execution_time: 总执行时间(秒) average_execution_time: 平均执行时间(秒) last_run_time: 最后一次运行时间(秒) last_success_time: 最后一次成功时间(秒) last_failure_time: 最后一次失败时间(秒) consecutive_failures: 连续失败次数