epscheduler 0.1.1

可扩展的 Python 定时任务调度框架

epscheduler

可扩展的 Python 定时任务调度框架，支持 DAG 工作流。

特性

• 三种调度器模式：Async（异步应用）、Blocking（简单脚本）、Background（Web/GUI 后台）
• 四种触发器：Cron（cron 表达式）、Interval（固定间隔）、Date（一次性）、Manual（手动触发）
• DAG 工作流：条件分支、循环、并行执行、生命周期钩子
• 多种节点类型：Func（函数）、Condition（条件分支）、Loop（循环）、Script（脚本）、Command（命令）
• 多种存储后端：Memory（内存）、File（文件持久化）、SQLite、多数据库 ORM（PostgreSQL/MySQL 等）
• 多种执行器：Async（协程池）、Thread（线程池）、Sync（同步）
• 完善的类型标注：Python 3.11+ 语法，支持静态类型检查

安装

pip install epscheduler

依赖要求：

• Python 3.11+
• pydantic >= 2.0
• croniter >= 2.0

快速上手

基础示例

from epscheduler import BlockingScheduler
from epscheduler.triggers import IntervalTrigger

scheduler = BlockingScheduler()

def hello():
    print("Hello, World!")

scheduler.add_job(hello, IntervalTrigger(seconds=5), job_id="hello")
scheduler.start()

DAG 工作流示例

from epscheduler import BlockingScheduler
from epscheduler.dag import DAG
from epscheduler.nodetypes import FuncNode
from epscheduler.triggers import CronTrigger

dag = DAG(params={"count": 0})

# 定义节点
node_a = FuncNode(fn=lambda ctx: {"step": "a"}, id="a", name="Step A")
node_b = FuncNode(fn=lambda ctx: {"step": "b"}, id="b", name="Step B")
node_c = FuncNode(fn=lambda ctx: {"result": ctx.get("count", 0) + 1}, id="c", name="Step C")

# 构建拓扑
dag.add_node(node_a)
dag.add_node(node_b)
dag.add_node(node_c)
dag.add_edge("a", "b")  # A → B
dag.add_edge("b", "c")  # B → C

# 调度 DAG 作业
scheduler = BlockingScheduler()
scheduler.add_dag_job(dag, CronTrigger(cron_expression="0 9 * * *"), job_id="daily_task")
scheduler.start()

核心组件

Scheduler（调度器）

调度器	适用场景	使用方式
AsyncScheduler	asyncio 应用	await scheduler.start()
BlockingScheduler	简单脚本	scheduler.start() 阻塞
BackgroundScheduler	Web/GUI 应用	scheduler.start() 后台线程

Trigger（触发器）

触发器	说明	示例
CronTrigger	Cron 表达式	CronTrigger(cron_expression="0 9 * * *")
IntervalTrigger	固定间隔	IntervalTrigger(seconds=60)
DateTrigger	一次性触发	DateTrigger(run_date=datetime(2025, 1, 1))
ManualTrigger	手动触发	ManualTrigger()

JobStore（作业存储）

存储后端	说明	适用场景
MemoryJobStore	内存存储	开发/测试
FileJobStore	文件持久化	简单场景
SQLiteJobStore	SQLite 数据库	单机应用
DBJobStore	多数据库 ORM	生产环境（PostgreSQL/MySQL 等）

Executor（执行器）

执行器	说明	适用场景
AsyncExecutor	协程池	I/O 密集型
ThreadExecutor	线程池	CPU 密集型
SyncExecutor	同步执行	测试

DAG（工作流）

DAG 支持复杂工作流编排：

• 条件分支：ConditionNode 根据上下文选择分支
• 循环：LoopNode 重复执行直到条件满足
• 并行：max_parallelism 控制并发度
• 钩子：DAGHooks 在节点执行前后插入自定义逻辑

NodeType（节点类型）

节点类型	说明
FuncNode	执行 Python 函数
ConditionNode	条件分支路由
LoopNode	循环执行
ScriptNode	执行外部 Python 脚本
CommandNode	执行终端命令

开发指南

# 运行测试
PYTHONPATH=src python3 -m pytest tests/ --tb=short

# 代码检查
python3 -m ruff check src/ tests/

# 类型检查
PYTHONPATH=src mypy src/epscheduler/

# 格式化
python3 -m ruff format src/ tests/

许可证

MIT License