Jobify: 브로커 없는 asyncio 스케줄러

Jobify: 외부 인프라 없이 비동기 작업 스케줄러 구축하기

Jobify를 사용하면 몇 줄의 코드만으로 비동기 작업 스케줄러를 설정할 수 있습니다. 메시지 브로커가 필요 없으며, 기본적으로 SQLite를 사용해 지속성을 보장합니다. 작업은 디스크에 저장되어 애플리케이션 재시작 후에도 유지됩니다.

기본 예시:

import asyncio
from jobify import Jobify

app = Jobify()


@app.task
async def send_notification(user_id: int) -> None:
    print(f"사용자 {user_id}에게 알림 전송")


async def main() -> None:
    async with app:
        await send_notification.push(42)
        print("백그라운드에서 작업이 예약되었습니다")


asyncio.run(main())

push() 메서드는 블로킹 없이 백그라운드에서 작업을 큐에 추가합니다. Jobify는 폴링 루프 대신 loop.call_at을 사용해 정확한 타이밍을 구현하여 유휴 부하를 줄입니다.

백그라운드 작업의 확장성 문제

작업 수가 증가함에 따라 단순한 해결책은 한계를 보입니다. 표준 라이브러리의 sched는 지속성과 재시도 기능 없이 동기 지연 호출에만 적합합니다.

Linux cron은 시스템 스크립트에는 훌륭하지만 애플리케이션과 통합되지 않습니다. Celery는 브로커(RabbitMQ/Redis)와 인프라가 필요합니다. APScheduler는 더 가깝지만 작업 생명주기 관리에 제한이 있습니다.

Jobify는 다음을 통해 이 문제를 해결합니다:

• 완전한 작업 생명주기: 상태, 재시도, 타임아웃
• 실행 및 스케줄링을 위한 미들웨어(외부 미들웨어)
• 수명 주기를 통한 의존성 주입
• 자동 작업 중복 제거
• 네이티브 asyncio 지원

FastAPI 스타일 API

FastAPI 개발자는 Jobify를 빠르게 익힐 수 있습니다. 익숙한 개념은 다음과 같습니다:

• 작업 데코레이터 @app.task
• 상태 초기화를 위한 수명 주기
• 미들웨어 스택
• INJECT를 통한 상태 주입

결과 대기 예시:

import asyncio
from jobify import Jobify

app = Jobify()


@app.task
async def send_email(to: str, subject: str) -> None:
    print(f"{to}에게 이메일 전송: {subject}")


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


asyncio.run(main())

작업 객체는 상태, 결과, 취소에 대한 접근을 제공합니다.

수명 주기와 의존성 주입

수명 주기를 통한 구성:

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

실행 정책은 데코레이터에서 설정됩니다:

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

미들웨어: 스케줄링과 실행 관리

외부 미들웨어는 작업이 큐에 추가될 때 트리거됩니다:

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"{context.job.id} 작업 스케줄링: 트리거 - {context.trigger}")
        return await call_next(context)


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

일반 미들웨어는 실행을 확장합니다(로깅, 메트릭, 재시도).

폴링 없는 아키텍처

Jobify는 while True: sleep(1): check_tasks()를 피합니다. 각 작업은 정확한 시간에 loop.call_at을 통해 즉시 등록됩니다.

장점:

• 유휴 부하 제로
• 정밀한 타이밍
• 이벤트 루프를 블로킹하지 않음

단점: 단조 시간에 의존합니다. 시스템 시간이 재설정되면 작업 재계산이 필요합니다.

주요 기능

• 지속성: 기본 SQLite, 작업 재시작 후 유지
• 중복 제거: ID별 자동 중복 방지
• 미들웨어 스택: 스케줄링 및 실행 개입
• FastAPI 스타일 API: 익숙한 개발자를 위한 최소 학습 곡선
• 경량: 브로커, 워커, 외부 의존성 없음

— Editorial Team