Jobify: asyncio plánovač bez brokerů

Jobify: asynchronní plánovač úloh bez externí infrastruktury

Jobify umožňuje nastavit asynchronní plánovač úloh během několika řádků kódu. Bez zprostředkovatelů zpráv, s SQLite jako výchozím řešením pro trvalost. Úlohy se ukládají na disk a přežijí restart aplikace.

Základní příklad:

import asyncio
from jobify import Jobify

app = Jobify()


@app.task
async def send_notification(user_id: int) -> None:
    print(f"Upozornit uživatele {user_id}")


async def main() -> None:
    async with app:
        await send_notification.push(42)
        print("Úloha naplánována na pozadí")


asyncio.run(main())

Metoda push() zařadí úlohu na pozadí bez blokování. Jobify používá loop.call_at pro přesné načasování místo cyklu s dotazováním, což snižuje zátěž při nečinnosti.

Problémy škálování úloh na pozadí

S rostoucím počtem úloh jednoduchá řešení přestávají stačit. sched ze standardní knihovny je vhodný pouze pro synchronní odložená volání bez trvalosti a opakování.

Linux cron je dobrý pro systémové skripty, ale neintegruje se s aplikací. Celery vyžaduje zprostředkovatele (RabbitMQ/Redis) a infrastrukturu. APScheduler je blíže, ale omezuje správu životního cyklu úloh.

Jobify to řeší pomocí:

• Úplný životní cyklus úlohy: stavy, opakování, časové limity
• Middleware pro provádění a plánování (outer middleware)
• Dependency injection přes lifespan
• Automatická deduplikace úloh
• Nativní podpora asyncio

API ve stylu FastAPI

Vývojáři FastAPI rychle zvládnou Jobify. Známé koncepty:

• Dekorátory úloh @app.task
• Lifespan pro inicializaci stavu
• Middleware zásobník
• State injection přes INJECT

Příklad s čekáním na výsledek:

import asyncio
from jobify import Jobify

app = Jobify()


@app.task
async def send_email(to: str, subject: str) -> None:
    print(f"Odesílání e-mailu na {to}: {subject}")


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


asyncio.run(main())

Job-objekt poskytuje přístup ke stavu, výsledku, zrušení.

Lifespan a dependency injection

Konfigurace přes 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}")

Politiky provádění se nastavují na dekorátoru:

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

Middleware: správa plánování a provádění

Outer middleware se spouští při zařazení úlohy:

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"Plánování {context.job.id} s triggerem: {context.trigger}")
        return await call_next(context)


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

Běžné middleware rozšiřují provádění (logování, metriky, opakování).

Architektura bez dotazování

Jobify se vzdává while True: sleep(1): check_tasks(). Každá úloha se okamžitě registruje přes loop.call_at na přesný čas.

Výhody:

• Nulová zátěž při nečinnosti
• Přesnost načasování
• Nenarušuje event loop

Kompromis: vázanost na monotónní čas. Při resetu systémového času vyžadují úlohy přepočet.

Co je důležité

• Trvalost: SQLite jako výchozí, úlohy přežijí restart
• Deduplikace: automatické zabránění duplicitám podle ID
• Middleware zásobník: zásah do plánování a provádění
• FastAPI-like API: minimální křivka učení pro zkušené vývojáře
• Lehkost: bez zprostředkovatelů, workerů, externích závislostí

— Editorial Team