fast_job

中文文档 | English

Reliable distributed scheduling for Python, powered by APScheduler and Redis Streams.

fast_job keeps APScheduler where it is strongest: trigger calculation, time zones, cron expressions, intervals, one-off runs, misfire handling, and schedule state. It adds the runtime pieces an application needs when scheduled work must survive process restarts and run on distributed workers:

• one durable Fire for each APScheduler scheduled_run_time
• Redis Stream delivery with at-least-once semantics
• worker concurrency, retry, timeout, pending recovery, and dead letters
• scheduler-only, worker-only, or combined deployment roles
• a small instance-scoped API for Python and FastAPI applications

The result is not a replacement for APScheduler. It is a distributed execution layer built on top of APScheduler.

Why fast_job

APScheduler answers when should this job run? A production service also needs clear answers to the questions that follow:

Use fast_job when scheduled execution is part of your product, not just a single-process utility.

Install

pip install fast-job

For FastAPI integration:

pip install "fast-job[fastapi]"

The package requires Python 3.10+, APScheduler 3.x, and Redis 6.2+. Redis 6.2 is required for pending-message recovery through XAUTOCLAIM.

Start a local Redis instance from this repository:

docker compose up -d redis

60-second quick start

import asyncio
from datetime import datetime, timedelta, timezone

from fast_job import FastJob, current_task

jobs = FastJob(
    redis="redis://127.0.0.1:6379/0",
    namespace="reports",
    max_concurrency=4,
)


@jobs.task(id="send_report", description="Generate and send a customer report")
def send_report(customer_id: str):
    context = current_task()
    print(
        {
            "customer_id": customer_id,
            "fire_id": context.fire_id if context else None,
        }
    )


send_report.once(
    id="welcome-report",
    at=datetime.now(timezone.utc) + timedelta(seconds=3),
    args=["customer-42"],
    replace_existing=True,
)


async def main():
    async with jobs:
        await asyncio.sleep(6)


asyncio.run(main())

The application registers a callable Task, declares a one-off APScheduler schedule, starts the scheduler and worker, creates one durable Fire when the run is due, and shuts down gracefully.

A runnable version is available at example/quickstart.py.

Built on APScheduler

fast_job deliberately reuses APScheduler instead of implementing another trigger engine:

APScheduler trigger
        |
        v
scheduled_run_time
        |
        v
deterministic Fire ID -> Redis Stream -> fast_job worker -> user task

You can use the convenient task methods:

send_report.once(at=run_at)
send_report.every(minutes=10)
send_report.cron(hour=8, minute=0, timezone="Asia/Shanghai")

Or pass any APScheduler 3.x trigger directly:

from apscheduler.triggers.calendarinterval import CalendarIntervalTrigger

jobs.schedule(
    send_report,
    trigger=CalendarIntervalTrigger(
        months=1,
        hour=9,
        timezone="Asia/Shanghai",
    ),
    id="monthly-report",
    args=["customer-42"],
    replace_existing=True,
)

APScheduler remains responsible for schedule calculation. fast_job owns the durable Fire delivery path.

Three concepts

This separation keeps schedule management distinct from task execution and from the identity of one delivery attempt.

Scheduling recipes

Run once

send_report.once(
    id="trial-expiration-reminder",
    at=datetime.now(timezone.utc) + timedelta(days=7),
    args=["customer-42"],
    replace_existing=True,
)

Run at an interval

send_report.every(
    id="refresh-report",
    minutes=15,
    args=["customer-42"],
    replace_existing=True,
)

Duration strings and timedelta values are also accepted:

send_report.every("30m", args=["customer-42"])
send_report.every(timedelta(hours=2), args=["customer-42"])

Run on a cron schedule

send_report.cron(
    id="daily-report",
    hour=8,
    minute=0,
    timezone="Asia/Shanghai",
    args=["customer-42"],
    replace_existing=True,
)

Crontab expressions are supported:

send_report.cron(
    "0 8 * * 1-5",
    id="weekday-report",
    timezone="Asia/Shanghai",
    args=["customer-42"],
    replace_existing=True,
)

Enqueue immediately

fire = send_report.enqueue(
    "customer-42",
    fire_id="report:customer-42:2026-08-04",
)

print(fire.fire_id, fire.created, fire.message_id)

Providing the same fire_id again returns the existing Fire reference instead of appending another Stream entry. FireRef is awaitable for async API convenience, but awaiting it returns the reference; it does not wait for task completion.

FastAPI

Register tasks and schedules before installing the integration:

from fastapi import FastAPI
from fast_job import FastJob

app = FastAPI()

jobs = FastJob(
    redis="redis://127.0.0.1:6379/0",
    namespace="api",
    max_concurrency=8,
)


@jobs.task
async def rebuild_search_index(tenant_id: str):
    return {"tenant_id": tenant_id, "rebuilt": True}


rebuild_search_index.cron(
    id="nightly-index",
    hour=2,
    minute=0,
    timezone="UTC",
    args=["default-tenant"],
    replace_existing=True,
)

jobs.install(
    app,
    api_prefix="/jobs",
    graceful_timeout=30,
)

install() binds the FastJob lifecycle to FastAPI and mounts health, schedule query, pause, resume, remove, and registered task routes. Put these routes behind your application's authentication and authorization layer before exposing them outside a trusted network.

See example/fastapi_app.py.

Scheduler and worker deployment

The same application module can run in three roles:

jobs = FastJob(
    redis=REDIS_URL,
    namespace="billing",
    role="worker",
    max_concurrency=16,
)

Environment-based deployment keeps one code artifact for every role:

FAST_JOB_REDIS_URL=redis://redis:6379/0 \
FAST_JOB_NAMESPACE=billing \
FAST_JOB_ROLE=scheduler \
python -m example.service

FAST_JOB_REDIS_URL=redis://redis:6379/0 \
FAST_JOB_NAMESPACE=billing \
FAST_JOB_ROLE=worker \
FAST_JOB_MAX_CONCURRENCY=16 \
python -m example.service

Every worker process must load the same task registrations as the scheduler process. See example/service.py.

Retry, timeout, and task context

from fast_job import FastJob, PermanentError, Retry, current_task

jobs = FastJob(
    redis=REDIS_URL,
    namespace="billing",
    max_attempts=5,
    retry_backoff=(1, 5, 30, 120),
    task_timeout=60,
)


@jobs.task(
    id="capture-payment",
    max_attempts=4,
    retry_backoff=(2, 10, 30),
    timeout=20,
)
async def capture_payment(payment_id: str):
    context = current_task()

    if payment_id.startswith("invalid:"):
        raise PermanentError("invalid payment")

    if context and context.attempt < 2:
        raise Retry(delay=3)

    return {"payment_id": payment_id, "captured": True}

TaskContext exposes:

• job_id
• task_id
• fire_id
• scheduled_at
• stream_message_id
• worker_id
• worker_generation
• attempt
• metadata

Use fire_id as the idempotency key for external side effects.

See example/retries_and_context.py.

Schedule management

job = jobs.get_schedule("daily-report")

if job is not None:
    job.pause()
    job.modify(hour=9, timezone="Asia/Shanghai")
    job.resume()

jobs.remove("daily-report")

Available high-level operations include:

• get_schedule() and get_schedules()
• pause(), resume(), and remove()
• ScheduledJob.modify()
• access to the underlying trigger and next run time

Product use cases

fast_job works well when the scheduled run itself is a product event:

Detailed recipes are in Use cases.

Comparison with popular libraries

fast_job occupies a specific layer in the Python task ecosystem:

The main fast_job distinction is the path from an APScheduler scheduled_run_time to one deterministic, durable Fire. It is intentionally narrower than a general task queue and much lighter than a workflow engine.

Read the version-aligned comparison and official references in Comparison.

Delivery guarantee

fast_job provides at-least-once delivery, not exactly-once execution.

A worker can complete an external side effect and exit before Redis receives the ACK. The pending Fire may then be reclaimed and executed again. Tasks that change external state must therefore be idempotent.

The runtime makes Fire creation idempotent. It cannot make an arbitrary database write, HTTP request, email send, and Redis ACK one atomic transaction.

Read Delivery semantics.

Configuration

Simple services can use flat arguments:

jobs = FastJob(
    redis=REDIS_URL,
    namespace="reports",
    role="all",
    max_concurrency=8,
    max_attempts=5,
    retry_backoff="exponential",
    task_timeout=60,
    pending_claim_timeout=300,
)

Larger services can group worker and retry settings:

from fast_job import FastJob, RetryPolicy, WorkerConfig

jobs = FastJob(
    redis=REDIS_URL,
    namespace="reports",
    worker=WorkerConfig(
        concurrency=8,
        pending_claim_timeout=300,
        task_timeout=60,
        monitor_event_loop_lag=True,
    ),
    retry=RetryPolicy(
        max_attempts=5,
        backoff="exponential",
        min_delay=1,
        max_delay=300,
    ),
)

FastJob.from_env() reads:

Production checklist

• Use Redis persistence, backups, and noeviction for protocol keys.
• Keep a stable namespace across scheduler and worker deployments.
• Load identical task IDs in every process that can execute Fires.
• Make external side effects idempotent with TaskContext.fire_id.
• Set task_timeout, pending_claim_timeout, and graceful shutdown values from measured task durations.
• Monitor ready backlog, pending messages, retries, dead letters, task failures, and event-loop lag.
• Protect FastAPI management routes with authentication and authorization.
• Test Redis restart, worker termination, scheduler restart, and rolling deployment behavior before production rollout.

Read Production guide.

Documentation

• Getting started
• Use cases
• Comparison
• Production guide
• Architecture
• Delivery semantics
• Redis protocol
• Runnable examples

Development and tests

uv sync
docker compose up -d redis
uv run pytest -q

Run the Redis Cluster acceptance suite:

docker compose --profile cluster up -d
FAST_JOB_REDIS_CLUSTER_URL=redis://127.0.0.1:7100 \
  uv run pytest tests/test_cluster_multi_instance.py -q

The test suite covers deterministic Fire creation, multiple scheduler and worker processes, capacity backpressure, pending recovery, retries, dead letters, task timeouts, Redis Cluster routing, process termination, and application-instance recovery.

Contributions should include tests for behavioral changes and preserve the documented delivery contract.