epscheduler 0.1.2

可扩展的 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
• sqlalchemy >= 2.0
• aiosqlite >= 0.19

快速上手

基础示例

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()

异步调度器

适用于 asyncio 应用，如 FastAPI、aiohttp 等：

import asyncio
from epscheduler import AsyncScheduler
from epscheduler.triggers import CronTrigger
from epscheduler.executors import AsyncExecutor

async def fetch_data():
    print("Fetching data...")

async def main():
    scheduler = AsyncScheduler(executor=AsyncExecutor(max_workers=10))
    scheduler.add_job(fetch_data, CronTrigger(cron_expression="0 9 * * *"))
    await scheduler.start()
    # 在后台运行，可以继续执行其他任务
    await asyncio.sleep(60)
    await scheduler.shutdown()

asyncio.run(main())

后台调度器

适用于 Web/GUI 应用：

from epscheduler import BackgroundScheduler
from epscheduler.triggers import IntervalTrigger
from epscheduler.jobstores import SQLiteJobStore

scheduler = BackgroundScheduler(
    jobstore=SQLiteJobStore("jobs.db")  # 持久化存储
)

def cleanup():
    print("Running cleanup...")

scheduler.add_job(cleanup, IntervalTrigger(minutes=30))
scheduler.start()

# 主线程继续处理 Web 请求
# 调度器在后台线程运行

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()

条件分支节点

from epscheduler.dag import DAG
from epscheduler.nodetypes import FuncNode, ConditionNode

dag = DAG()

# 定义节点
check_node = ConditionNode(
    id="check",
    condition=lambda ctx: ctx.get("value") > 10,
    true_branch="high",
    false_branch="low"
)
high_node = FuncNode(fn=lambda ctx: {"result": "high"}, id="high")
low_node = FuncNode(fn=lambda ctx: {"result": "low"}, id="low")

dag.add_node(check_node)
dag.add_node(high_node)
dag.add_node(low_node)
dag.add_edge("check", "high")
dag.add_edge("check", "low")

持久化存储

使用 SQLite 存储作业状态，重启后恢复：

from epscheduler import BlockingScheduler
from epscheduler.jobstores import SQLiteJobStore

scheduler = BlockingScheduler(
    jobstore=SQLiteJobStore("jobs.db")
)

# 添加作业后，即使程序重启，作业状态也会保留

核心组件

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	执行终端命令

API 文档

Scheduler 核心方法

# 添加作业
scheduler.add_job(func, trigger, job_id=None, params=None) -> str

# 添加 DAG 作业
scheduler.add_dag_job(dag, trigger, job_id=None) -> str

# 移除作业
scheduler.remove_job(job_id) -> bool

# 获取作业
scheduler.get_job(job_id) -> Job | None

# 获取所有作业
scheduler.get_jobs() -> list[Job]

# 启动/停止
scheduler.start()
scheduler.shutdown(wait=True)

# AsyncScheduler 异步版本
await scheduler.start()
await scheduler.shutdown()

Trigger 核心方法

# 获取下次触发时间
trigger.next_fire_time(last_fire_time: datetime | None) -> datetime | None

JobStore 核心方法

# 添加作业
await jobstore.add(job: Job) -> str

# 获取作业
await jobstore.get(job_id: str) -> Job | None

# 获取所有作业
await jobstore.list_all() -> list[Job]

# 获取到期作业
await jobstore.get_due_jobs(now: datetime) -> list[Job]

# 更新/移除作业
await jobstore.update(job: Job) -> bool
await jobstore.remove(job_id: str) -> bool

开发指南

# 克隆项目
git clone https://github.com/sidleo/epscheduler.git

# 安装开发依赖
pip install -e ".[dev]"

# 运行测试
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/

贡献指南

欢迎贡献代码、报告问题或提出建议！

1. Fork 项目
2. 创建特性分支 (git checkout -b feature/amazing-feature)
3. 提交更改 (git commit -m 'Add amazing feature')
4. 推送到分支 (git push origin feature/amazing-feature)
5. 创建 Pull Request

请确保代码通过测试和类型检查。

许可证

MIT License

致谢

本项目灵感来源于 APScheduler，采用现代化架构设计：

• 使用 Pydantic 进行数据验证
• 使用 Python 3.11+ 语法
• 异步优先设计
• 模块化可扩展架构