Jobify: planificador asyncio sin brokers

Jobify: Un Planificador de Tareas Asíncrono Sin Infraestructura Externa

Jobify te permite configurar un planificador de tareas asíncrono en solo unas líneas de código. No se necesitan intermediarios de mensajes, con SQLite como opción predeterminada para persistencia. Las tareas se guardan en disco y sobreviven a reinicios de la aplicación.

Ejemplo básico:

import asyncio
from jobify import Jobify

app = Jobify()


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


async def main() -> None:
    async with app:
        await send_notification.push(42)
        print("Tarea programada en segundo plano")


asyncio.run(main())

El método push() encola una tarea en segundo plano sin bloquear. Jobify usa loop.call_at para una programación precisa en lugar de un bucle de sondeo, reduciendo la carga inactiva.

Desafíos de Escalabilidad con Tareas en Segundo Plano

A medida que crece el número de tareas, las soluciones simples se quedan cortas. La biblioteca estándar sched solo es adecuada para llamadas retrasadas sincrónicas sin persistencia ni reintentos.

Linux cron es excelente para scripts del sistema pero no se integra con tu aplicación. Celery requiere intermediarios (RabbitMQ/Redis) e infraestructura. APScheduler está más cerca pero limita la gestión del ciclo de vida de las tareas.

Jobify aborda esto mediante:

• Ciclo de vida completo de tareas: estados, reintentos, tiempos de espera
• Middleware para ejecución y programación (middleware externo)
• Inyección de dependencias a través de lifespan
• Deduplicación automática de tareas
• Soporte nativo de asyncio

API al Estilo FastAPI

Los desarrolladores de FastAPI aprenderán Jobify rápidamente. Los conceptos familiares incluyen:

• Decoradores de tareas @app.task
• Lifespan para inicialización de estado
• Middleware en pila
• Inyección de estado vía INJECT

Ejemplo con espera de resultado:

import asyncio
from jobify import Jobify

app = Jobify()


@app.task
async def send_email(to: str, subject: str) -> None:
    print(f"Enviando correo a {to}: {subject}")


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


asyncio.run(main())

El objeto job proporciona acceso al estado, resultado y cancelación.

Lifespan e Inyección de Dependencias

Configuración a través de 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}")

Las políticas de ejecución se establecen en el decorador:

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

Middleware: Gestión de Programación y Ejecución

Middleware externo se activa cuando una tarea se encola:

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"Programando {context.job.id} con activador: {context.trigger}")
        return await call_next(context)


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

El middleware regular extiende la ejecución (registro, métricas, reintentos).

Arquitectura Sin Sondeo

Jobify evita while True: sleep(1): check_tasks(). Cada tarea se registra inmediatamente vía loop.call_at para un tiempo exacto.

Ventajas:

• Carga inactiva cero
• Programación precisa
• No bloquea el bucle de eventos

Contrapartida: dependencia del tiempo monótono. Si el tiempo del sistema se reinicia, las tareas requieren recálculo.

Características Clave

• Persistencia: SQLite por defecto, las tareas sobreviven a reinicios
• Deduplicación: prevención automática de duplicados por ID
• Pila de middleware: intervención en programación y ejecución
• API similar a FastAPI: curva de aprendizaje mínima para desarrolladores familiarizados
• Ligero: sin intermediarios, trabajadores o dependencias externas

— Editorial Team