jobify 0.10.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 task scheduler for Python.

---

Documentation | Community Extensions

Jobify is a powerful framework for scheduling and managing background jobs in Python. It allows you to easily define and schedule jobs using an intuitive decorator-based API, similar to that of modern web frameworks such as FastAPI.

Unlike many other frameworks that use polling (continuous loops) to schedule tasks, Jobify uses the native timer mechanisms of asyncio for efficient and precise task scheduling.

Key Features

• Precision: No polling! Uses native asyncio timers for sub-millisecond accuracy and zero idle CPU usage.
• Scheduling: Run jobs immediately, with a delay, at a specified time, or using Cron expressions (second-level precision supported).
• Storage: Built-in SQLite ensures scheduled jobs persist through application restarts.
• Routing: Organize tasks with JobRouter, similar to FastAPI or Aiogram.
• Inject Context: Inject application state or custom dependencies directly into your tasks.
• Middlewares: Powerful interceptors for both job execution and the scheduling process.
• Exception Handlers: Hierarchical error management at the task, router, or global level.
• Lifespan Support: Manage startup and shutdown events, just like in FastAPI.
• Job Control: Full control over jobs — wait for completion, cancel tasks, or check results with ease.
• Concurrency: Supports asyncio, ThreadPoolExecutor, and ProcessPoolExecutor for efficient task handling.
• Many different adapters to the database.
• Distributed task queue. Soon.
• Many different serializers. Soon.

Installation

Install Jobify from PyPI:

pip install jobify

Quick Start

Here is an example of how to define and schedule a task:

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

from jobify import Jobify

UTC = ZoneInfo("UTC")

# 1. Initialize Jobify
app = Jobify(tz=UTC)


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


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


async def main() -> None:
    # 2. Run the Jobify application context
    async with app:
        # Run immediately in the background.
        job = await my_job.push("Alex")

        # Schedule a one-time job at a specific time.
        run_next_day = datetime.now(tz=UTC) + timedelta(days=1)
        job_at = await my_job.schedule("Connor").at(run_next_day)

        # Schedule a one-time job after a delay.
        job_delay = await my_job.schedule("Sara").delay(seconds=20)

        # Start a dynamic cron job.
        job_cron = await my_cron.schedule().cron(
            "* * * * *",
            job_id="dynamic_cron_id",
        )

        # Wait for specific jobs to complete
        await job_delay.wait()

        # You can also use `await app.wait_all()` to wait for all currently
        # running jobs. Note that infinite cron jobs will block this indefinitely.
        # await app.wait_all()


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

License

This project is licensed under the terms of the MIT license.