task-schedule 1.0.0

A simple and powerful task scheduling framework with web UI

Task Schedule

一个简单而强大的Python定时任务调度框架，支持装饰器定义任务、Web界面管理、数据库持久化存储。

功能特性

• 装饰器定义任务: 使用 @scheduled_task 装饰器轻松定义定时任务
• 灵活调度: 支持Cron表达式和固定间隔两种调度方式
• Web管理界面: 直观的Web界面查看任务状态和运行日志
• 实时日志: 支持查看任务的实时运行日志
• 数据库存储: 使用SQLite和SQLAlchemy持久化任务信息
• 日志管理: 使用loguru记录任务日志，自动归档和清理
• 易于部署: 支持打包发布到PyPI

安装

pip install task_schedule

快速开始

1. 创建任务文件

在 tasks 目录下创建任务文件：

# tasks/my_tasks.py
from task_schedule import scheduled_task

@scheduled_task(
    task_id="my_task",
    name="我的任务",
    description="这是一个示例任务",
    schedule_type="cron",
    schedule_value="* * * * *"
)
def my_task():
    """任务函数"""
    print("任务执行中...")
    return {"status": "success"}

2. 启动框架

# main.py
from task_schedule import main

if __name__ == "__main__":
    main(host="0.0.0.0", port=8080)

或者使用命令行：

python main.py

3. 访问Web界面

打开浏览器访问 http://localhost:8080

任务配置

Cron表达式

@scheduled_task(
    task_id="daily_task",
    name="每日任务",
    schedule_type="cron",
    schedule_value="0 2 * * *"  # 每天凌晨2点执行
)
def daily_task():
    pass

固定间隔

@scheduled_task(
    task_id="interval_task",
    name="间隔任务",
    schedule_type="interval",
    schedule_value="60"  # 每60秒执行
)
def interval_task():
    pass

项目结构

task_schedule/
├── task_schedule/          # 主包
│   ├── core/               # 核心模块
│   │   ├── scheduler.py    # 任务调度器
│   │   ├── decorator.py    # 任务装饰器
│   │   ├── scanner.py      # 任务扫描器
│   │   ├── database.py     # 数据库模型
│   │   └── logger.py       # 日志管理
│   ├── web/                # Web模块
│   │   ├── routes.py       # 路由定义
│   │   └── ...
│   └── templates/          # HTML模板
│       ├── index.html
│       ├── task_detail.html
│       ├── task_history.html
│       └── run_detail.html
├── tasks/                  # 任务目录
│   ├── __init__.py
│   └── example_tasks.py
├── logs/                   # 日志目录
├── pyproject.toml
└── README.md

API接口

• GET / - 首页，任务列表
• GET /task/<task_id> - 任务详情
• GET /task/<task_id>/history - 任务运行历史
• GET /task/<task_id>/run/<instance_id> - 运行记录详情
• GET /api/tasks - 获取所有任务
• GET /api/tasks/<task_id> - 获取任务详情
• GET /api/tasks/<task_id>/runs - 获取任务运行记录
• POST /api/tasks/<task_id>/trigger - 手动触发任务
• POST /api/tasks/<task_id>/pause - 暂停任务
• POST /api/tasks/<task_id>/resume - 恢复任务

打包发布

# 构建包
python -m build

# 发布到PyPI
twine upload dist/*

License

MIT