jobify 0.12.1

A modern, lightweight and robust task manager with a user-friendly interface and a wide range of features, using the low-level asyncio API and more.

Jobify

Robust async task scheduler for Python.

Event-driven timing, typed APIs, persistence, routing, and queue-based backpressure.

Documentation • Quick Start • Community Extensions • Telegram

Contents

• Why Jobify
• Installation
• Quick Start
• Key Features
• When to Choose Jobify
• Roadmap
• License

Why Jobify

Most Python schedulers rely on polling loops. Jobify uses low-level asyncio timers (call_at) to schedule jobs directly.

• No idle polling CPU cost
• Sub-millisecond trigger precision
• Native async-first execution model

If your workload is bursty or downstream services are sensitive, use Queue Middleware to add bounded buffering, backpressure, and priority routing.

Installation

pip install jobify

Quick Start

import asyncio

from jobify import Jobify

app = Jobify()


@app.task
async def hello(name: str) -> None:
    print(f"Hello, {name}")


async def main() -> None:
    async with app:
        job = await hello.push("Alex")
        await job.wait()


if __name__ == "__main__":
    asyncio.run(main())

Extended example (cron, delay, absolute time)

import asyncio
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo

from jobify import Jobify

UTC = ZoneInfo("UTC")
app = Jobify(tz=UTC)


@app.task(cron="* * * * * * *")  # every second
async def my_cron() -> None:
    print("cron tick")


@app.task
def my_job(name: str) -> None:
    now = datetime.now(tz=UTC)
    print(f"Hello, {name}! at {now!r}")


async def main() -> None:
    async with app:
        await my_job.push("Alex")

        run_next_day = datetime.now(tz=UTC) + timedelta(days=1)
        job_at = await my_job.schedule("Connor").at(run_next_day)
        job_delay = await my_job.schedule("Sara").delay(seconds=20)
        job_cron = await my_cron.schedule().cron("* * * * *", job_id="dynamic_cron_id")

        await job_at.wait()
        await job_delay.wait()
        await job_cron.wait()


if __name__ == "__main__":
    asyncio.run(main())

Key Features

• Precision scheduling: event-driven timers, no polling loop.
• Flexible triggers: now, delay, timestamp, cron.
• Persistence: built-in SQLite storage for scheduled jobs.
• Routing: organize tasks with JobRouter.
• Context injection: inject state and dependencies into tasks.
• Middleware pipeline: execution and scheduling interceptors.
• Queue middleware: FIFO/LIFO/PriorityQueue with backpressure.
• Exception handlers: hierarchical error handling.
• Run modes: asyncio, thread pool, process pool.
• Community DB adapters.

Feature comparison (Jobify vs Taskiq/APScheduler/Celery)

Feature name	Jobify	Taskiq	APScheduler (v3)	Celery
Event-driven Scheduling	✅ (Low-level timer)	❌ (Polling/Loop)	❌ (Interval)	❌ (Polling/Loop)
Async Native (asyncio)	✅	✅	❌ (Sync mostly)	❌
Context Injection	✅	✅	❌	❌
FastAPI-style Routing	✅	❌	❌	❌
Middleware Support	✅	✅	❌ (Events only)	❌ (Signals)
Lifespan Support	✅	✅	❌	❌
Exception Handlers	✅ (Hierarchical)	❌	❌	❌
Job Cancellation	✅	❌	✅	✅
Cron Scheduling	✅ (Seconds level)	✅ (Minutes)	✅	✅
Misfire Policy	✅	❌	✅	❌
Run Modes (Thread/Process)	✅	✅	✅	✅
Zero-config Persistence	✅ (SQLite default)	❌ (Needs Broker)	✅	❌ (Needs Broker)
Broker-backend execution	❌ (roadmap)	✅	❌	✅

When to Choose Jobify

Use Jobify when:

• you need precise in-process scheduling without polling overhead
• you want typed, framework-like APIs for task registration and routing
• you need queue-based backpressure and priority controls in a single process

License

This project is licensed under the MIT license.