Metadata-Version: 2.4
Name: fast_job
Version: 0.3.0
Summary: Reliable distributed scheduling for Python, powered by APScheduler and Redis Streams
Project-URL: Homepage, https://github.com/fastpkg/fast-job
Project-URL: Documentation, https://github.com/fastpkg/fast-job/tree/main/docs
Project-URL: Repository, https://github.com/fastpkg/fast-job
Project-URL: Issues, https://github.com/fastpkg/fast-job/issues
Author-email: Euraxluo <euraxluo@qq.com>
License: MIT
License-File: LICENSE
License-File: LICENSE.md
Keywords: apscheduler,distributed-scheduler,fastapi,job-queue,redis-streams,task-scheduler
Classifier: Development Status :: 4 - Beta
Classifier: Framework :: AsyncIO
Classifier: Framework :: FastAPI
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: System :: Distributed Computing
Requires-Python: >=3.10
Requires-Dist: apscheduler<4,>=3.7.0
Requires-Dist: pydantic<3,>=1.10
Requires-Dist: pytz>=2022.1
Requires-Dist: redis>=4.5.0
Requires-Dist: six>=1.16.0
Provides-Extra: dev
Requires-Dist: fastapi>=0.100.0; extra == 'dev'
Requires-Dist: httpx>=0.24; extra == 'dev'
Requires-Dist: loguru>=0.6.0; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.21; extra == 'dev'
Requires-Dist: pytest>=7.0; extra == 'dev'
Requires-Dist: uvicorn>=0.23.0; extra == 'dev'
Provides-Extra: fastapi
Requires-Dist: fastapi>=0.100.0; extra == 'fastapi'
Description-Content-Type: text/markdown

# fast_job

[中文文档](https://github.com/fastpkg/fast-job/blob/main/README.zh-CN.md) | English

[![PyPI version](https://img.shields.io/pypi/v/fast-job.svg)](https://pypi.org/project/fast-job/)
[![Python versions](https://img.shields.io/pypi/pyversions/fast-job.svg)](https://pypi.org/project/fast-job/)
[![License: MIT](https://img.shields.io/badge/license-MIT-green.svg)](https://github.com/fastpkg/fast-job/blob/main/LICENSE)

**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:

| Production question | fast_job answer |
| --- | --- |
| What identifies one due run? | A deterministic Fire ID derived from the APScheduler job ID and scheduled run time |
| What happens when several scheduler replicas see the same run? | Redis atomically keeps one Fire entry |
| What happens when every worker is busy? | The Fire remains durable in the Redis Stream |
| What happens when a worker exits before ACK? | Another worker reclaims the pending message |
| What happens when a task fails? | The Fire is retried with backoff or moved to the dead-letter Stream |
| How do web applications manage lifecycle? | `jobs.install(app)` binds startup, routes, and graceful shutdown |

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

## Install

```bash
pip install fast-job
```

For FastAPI integration:

```bash
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:

```bash
docker compose up -d redis
```

## 60-second quick start

```python
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`](https://github.com/fastpkg/fast-job/blob/main/example/quickstart.py).

## Built on APScheduler

`fast_job` deliberately reuses APScheduler instead of implementing another
trigger engine:

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

You can use the convenient task methods:

```python
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:

```python
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

| Object | Meaning |
| --- | --- |
| `Task` | A registered callable that a worker can execute |
| `ScheduledJob` | A trigger and its schedule state |
| `Fire` | One durable execution instance for one scheduled run time |

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

## Scheduling recipes

### Run once

```python
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

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

Duration strings and `timedelta` values are also accepted:

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

### Run on a cron schedule

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

Crontab expressions are supported:

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

### Enqueue immediately

```python
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:

```python
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`](https://github.com/fastpkg/fast-job/blob/main/example/fastapi_app.py).

## Scheduler and worker deployment

The same application module can run in three roles:

| Role | Scheduler | Worker |
| --- | ---: | ---: |
| `all` | yes | yes |
| `scheduler` | yes | no |
| `worker` | no | yes |

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

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

```bash
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`](https://github.com/fastpkg/fast-job/blob/main/example/service.py).

## Retry, timeout, and task context

```python
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`](https://github.com/fastpkg/fast-job/blob/main/example/retries_and_context.py).

## Schedule management

```python
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:

| Use case | Recommended API |
| --- | --- |
| Daily customer reports | `task.cron(..., timezone=...)` |
| Subscription renewal and reconciliation | `task.cron()` plus retry policy |
| Trial, reservation, or order expiration | `task.once()` |
| Periodic data synchronization | `task.every()` |
| Idempotent manual reruns | `task.enqueue(fire_id=...)` |
| SaaS scheduler/worker separation | `role="scheduler"` and `role="worker"` |
| FastAPI operational services | `jobs.install(app)` |

Detailed recipes are in [Use cases](https://github.com/fastpkg/fast-job/blob/main/docs/use-cases.md).

## Comparison with popular libraries

`fast_job` occupies a specific layer in the Python task ecosystem:

| Project | Relationship to fast_job |
| --- | --- |
| APScheduler | The trigger engine and scheduling foundation used by `fast_job` |
| Celery | A broader general-purpose task queue with many brokers, routing options, result backends, and workflow primitives |
| RQ | A mature Redis job queue with a simple API, job results, process workers, and scheduler components |
| Dramatiq | A focused distributed actor runtime with Redis/RabbitMQ brokers and mature worker middleware |
| Taskiq | An async-first typed task queue with pluggable brokers and framework integrations |
| Temporal | A durable workflow orchestration platform for long-running, stateful business processes |

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](https://github.com/fastpkg/fast-job/blob/main/docs/comparison.md).

## 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](https://github.com/fastpkg/fast-job/blob/main/docs/delivery-semantics.md).

## Configuration

Simple services can use flat arguments:

```python
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:

```python
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:

| Variable | Meaning |
| --- | --- |
| `FAST_JOB_REDIS_URL` | Redis or Redis Cluster URL |
| `FAST_JOB_NAMESPACE` | Logical namespace and Redis hash tag |
| `FAST_JOB_PREFIX` | Explicit Redis key prefix |
| `FAST_JOB_ROLE` | `all`, `scheduler`, or `worker` |
| `FAST_JOB_MAX_CONCURRENCY` | Worker concurrency |
| `FAST_JOB_MAX_ATTEMPTS` | Default attempt limit |
| `FAST_JOB_PENDING_CLAIM_TIMEOUT` | Pending reclaim idle threshold |
| `FAST_JOB_TASK_TIMEOUT` | Default task timeout |
| `FAST_JOB_CONSUMER_GROUP` | Redis Stream consumer group |
| `FAST_JOB_DISTRIBUTED` | Enable the durable distributed path |

## 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](https://github.com/fastpkg/fast-job/blob/main/docs/production.md).

## Documentation

- [Getting started](https://github.com/fastpkg/fast-job/blob/main/docs/getting-started.md)
- [Use cases](https://github.com/fastpkg/fast-job/blob/main/docs/use-cases.md)
- [Comparison](https://github.com/fastpkg/fast-job/blob/main/docs/comparison.md)
- [Production guide](https://github.com/fastpkg/fast-job/blob/main/docs/production.md)
- [Architecture](https://github.com/fastpkg/fast-job/blob/main/docs/architecture.md)
- [Delivery semantics](https://github.com/fastpkg/fast-job/blob/main/docs/delivery-semantics.md)
- [Redis protocol](https://github.com/fastpkg/fast-job/blob/main/docs/redis-protocol.md)
- [Runnable examples](https://github.com/fastpkg/fast-job/blob/main/example/README.md)

## Development and tests

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

Run the Redis Cluster acceptance suite:

```bash
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.
