Jobify: asyncio-Planer ohne Broker

Jobify: Ein asynchroner Task-Scheduler ohne externe Infrastruktur

Jobify ermöglicht es Ihnen, einen asynchronen Task-Scheduler mit nur wenigen Codezeilen einzurichten. Keine Message-Broker erforderlich, mit SQLite als Standard für Persistenz. Tasks werden auf der Festplatte gespeichert und überstehen Anwendungsneustarts.

Grundlegendes Beispiel:

import asyncio
from jobify import Jobify

app = Jobify()


@app.task
async def send_notification(user_id: int) -> None:
    print(f"Benutzer {user_id} benachrichtigen")


async def main() -> None:
    async with app:
        await send_notification.push(42)
        print("Job im Hintergrund geplant")


asyncio.run(main())

Die Methode push() reiht einen Task im Hintergrund ein, ohne zu blockieren. Jobify verwendet loop.call_at für präzise Zeitsteuerung anstelle einer Abfrageschleife, was die Leerlauflast reduziert.

Skalierungsherausforderungen mit Hintergrundtasks

Mit steigender Anzahl von Tasks stoßen einfache Lösungen an ihre Grenzen. Die Standardbibliothek sched eignet sich nur für synchrone verzögerte Aufrufe ohne Persistenz und Wiederholungsversuche.

Linux cron ist ideal für Systemskripte, integriert sich aber nicht in Ihre Anwendung. Celery erfordert Broker (RabbitMQ/Redis) und Infrastruktur. APScheduler kommt näher, schränkt aber die Task-Lebenszyklusverwaltung ein.

Jobify löst dies durch:

• Vollständigen Task-Lebenszyklus: Status, Wiederholungen, Timeouts
• Middleware für Ausführung und Planung (äußere Middleware)
• Dependency Injection über Lifespan
• Automatische Task-Deduplizierung
• Native asyncio-Unterstützung

FastAPI-ähnliche API

FastAPI-Entwickler werden Jobify schnell verstehen. Vertraute Konzepte umfassen:

• Task-Dekoratoren @app.task
• Lifespan für Zustandsinitialisierung
• Middleware-Stack
• Zustandsinjektion via INJECT

Beispiel mit Ergebniswarte:

import asyncio
from jobify import Jobify

app = Jobify()


@app.task
async def send_email(to: str, subject: str) -> None:
    print(f"E-Mail an {to} senden: {subject}")


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


asyncio.run(main())

Das Job-Objekt bietet Zugriff auf Status, Ergebnis und Abbruch.

Lifespan und Dependency Injection

Konfiguration über Lifespan:

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}")

Ausführungsrichtlinien werden im Dekorator festgelegt:

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

Middleware: Verwaltung von Planung und Ausführung

Äußere Middleware wird ausgelöst, wenn ein Task in die Warteschlange gestellt wird:

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"Planung von {context.job.id} mit Trigger: {context.trigger}")
        return await call_next(context)


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

Reguläre Middleware erweitern die Ausführung (Protokollierung, Metriken, Wiederholungen).

Abfragefreie Architektur

Jobify vermeidet while True: sleep(1): check_tasks(). Jeder Task wird sofort über loop.call_at für einen genauen Zeitpunkt registriert.

Vorteile:

• Keine Leerlauflast
• Präzise Zeitsteuerung
• Blockiert nicht den Event-Loop

Kompromiss: Abhängigkeit von monotoner Zeit. Bei Systemzeitrücksetzung müssen Tasks neu berechnet werden.

Hauptmerkmale

• Persistenz: SQLite standardmäßig, Tasks überstehen Neustarts
• Deduplizierung: automatische Verhinderung von Duplikaten durch ID
• Middleware-Stack: Eingriff in Planung und Ausführung
• FastAPI-ähnliche API: minimale Lernkurve für vertraute Entwickler
• Leichtgewichtig: keine Broker, Worker oder externen Abhängigkeiten

— Editorial Team