Metadata-Version: 2.4
Name: xscheduler
Version: 0.1.0
Summary: An embedded Python task scheduler with SQLAlchemy persistence
Requires-Python: >=3.12
Requires-Dist: alembic<2,>=1.13
Requires-Dist: croniter<7,>=6
Requires-Dist: sqlalchemy<3,>=2.0
Description-Content-Type: text/markdown

# xscheduler

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

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

## 安装

首版按源码使用：

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

需要 Python 3.12 或更高版本。

## 快速开始

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

```python
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 无法安全地强制终止运行中的线程。长任务必须在自然检查点主动检查：

```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 与时间

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

```python
from xscheduler import run_migrations

run_migrations(engine)
```

模块拥有以下表：

- `xscheduler_task_definition`
- `xscheduler_task_execution`
- `xscheduler_task_schedule`

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

## 事件转发

`event_sink` 可以是同步或异步 callable：

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


scheduler = Scheduler(store, event_sink=publish)
```

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

## FastAPI lifespan

```python
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 表；分布式租约、主节点选举和强制终止不在首版范围内。

## 开发

```bash
uv sync
uv run ruff check .
uv run pytest
```
