Skip to content

Jobify

Jobify is an asynchronous job scheduling framework for Python with event-driven timers, typed APIs, and production-focused execution controls.

  • Event-driven Precision

    No polling loops. Jobs are triggered via low-level asyncio timers.

    Why Jobify

  • Flexible Scheduling

    Run now, after delay, at timestamp, or via cron (including seconds).

    Open Scheduling

  • Built-in Persistence

    SQLite storage keeps scheduled jobs across restarts.

    Open Storage

  • Queue + Backpressure

    Control throughput with bounded queues, workers, and priorities.

    Open Queue Middleware

  • Router-style Organization

    Structure tasks with JobRouter in a FastAPI-like style.

    Open Router

  • Failure Handling

    Hierarchical exception handlers at global, router, and task levels.

    Open Exception Handlers

Comparison

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) ✅ ✅ ✅ ✅
Rich Typing Support ✅ ✅ ❌ ❌
Zero-config Persistence ✅ (SQLite default) ❌ (Needs Broker) ✅ ❌ (Needs Broker)
Broker-backend execution ❌ (soon) ✅ ❌ ✅

Why Jobify?

Jobify uses asyncio.loop.call_at instead of polling loops.

  1. Efficiency: No idle CPU usage when nothing is scheduled.
  2. Precision: Sub-millisecond timing without polling jitter.
  3. Native behavior: Works with OS event-notification primitives.

Precision vs Polling Trade-off

Event-driven scheduling is sensitive to significant system clock shifts. See System Time and Scheduling.

Quick Start

Installation

pip install jobify

Basic Usage

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
import asyncio
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo

from jobify import Jobify, Job

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: Job = await my_job.schedule("Connor").at(run_next_day)
        job_delay: Job = await my_job.schedule("Sara").delay(20)
        job_cron: Job = 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())