Jobify: asynchroniczny planista zadań bez zewnętrznej infrastruktury
Jobify pozwala skonfigurować asynchroniczny planista zadań w kilku linijkach kodu. Bez brokerów wiadomości, z domyślnym SQLite do trwałości. Zadania są zapisywane na dysku i przetrwają restart aplikacji.
Podstawowy przykład:
import asyncio
from jobify import Jobify
app = Jobify()
@app.task
async def send_notification(user_id: int) -> None:
print(f"Powiadom użytkownika {user_id}")
async def main() -> None:
async with app:
await send_notification.push(42)
print("Zadanie zaplanowane w tle")
asyncio.run(main())Metoda push() umieszcza zadanie w tle bez blokowania. Jobify używa loop.call_at do precyzyjnego czasowania zamiast cyklu pollingowego, zmniejszając obciążenie w stanie bezczynności.
Problemy skalowania zadań w tle
Wraz ze wzrostem liczby zadań proste rozwiązania przestają wystarczać. sched z biblioteki standardowej nadaje się tylko do synchronicznych odroczonych wywołań bez trwałości i ponownych prób.
Linuxowy cron jest dobry dla skryptów systemowych, ale nie integruje się z aplikacją. Celery wymaga brokerów (RabbitMQ/Redis) i infrastruktury. APScheduler jest bliższy, ale ogranicza zarządzanie cyklem życia zadań.
Jobify rozwiązuje to poprzez:
- Pełny cykl życia zadania: statusy, ponowne próby, limity czasu
- Middleware do wykonywania i planowania (outer middleware)
- Wstrzykiwanie zależności przez lifespan
- Automatyczną deduplikację zadań
- Natywne wsparcie asyncio
API w stylu FastAPI
Deweloperzy FastAPI szybko opanują Jobify. Znane koncepcje:
- Dekoratory zadań
@app.task - Lifespan do inicjalizacji stanu
- Middleware stos
- Wstrzykiwanie stanu przez
INJECT
Przykład z oczekiwaniem na wynik:
import asyncio
from jobify import Jobify
app = Jobify()
@app.task
async def send_email(to: str, subject: str) -> None:
print(f"Wysyłanie e-maila do {to}: {subject}")
async def main() -> None:
async with app:
job = await send_email.schedule(
to="[email protected]",
subject="Witamy!",
).delay(10)
await job.wait()
asyncio.run(main())Obiekt Job daje dostęp do statusu, wyniku, anulowania.
Lifespan i wstrzykiwanie zależności
Konfiguracja przez 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}")Polityki wykonania są określane na dekoratorze:
@app.task(retry=3, timeout=30)
async def sync_user(user_id: int) -> None:
# ...Middleware: zarządzanie planowaniem i wykonaniem
Outer middleware uruchamiają się przy umieszczaniu zadania:
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"Planowanie {context.job.id} z wyzwalaczem: {context.trigger}")
return await call_next(context)
app = Jobify(outer_middleware=[ScheduleLoggerMiddleware()])Zwykłe middleware rozszerzają wykonanie (logowanie, metryki, ponowne próby).
Architektura bez polling
Jobify rezygnuje z while True: sleep(1): check_tasks(). Każde zadanie jest natychmiast rejestrowane przez loop.call_at na dokładny czas.
Zalety:
- Zerowe obciążenie w bezczynności
- Precyzja czasowania
- Nie blokuje event loop
Kompromis: powiązanie z czasem monotonicznym. Przy zresetowaniu czasu systemowego zadania wymagają przeliczenia.
Co jest ważne
- Trwałość: SQLite domyślnie, zadania przetrwają po restartach
- Deduplikacja: automatyczne zapobieganie duplikatom po ID
- Middleware stos: ingerencja w planowanie i wykonanie
- FastAPI-like API: minimalna krzywa uczenia dla znanych deweloperów
- Lekkość: bez brokerów, workerów, zewnętrznych zależności
— Editorial Team
Brak komentarzy.