Metadata-Version: 2.4
Name: epscheduler
Version: 0.1.2
Summary: 可扩展的 Python 定时任务调度框架
Project-URL: Homepage, https://github.com/sidleo/epscheduler
Project-URL: Repository, https://github.com/sidleo/epscheduler.git
Requires-Python: >=3.11
Description-Content-Type: text/markdown
Requires-Dist: pydantic>=2.0
Requires-Dist: croniter>=2.0
Requires-Dist: sqlalchemy[asyncio]>=2.0
Requires-Dist: aiosqlite>=0.19
Provides-Extra: dev
Requires-Dist: pytest>=8.0; extra == "dev"
Requires-Dist: pytest-asyncio>=0.24; extra == "dev"
Requires-Dist: ruff>=0.3; extra == "dev"
Requires-Dist: mypy>=1.9; extra == "dev"

# epscheduler

[![PyPI version](https://badge.fury.io/py/epscheduler.svg)](https://pypi.org/project/epscheduler/)
[![Python](https://img.shields.io/pypi/pyversions/epscheduler.svg)](https://pypi.org/project/epscheduler/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)

可扩展的 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+ 语法，支持静态类型检查

## 安装

```bash
pip install epscheduler
```

**依赖要求**：
- Python 3.11+
- pydantic >= 2.0
- croniter >= 2.0
- sqlalchemy >= 2.0
- aiosqlite >= 0.19

## 快速上手

### 基础示例

```python
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 等：

```python
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 应用：

```python
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 工作流示例

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

### 条件分支节点

```python
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 存储作业状态，重启后恢复：

```python
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 核心方法

```python
# 添加作业
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 核心方法

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

### JobStore 核心方法

```python
# 添加作业
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
```

## 开发指南

```bash
# 克隆项目
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+ 语法
- 异步优先设计
- 模块化可扩展架构
