Jobify : ordonnanceur asyncio sans brokers

Jobify : Un planificateur de tâches asynchrone sans infrastructure externe

Jobify vous permet de configurer un planificateur de tâches asynchrone en quelques lignes de code seulement. Aucun broker de messages nécessaire, avec SQLite par défaut pour la persistance. Les tâches sont sauvegardées sur disque et survivent aux redémarrages de l'application.

Exemple de base :

import asyncio
from jobify import Jobify

app = Jobify()


@app.task
async def send_notification(user_id: int) -> None:
    print(f"Notifier l'utilisateur {user_id}")


async def main() -> None:
    async with app:
        await send_notification.push(42)
        print("Tâche planifiée en arrière-plan")


asyncio.run(main())

La méthode push() met une tâche en file d'attente en arrière-plan sans bloquer. Jobify utilise loop.call_at pour un timing précis au lieu d'une boucle de scrutation, réduisant la charge inactive.

Défis de mise à l'échelle avec les tâches en arrière-plan

À mesure que le nombre de tâches augmente, les solutions simples ne suffisent plus. Le module sched de la bibliothèque standard n'est adapté qu'aux appels différés synchrones sans persistance ni nouvelles tentatives.

Le cron de Linux est excellent pour les scripts système mais ne s'intègre pas à votre application. Celery nécessite des brokers (RabbitMQ/Redis) et une infrastructure. APScheduler est plus proche mais limite la gestion du cycle de vie des tâches.

Jobify répond à cela grâce à :

• Cycle de vie complet des tâches : statuts, nouvelles tentatives, délais d'attente
• Middleware pour l'exécution et la planification (middleware externe)
• Injection de dépendances via lifespan
• Déduplication automatique des tâches
• Support natif d'asyncio

API de style FastAPI

Les développeurs FastAPI maîtriseront rapidement Jobify. Les concepts familiers incluent :

• Décorateurs de tâches @app.task
• Lifespan pour l'initialisation de l'état
• Pile de middleware
• Injection d'état via INJECT

Exemple avec attente du résultat :

import asyncio
from jobify import Jobify

app = Jobify()


@app.task
async def send_email(to: str, subject: str) -> None:
    print(f"Envoi d'email à {to} : {subject}")


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


asyncio.run(main())

L'objet job donne accès au statut, au résultat et à l'annulation.

Lifespan et injection de dépendances

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

Les politiques d'exécution sont définies sur le décorateur :

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

Middleware : Gestion de la planification et de l'exécution

Le middleware externe se déclenche lorsqu'une tâche est mise en file d'attente :

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"Planification de {context.job.id} avec déclencheur : {context.trigger}")
        return await call_next(context)


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

Le middleware régulier étend l'exécution (journalisation, métriques, nouvelles tentatives).

Architecture sans scrutation

Jobify évite while True: sleep(1): check_tasks(). Chaque tâche est immédiatement enregistrée via loop.call_at pour un temps exact.

Avantages :

• Charge inactive nulle
• Timing précis
• Ne bloque pas la boucle d'événements

Compromis : dépendance au temps monotone. Si l'heure système est réinitialisée, les tâches nécessitent un recalcul.

Fonctionnalités clés

• Persistance : SQLite par défaut, les tâches survivent aux redémarrages
• Déduplication : prévention automatique des doublons par ID
• Pile de middleware : intervention dans la planification et l'exécution
• API de type FastAPI : courbe d'apprentissage minimale pour les développeurs familiers
• Léger : pas de brokers, workers ou dépendances externes

— Editorial Team