Metadata-Version: 2.5
Name: taskscheduler-client
Version: 0.9.0
Summary: Async Python SDK for TaskScheduler — durable distributed background jobs on PostgreSQL + RabbitMQ
Project-URL: Homepage, https://github.com/4aK-Boris/TaskScheduler
Project-URL: Documentation, https://github.com/4aK-Boris/TaskScheduler/blob/master/clients/python/README.md
Author: TaskScheduler
License: Apache-2.0
Keywords: asyncio,background,jobs,postgresql,rabbitmq,scheduler
Classifier: Development Status :: 4 - Beta
Classifier: Framework :: AsyncIO
Classifier: Intended Audience :: Developers
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: Topic :: System :: Distributed Computing
Requires-Python: >=3.10
Requires-Dist: aio-pika>=9.5
Requires-Dist: croniter>=6.0
Requires-Dist: psycopg[binary,pool]>=3.2
Provides-Extra: dev
Requires-Dist: mypy>=1.14; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.25; extra == 'dev'
Requires-Dist: pytest>=8.3; extra == 'dev'
Requires-Dist: ruff>=0.9; extra == 'dev'
Description-Content-Type: text/markdown

# TaskScheduler — Python SDK

Async Python client for [TaskScheduler](https://github.com/4aK-Boris/TaskScheduler). A Python service becomes a
first-class participant in the same job system as your Kotlin services: it enqueues work,
runs handlers, and shows up on the same dashboard with the same retry, cancellation and
progress semantics.

```python
from dataclasses import dataclass
from taskscheduler import Scheduler, SchedulerConfig, job_type

@job_type
@dataclass
class SendInvoice:
    order_id: int

async with Scheduler(SchedulerConfig(dsn="postgresql://scheduler:scheduler@localhost/scheduler")) as s:
    await s.enqueue(SendInvoice(order_id=42))
```

## How it fits together

The SDK speaks the same wire protocol as the Kotlin client — it is not a proxy in front of
it. PostgreSQL holds all job state; RabbitMQ carries nothing but a 16-byte job id as a
delivery hint.

```
┌──────────────────────────┐        ┌──────────────────────────┐
│  scheduler-infra (Kotlin) │        │  your Python service      │
│  • owns the schema        │        │  • Scheduler.enqueue()    │
│  • outbox → RabbitMQ      │  PG +  │  • WorkerPool runs jobs   │
│  • recurring cron         │ Rabbit │  • heartbeats its lease   │
│  • orphan recovery        │◄──────►│                           │
│  • dashboard :8080        │        │                           │
└──────────────────────────┘        └──────────────────────────┘
             └────────► PostgreSQL ◄────────┘
             └────────► RabbitMQ   ◄────────┘
```

**`scheduler-infra` is required.** It owns the Flyway migrations and the background loops
that neither client implements: publishing the outbox to RabbitMQ, firing cron definitions,
recovering jobs whose worker died, and retention. This SDK verifies the schema version at
startup and refuses to run against a database that is too old.

## Install

```bash
pip install taskscheduler-client        # or: uv pip install taskscheduler-client
```

Requires Python 3.10+, and a RabbitMQ with the `rabbitmq_delayed_message_exchange` plugin
enabled (the same requirement the Kotlin side has — it is how delays and retry backoff work).

**Versioning.** The client shares one version with the rest of the project:
`taskscheduler-client 0.9.0` is the client for `scheduler-infra 0.9.0`, and the two are
released together. Run a client older than your infra and it may not know about a newer
column; newer, and it fails fast on the schema check. CI enforces that
`pyproject.toml`, `taskscheduler.__version__` and Gradle's `schedulerVersion` agree.

**On Windows**, select the other event loop before starting anything — psycopg cannot run on
the `ProactorEventLoop` that asyncio uses by default there:

```python
if sys.platform == "win32":
    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
```

The SDK raises a `ConfigurationError` naming this if you forget, rather than hanging on a
connection-pool timeout.

## Producing jobs

A payload is a plain dataclass. `@job_type` gives it a stable name that is stored in
`job.payload_type`:

```python
@job_type                       # -> "billing.jobs.SendInvoice"
@dataclass
class SendInvoice:
    order_id: int
    dry_run: bool = False

@job_type("billing.SendInvoice.v2")   # pin it before renaming or moving the class
@dataclass
class SendInvoiceV2:
    ...
```

```python
# now
await scheduler.enqueue(SendInvoice(order_id=42), queue="billing", priority=7)

# at a moment
await scheduler.schedule_at(SendReminder(order_id=42), at=datetime(2026, 6, 1, 10, tzinfo=timezone.utc))
await scheduler.schedule_in(SendReminder(order_id=42), delay=timedelta(days=1))

# at most one active job per key
await scheduler.enqueue_once(f"sync-user-{user_id}", SyncUser(user_id))

# strictly in order
await scheduler.chain(ExtractData(), TransformData(), LoadData())

# after a fan-out finishes
a = await scheduler.enqueue(LoadProductCache())
b = await scheduler.enqueue(LoadUserCache())
await scheduler.enqueue_after(StartPricingEngine(), wait_for=[a, b])

# on a cron
await scheduler.recurring("nightly-rollup", "0 3 * * *", NightlyRollup(), timezone_name="Europe/Berlin")
```

Every enqueue writes the job row and its outbox row in **one transaction**, so a job
becomes visible only if your surrounding business transaction commits.

## Discarding overdue jobs

```python
await scheduler.schedule_at(
    RefreshPrices(), at=planned_time, max_start_delay_seconds=30,
)
await scheduler.recurring(
    "refresh-prices", "0 * * * * *", RefreshPrices(), max_start_delay_seconds=30,
)
```

The latest permitted start is the scheduled time plus `max_start_delay_seconds`. An
immediate enqueue counts from enqueue time; recurring jobs count from the original cron
slot, including downtime. Workers cancel overdue pickups before invoking the handler and
resolve their dependencies normally. Retries and dependency waits retain the original
deadline. A handler that started on time can finish after it; execution timeouts are separate.

`None` (the default) leaves waiting unlimited. Values must be finite, non-negative, and
represent a whole number of milliseconds; zero rejects any lateness. Cancellation is
checked when a worker picks up the job, not by a timer while it is waiting without a
worker. This client requires schema V10; upgrade scheduler-infra and all workers before
enabling deadlines. Older jobs have no deadline.

## Consuming jobs

```python
from taskscheduler import HandlerRegistry, JobContext, RabbitConfig, WorkerConfig, WorkerPool

registry = HandlerRegistry()

@registry.handler(SendInvoice, retry_policy=ExponentialBackoff(max_attempts=5))
async def send_invoice(ctx: JobContext, job: SendInvoice) -> None:
    await billing.send(job.order_id, idempotency_key=str(ctx.job_id))

worker = WorkerPool(
    scheduler_config=SchedulerConfig(dsn=DSN, node_id="billing-1"),
    worker_config=WorkerConfig(node_id="billing-1").queue("billing", concurrency=8),
    rabbit_config=RabbitConfig(url=AMQP_URL, queues=["billing"]),
    registry=registry,
)

async with worker:
    await asyncio.Event().wait()      # run until the process is stopped
```

Class-based handlers work too, when you need `on_final_failure` or dependency injection:

```python
class SendInvoiceHandler(JobHandler[SendInvoice]):
    payload = SendInvoice
    retry_policy = ExponentialBackoff(max_attempts=5)

    def __init__(self, billing: BillingClient) -> None:
        self._billing = billing

    async def execute(self, ctx: JobContext, job: SendInvoice) -> None:
        await self._billing.send(job.order_id)

    async def on_final_failure(self, ctx: JobContext, job: SendInvoice, error: BaseException) -> None:
        await alerts.page(f"invoice {job.order_id} failed permanently")

registry.register(SendInvoiceHandler(billing))
```

### Progress and cancellation

```python
@registry.handler(ReindexCatalog)
async def reindex(ctx: JobContext, job: ReindexCatalog) -> None:
    bar = ctx.progress_bar(len(job.product_ids))
    for product_id in job.product_ids:
        if await ctx.is_cancellation_requested():
            raise JobCancellationError()      # ends as CANCELLED, not FAILED
        await index(product_id)
        await bar.succeeded()
```

Progress writes are throttled to one per second, so calling them in a tight loop is fine.
A cancelled job that ignores the flag is cancelled outright after
`WorkerConfig.cancel_grace_seconds`.

### Failing

| You raise | Outcome |
|---|---|
| any exception | retried per the policy, then FAILED |
| `NonRetriableError` | FAILED immediately, remaining attempts skipped |
| `JobCancellationError` | CANCELLED, no retry, no `on_final_failure` |
| nothing | SUCCEEDED |

## Configuration

```python
SchedulerConfig(
    dsn="postgresql://user:pass@host:5432/scheduler",   # or dsn_from_jdbc(...)
    node_id="billing-1",
    default_queue="default",
    default_max_attempts=3,
    default_timeout_seconds=300,
    default_retry_policy=ExponentialBackoff(max_attempts=3),
)

WorkerConfig(
    node_id="billing-1",
    node_tags=["eu-west"],
    heartbeat_interval_seconds=30,     # must be <= lock_duration / 3
    lock_duration_seconds=90,
    shutdown_timeout_seconds=30,
).queue("billing", concurrency=8, prefetch=8)

RabbitConfig(url="amqp://scheduler:scheduler@localhost:5672/", queues=["billing"])
```

Sharing `POSTGRES_URL` with the Kotlin services:

```python
dsn = dsn_from_jdbc(os.environ["POSTGRES_URL"], os.environ["POSTGRES_USER"], os.environ["POSTGRES_PASSWORD"])
```

## Running Python and Kotlin side by side

Both clients read and write the same tables, and the dashboard shows their jobs together.
What does **not** cross the language boundary is the payload itself: a `payload_type` is a
Python class name here and a Kotlin FQN there, and a worker that picks up a type it does
not know marks the job FAILED rather than passing it on.

**So give each language its own queues.** Point Python handlers at `python`, `ml`, or
whichever names you like, and keep Kotlin workers on theirs:

```python
RabbitConfig(url=AMQP_URL, queues=["ml"])
WorkerConfig(node_id="ml-1").queue("ml", concurrency=4)
```

```kotlin
schedulerRabbitModule { queues = listOf("default", "ml") }   // infra must declare every queue
schedulerWorkerModule { queue("default", concurrency = 8) }  // but only consumes its own
```

The infra process needs every queue name in its `schedulerRabbitModule.queues` so the
topology exists; it does not need to consume them.

If you do want the two languages to run each other's jobs, pin `@job_type("<kotlin FQN>")`
and keep the JSON field names identical to the Kotlin data class — this SDK will encode and
decode it, but nothing checks that the two definitions still agree.

## Guarantees

**At-least-once.** A job can run twice — a lease expiring during a long GC pause, a broker
redelivery, a network partition. `ctx.job_id` is stable across every attempt, so use it as
the idempotency key for anything with side effects:

```python
await payments.charge(order_id, idempotency_key=str(ctx.job_id))
```

**Leases, not locks.** A claimed job is held by `locked_until`, extended every
`heartbeat_interval_seconds`. If this process dies, the lease lapses and infra re-enqueues
the job — that is the recovery path, and it is why `heartbeat_interval` must stay at or
below a third of `lock_duration`.

**Schema evolution.** Adding a field with a default is safe. Removing one is safe — unknown
keys are ignored on decode. Renaming or retyping is not: version the payload
(`SendInvoiceV2`) and keep both handlers until the old jobs have drained. A payload that
cannot be decoded fails terminally without burning retries, since the stored bytes will
never change.

## Development

```bash
uv venv && uv pip install -e ".[dev]"
pytest tests/unit                                    # no infrastructure needed
ruff check src tests examples scripts && mypy src
```

Integration tests need a database with the migrations applied and a broker with the
delayed-message plugin. The whole suite runs in CI on every change under `clients/python`
(`.github/workflows/python-client.yml`); locally, bring the two up yourself. The commands
below assume a checkout of the repository, run from `clients/python`:

```bash
docker run -d --name ts-pg -e POSTGRES_USER=scheduler -e POSTGRES_PASSWORD=scheduler \
  -e POSTGRES_DB=scheduler -p 5432:5432 postgres:16-alpine

docker build -t taskscheduler-rabbit ../../docker/rabbitmq
docker run -d --name ts-rabbit -e RABBITMQ_DEFAULT_USER=scheduler \
  -e RABBITMQ_DEFAULT_PASS=scheduler -p 5672:5672 taskscheduler-rabbit

python scripts/apply_migrations.py "postgresql://scheduler:scheduler@localhost:5432/scheduler"

TASKSCHEDULER_TEST_DSN="postgresql://scheduler:scheduler@localhost:5432/scheduler" \
TASKSCHEDULER_TEST_AMQP="amqp://scheduler:scheduler@localhost:5672/" \
  pytest tests/integration
```

`scripts/apply_migrations.py` replays the project's Flyway migrations without a JVM, so the
tests don't need a built `scheduler-infra` image. It is a development shortcut — in a real
deployment `scheduler-infra` owns the schema.

No Kotlin process runs during the tests: an `outbox_pump` fixture stands in for the infra
leader that would otherwise drain the outbox into RabbitMQ. Tests that expect a job to be
delivered more than once (retries, DAG promotions, paused-type redelivery) request it.

Alternatively `docker compose up -d` at the repo root brings up Postgres, RabbitMQ and a
real `scheduler-infra` — closer to production, but it needs the Gradle-built image
(`./gradlew :standalone-runner:dockerImage`).
