xscheduler

xscheduler 是一个可嵌入 Python 服务的单进程任务调度模块，提供任务注册、优先级队列、同步/异步执行、cron、SQLAlchemy 持久化、进度事件，以及协作式暂停和取消。

它不依赖 FastAPI、WebSocket、SSE 或特定日志框架。宿主应用只需提供 SQLAlchemy sessionmaker，并自行决定如何转发 SchedulerEvent。

安装

首版按源码使用：

uv add /path/to/xscheduler
# 或
uv add 'xscheduler @ git+ssh://your-git/xscheduler.git@main'

需要 Python 3.12 或更高版本。

快速开始

import asyncio

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from xscheduler import (
    ExecutionContext,
    Scheduler,
    SqlAlchemyStore,
    create_schema,
    task,
)

engine = create_engine("sqlite:///tasks.db")
SessionLocal = sessionmaker(engine, expire_on_commit=False)
create_schema(engine)  # 仅建议开发/测试使用；生产环境使用 run_migrations(engine)


@task("hello", name="Say hello")
def hello(ctx: ExecutionContext, who: str) -> dict:
    ctx.emit(progress=50, message=f"Hello, {who}")
    return {"who": who}


async def main() -> None:
    scheduler = Scheduler(SqlAlchemyStore(SessionLocal))
    await scheduler.start()
    execution_id = scheduler.submit("hello", parameters={"who": "world"})

    while scheduler.get_execution(execution_id).status in {"pending", "running"}:
        await asyncio.sleep(0.05)

    print(scheduler.get_execution(execution_id))
    await scheduler.close()


asyncio.run(main())

任务注册与发现

可以使用默认注册器，也可以为测试或多套独立任务创建 TaskRegistry：

from xscheduler import TaskRegistry

registry = TaskRegistry()


@registry.task("reindex", triggerable=False, task_type="maintenance")
async def reindex(ctx, batch_size: int = 100): ...


registry.autodiscover(["myapp.tasks"])
scheduler = Scheduler(store, registry=registry)

autodiscover 会导入指定模块；如果目标是包，则会递归导入包下所有模块。任务参数在提交和创建 cron 时通过 Python 函数签名校验，不负责类型转换。

暂停与取消

控制是协作式的，Python 无法安全地强制终止运行中的线程。长任务必须在自然检查点主动检查：

def sync_task(ctx, items):
    for item in items:
        if not ctx.wait_if_paused() or ctx.is_cancelled():
            return
        process(item)


async def async_task(ctx, items):
    for item in items:
        if not await ctx.wait_if_paused_async() or ctx.is_cancelled():
            return
        await process(item)

scheduler.cancel(id) 可以直接取消排队任务；对运行中任务只设置取消信号。pause 和 resume 仅对当前进程内正在执行的任务有效。

cron 与时间

schedule = scheduler.create_schedule(
    "hello",
    name="Daily greeting",
    cron_expression="0 9 * * *",
    parameters={"who": "team"},
)

cron 默认按 UTC 解释，可通过 SchedulerConfig(timezone="Asia/Shanghai") 指定 IANA 时区。数据库中的时间统一保存为 UTC 的无时区值，以兼容 SQLite 和 MySQL。

数据库迁移

生产环境使用包内 Alembic revision：

from xscheduler import run_migrations

run_migrations(engine)

模块拥有以下表：

• xscheduler_task_definition
• xscheduler_task_execution
• xscheduler_task_schedule

任务代码从注册器消失时，定义会标记为 registered = false，历史执行不会被删除或改挂到其他任务。

事件转发

event_sink 可以是同步或异步 callable：

async def publish(event):
    await websocket_hub.broadcast(event)


scheduler = Scheduler(store, event_sink=publish)

事件包括 scheduler 启停、执行状态变化和任务主动发送的进度消息。事件处理器抛出的异常只会被记录，不会改变任务结果。

FastAPI lifespan

from contextlib import asynccontextmanager
from fastapi import FastAPI

scheduler = Scheduler(store)


@asynccontextmanager
async def lifespan(app: FastAPI):
    await scheduler.start()
    try:
        yield
    finally:
        await scheduler.close(cancel_running=True)


app = FastAPI(lifespan=lifespan)

FastAPI 只是宿主示例，不是 xscheduler 的依赖。HTTP 路由、认证和响应格式由应用自行实现。

部署边界

首版只支持单进程、单 scheduler 实例。不要让多个进程或容器同时消费同一套 xscheduler 表；分布式租约、主节点选举和强制终止不在首版范围内。

开发

uv sync
uv run ruff check .
uv run pytest