Jobify：无代理的 asyncio 调度器

Jobify：无需外部基础设施的异步任务调度器

Jobify 让你仅用几行代码就能搭建异步任务调度器。无需消息代理，默认使用 SQLite 进行持久化存储。任务保存到磁盘，应用重启后依然存在。

基础示例：

import asyncio
from jobify import Jobify

app = Jobify()


@app.task
async def send_notification(user_id: int) -> None:
    print(f"通知用户 {user_id}")


async def main() -> None:
    async with app:
        await send_notification.push(42)
        print("任务已在后台调度")


asyncio.run(main())

push() 方法在后台排队任务而不阻塞。Jobify 使用 loop.call_at 实现精确计时，而非轮询循环，从而降低空闲负载。

后台任务的扩展挑战

随着任务数量增长，简单方案难以应对。标准库的 sched 仅适用于无持久化和重试的同步延迟调用。

Linux cron 适合系统脚本，但无法与你的应用集成。Celery 需要代理（RabbitMQ/Redis）和基础设施。APScheduler 更接近，但限制了任务生命周期管理。

Jobify 通过以下方式解决：

• 完整的任务生命周期：状态、重试、超时
• 用于执行和调度的中间件（外层中间件）
• 通过生命周期进行依赖注入
• 自动任务去重
• 原生 asyncio 支持

FastAPI 风格的 API

FastAPI 开发者能快速上手 Jobify。熟悉的概念包括：

• 任务装饰器 @app.task
• 生命周期 用于状态初始化
• 中间件 栈
• 状态注入 通过 INJECT

等待结果的示例：

import asyncio
from jobify import Jobify

app = Jobify()


@app.task
async def send_email(to: str, subject: str) -> None:
    print(f"发送邮件给 {to}: {subject}")


async def main() -> None:
    async with app:
        job = await send_email.schedule(
            to="[email protected]",
            subject="欢迎！",
        ).delay(10)
        await job.wait()


asyncio.run(main())

任务对象提供状态、结果和取消的访问。

生命周期和依赖注入

通过生命周期配置：

import asyncio
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from jobify import INJECT, Jobify, State


@asynccontextmanager
async def lifespan(_: Jobify) -> AsyncIterator[dict[str, str]]:
    yield {"sender": "[email protected]"}


app = Jobify(lifespan=lifespan)


@app.task
def send_email(to: str, subject: str, state: State = INJECT) -> None:
    print(f"{state.sender} -> {to}: {subject}")

执行策略在装饰器上设置：

@app.task(retry=3, timeout=30)
async def sync_user(user_id: int) -> None:
    # ...

中间件：管理调度和执行

外层中间件 在任务排队时触发：

import asyncio
from jobify import Jobify, OuterContext
from jobify.middleware import BaseOuterMiddleware, CallNextOuter


class ScheduleLoggerMiddleware(BaseOuterMiddleware):
    async def __call__(
        self,
        call_next: CallNextOuter,
        context: OuterContext,
    ) -> asyncio.Handle:
        print(f"调度 {context.job.id}，触发器: {context.trigger}")
        return await call_next(context)


app = Jobify(outer_middleware=[ScheduleLoggerMiddleware()])

常规中间件扩展执行（日志记录、指标、重试）。

无轮询架构

Jobify 避免 while True: sleep(1): check_tasks()。每个任务通过 loop.call_at 立即注册到精确时间。

优势：

• 零空闲负载
• 精确计时
• 不阻塞事件循环

权衡：依赖单调时间。如果系统时间重置，任务需要重新计算。

主要特性

• 持久化：默认 SQLite，任务重启后保留
• 去重：通过 ID 自动防止重复
• 中间件栈：干预调度和执行
• 类 FastAPI 的 API：熟悉开发者学习曲线最小
• 轻量级：无需代理、工作进程或外部依赖

— Editorial Team