Background work¶

Scheduled jobs, the task queue, leader election and locks shared by every process, broadcast, and the outbox. In fastapi_alcazaba.background; the outbox in fastapi_alcazaba.background.outbox.

In the guide: Scheduled jobs, Task queue, Leader election, Locks, Broadcast, Outbox. In an app: the example.

fastapi_alcazaba.background.FastAPIScheduler ¶

FastAPIScheduler(cache: Cache | None = None, *, fail_open: bool = False, grace: float = 10.0)

Runs jobs on a schedule in each process, each tick once across them all.

Start it with the app: FastAPI(lifespan=scheduler.lifespan), or async with scheduler: inside your own lifespan (after FastAPICache.init).

Parameters:

Name	Type	Description	Default
`cache`	`Cache \| None`	where the ticks are claimed; FastAPICache's by default, resolved at `start()`. Not needed when every job has `exclusive=False`.	`None`
`fail_open`	`bool`	when the backend fails, run the exclusive jobs anyway, maybe in every process, instead of skipping the tick (the default).	`False`
`grace`	`float`	seconds `stop()` waits for the jobs that are running before cancelling them.	`10.0`

every ¶

every(*, seconds: float = 0, minutes: float = 0, hours: float = 0, days: float = 0, name: str | None = None, exclusive: bool = True) -> Callable[[F], F]

Run the decorated function (async, or sync in a thread; no arguments) every interval, at multiples of it since the Unix epoch: every(hours=1) runs on the hour (UTC). For other times, use cron.

Parameters:

Name	Type	Description	Default
`name`	`str \| None`	the job's name in the backend, logs and metrics; the function's module and name by default. Two jobs can't have the same.	`None`
`exclusive`	`bool`	run each tick in one process only.	`True`

cron ¶

cron(expression: str, *, tz: str | tzinfo = 'UTC', name: str | None = None, exclusive: bool = True) -> Callable[[F], F]

Run the decorated function when the cron expression says, in the time zone tz: five fields (minute, hour, day of the month, month, day of the week, Sunday being 0 or 7) with *, lists, ranges, steps and names (*/15 9-17 * * mon-fri), or @hourly, @daily, @weekly, @monthly, @yearly.

next_run ¶

next_run(name: str) -> datetime | None

When the job name runs next (UTC), None while the scheduler isn't running.

start `async` ¶

start() -> None

Start running the jobs.

stop `async` ¶

stop() -> None

Stop: jobs waiting for their next run are cancelled at once, those running get grace seconds to finish (a sync job's thread can't be stopped: it runs to its end).

lifespan `async` ¶

lifespan(app: Any = None) -> AsyncIterator[None]

For FastAPI(lifespan=scheduler.lifespan): runs the jobs while the app runs.

fastapi_alcazaba.background.FastAPITaskQueue ¶

FastAPITaskQueue(cache: Cache | None = None, *, name: str = 'default', concurrency: int = 10, grace: float = 10.0, poll_interval: float = 1.0, visibility_timeout: float = 60.0, max_deliveries: int = 5, max_payload_size: int = 65536, max_length: int = 100000, max_dead_letters: int = 10000, replay_window: float = 7 * 24 * 3600)

Tasks that run in the background, in any process sharing the cache's backend.

Every process enqueues; those that call start() (or use lifespan or async with) also run tasks. A worker-only process: async with queue: await asyncio.Event().wait().

Needs Redis 6.2+ or Valkey through RedisBackend or ValkeyBackend (or a TieredBackend in front of one). Its keys are state keys (Cache.state_key), which clear() never deletes.

Parameters:

Name	Type	Description	Default
`cache`	`Cache \| None`	whose backend holds the queue; FastAPICache's by default, resolved when first needed.	`None`
`name`	`str`	queues with different names are independent.	`'default'`
`concurrency`	`int`	tasks run at a time by `start()` in this process.	`10`
`grace`	`float`	seconds `stop()` waits for the tasks that are running; those still running then are cancelled and go back to the queue (a sync task's thread can't be stopped: it runs to its end, so that task may run twice).	`10.0`
`poll_interval`	`float`	seconds between looks at the backend when there's nothing to do. A task enqueued in this process wakes its workers at once.	`1.0`
`visibility_timeout`	`float`	seconds after which the tasks of a worker that stopped answering (its process died) are taken over by another. Running tasks are kept from it every third of it, however long they run.	`60.0`
`max_deliveries`	`int`	times a task is handed to a worker (once, plus each time it's taken over from a worker that died) before it goes to the dead letters instead: it may be what kills them.	`5`
`max_payload_size`	`int`	largest arguments, in bytes of JSON; anything larger is refused.	`65536`
`max_length`	`int`	most tasks the queue holds (waiting, running or delayed); `enqueue` raises QueueFull beyond, instead of dropping tasks.	`100000`
`max_dead_letters`	`int`	dead letters kept, about: the oldest go first.	`10000`
`replay_window`	`float`	seconds a task may wait in the queue after it's due; later, it's refused as stale and goes to the dead letters (`requeue_dead_letters()` sends it again). Its run id is kept that long, so that a copy of it is refused.	`7 * 24 * 3600`

task ¶

task(name: str | None = None, *, retries: int = 3, backoff: float = 1.0, max_backoff: float = 600.0) -> Callable[[F], F]

Register the decorated function (async, or sync run in a thread) as a task. What it returns is ignored: keep results elsewhere.

Parameters:

Name	Type	Description	Default
`name`	`str \| None`	the task's name in the queue, which travels with every task (keep it when the function moves); the function's module and name by default.	`None`
`retries`	`int`	times a task that raised is tried again: `backoff` seconds after the first failure, twice that after the second... up to `max_backoff`, each with jitter. Then it goes to the dead letters. 0: at the first failure.	`3`

enqueue `async` ¶

enqueue(task: Callable[..., Any] | str, /, *args: Any, delay: float | timedelta | None = None, **kwargs: Any) -> str

Queue a run of task (a registered function, or its name) with these arguments, to start in delay seconds or as soon as a worker is free; returns the task's ID.

The arguments must fit the function and be JSON-serialisable (checked here, raising TypeError); they arrive as JSON gives them back: tuples as lists, dict keys as strings. A task can't take an argument named delay by keyword. Raises QueueFull when the queue holds max_length tasks, and BackendUnavailable when the backend fails.

start `async` ¶

start(concurrency: int | None = None) -> None

Start running tasks, up to concurrency at a time (the queue's by default).

stop `async` ¶

stop() -> None

Stop: no more tasks are taken, those running get grace seconds to finish, and those still running then are cancelled and go back to the queue.

lifespan `async` ¶

lifespan(app: Any = None) -> AsyncIterator[None]

For FastAPI(lifespan=queue.lifespan): runs tasks while the app runs.

pending `async` ¶

pending() -> int

How many tasks haven't finished: waiting for a worker, running, or waiting for their time (delayed ones and retries).

dead_letters `async` ¶

dead_letters(count: int = 100) -> list[DeadLetter]

The latest count dead letters, newest first.

requeue_dead_letters `async` ¶

requeue_dead_letters(ids: Iterable[str] | None = None) -> int

Put dead letters back in the queue (all of them by default) with their attempts reset, and return how many. Those whose task can't run here stay: tampered, malformed or too large, or named after no task of this process. Raises QueueFull when the queue fills up (those requeued until then stay requeued).

discard_dead_letters `async` ¶

discard_dead_letters(ids: Iterable[str]) -> int

Delete these dead letters; returns how many there were.

fastapi_alcazaba.background.DeadLetter `dataclass` ¶

DeadLetter(id: str, reason: str, task: str | None, task_id: str | None, error: str | None, attempts: int, failed_at: datetime)

A task that won't run again unless it's requeued. Its arguments aren't shown.

fastapi_alcazaba.background.QueueFull ¶

Bases: Exception

The queue already holds max_length tasks: the workers aren't keeping up.

fastapi_alcazaba.background.FastAPILeader ¶

FastAPILeader(name: str, *, ttl: float = 15.0, cache: Cache | None = None)

Elects one process as the leader named name and runs functions there while it leads.

Parameters:

Name	Type	Description	Default
`name`	`str`	the leadership's name in the backend; processes with the same name and backend compete for it.	required
`ttl`	`float`	seconds after which the leadership of a process that died is lost; renewed every `ttl / 3` seconds while held.	`15.0`
`cache`	`Cache \| None`	the Cache whose backend holds the lock; FastAPICache's by default, resolved at `start()`.	`None`

is_leader `property` ¶

is_leader: bool

Whether this process leads now.

run ¶

run(func: F) -> F

Run the decorated async function (no arguments) whenever this process leads; it's cancelled when the leadership is lost or the leader stops, and runs again at the next election. It must let the cancellation through. Add them before start().

start `async` ¶

start() -> None

Start campaigning.

stop `async` ¶

stop() -> None

Stop: the functions are cancelled and the leadership released, for another process to take over.

lifespan `async` ¶

lifespan(app: Any = None) -> AsyncIterator[None]

For FastAPI(lifespan=leader.lifespan): campaigns while the app runs.

fastapi_alcazaba.background.Lock ¶

Lock(cache: Cache, name: str, *, ttl: float = 30.0, wait: float | None = 0.0, renew: bool = True, record_wait: bool = True)

A lock named name in the cache's backend (Cache.lock creates them).

Parameters:

Name	Type	Description	Default
`ttl`	`float`	seconds after which the lock expires if its holder neither renews nor releases it.	`30.0`
`wait`	`float \| None`	seconds to wait for the lock when someone else holds it; 0 tries once, None waits as long as it takes.	`0.0`
`renew`	`bool`	keep renewing it while it's held.	`True`
`record_wait`	`bool`	record the time spent acquiring it (`fastapi_alcazaba.lock.wait`); off for the locks that are tried over and over (concurrency slots, leader elections).	`True`

held `property` ¶

held: bool

Acquired, not released, and not expired since the last renewal.

acquire `async` ¶

acquire() -> bool

Take the lock, waiting up to wait seconds; False if someone else still holds it. Raises BackendUnavailable when the backend fails: without it nobody can tell who holds the lock.

release `async` ¶

release() -> None

Release it, if this Lock holds it. A backend failure is only logged: the lock expires after ttl anyway.

fastapi_alcazaba.background.LockNotAcquired ¶

Bases: Exception

Someone else held the lock for longer than the time to wait for it.

fastapi_alcazaba.background.FastAPIBroadcast ¶

FastAPIBroadcast(client: Any = None, *, signing_key: Secrets | None = None, prefix: str = 'fastapi_alcazaba:broadcast:', queue_size: int = 1000)

Publishes messages to channels, which every process's subscribers of the channel get.

At most once: a message is for those subscribed when it's published; one that a subscriber's connection missed doesn't come back, it gets a gap instead. For work that must happen, a task queue or an outbox.
Signed: with signing_key, each message carries an HMAC of its channel and data, and those without it (planted by whoever can write to Redis) are dropped.
Bounded: a message has MAX_MESSAGE bytes of JSON at most, and each subscriber queue_size messages waiting: past it, it loses the oldest and gets a gap.
Reconnects by itself, with a backoff, and tells every subscriber of the gap.

Parameters:

Name	Type	Description	Default
`client`	`Any`	a redis-py or valkey-py asyncio client (the one of your RedisBackend); None to keep messages in this process.	`None`
`signing_key`	`Secrets \| None`	a secret of 32 bytes or more, the same in every process (a list rotates it: the first signs, all of them verify).	`None`
`prefix`	`str`	what the channels are under in Redis.	`'fastapi_alcazaba:broadcast:'`
`queue_size`	`int`	messages a subscriber may have waiting.	`1000`

publish `async` ¶

publish(channel: str, data: Any) -> None

Send data (JSON) to the subscribers of channel, in every process. Raises when Redis fails: the message isn't sent.

subscribe `async` ¶

subscribe(channel: str) -> AsyncIterator[AsyncIterator[Message]]

The messages of channel from now on, for as long as the block lasts.

close `async` ¶

close() -> None

Stop listening (in the lifespan, at shutdown).

fastapi_alcazaba.background.Message `dataclass` ¶

Message(channel: str, data: Any = None, gap: bool = False)

A message of a channel, as subscribers get it.

Parameters:

Name	Type	Description	Default
`channel`	`str`	the channel it was published to.	required
`data`	`Any`	what was published (JSON), None for a gap.	`None`
`gap`	`bool`	messages may have been missed before this one (the subscription to Redis dropped, or this subscriber fell behind): show the state again, from where it's kept.	`False`

fastapi_alcazaba.background.outbox.SQLAlchemyOutbox ¶

SQLAlchemyOutbox(table: Table, sessions: Callable[[], AnySession])

The outbox in a table of your database (outbox_table), written through your sessions.

Parameters:

Name	Type	Description	Default
`table`	`Table`	the outbox's table.	required
`sessions`	`Callable[[], AnySession]`	makes the relay's sessions (a `sessionmaker` or an `async_sessionmaker`).	required

add `async` ¶

add(session: AnySession, topic: str, payload: Any) -> str

Write a message in session's transaction: it's sent once the transaction commits, and never if it's rolled back. Returns its id.

due `async` ¶

due(limit: int) -> list[OutboxMessage]

The oldest messages not sent, whose time to be tried has come.

give_up `async` ¶

give_up(message_id: str, attempts: int, error: str) -> None

No more tries: kept as sent, with failed and its last error, for someone to look.

fastapi_alcazaba.background.outbox.outbox_table ¶

outbox_table(metadata: MetaData, name: str = 'outbox') -> Table

The outbox's table, in your metadata (its migrations create it): each message's id, topic, payload (JSON), when it was written, the attempts at delivering it, when it may be tried again, and when it was sent (or given up on).

fastapi_alcazaba.background.outbox.OutboxMessage `dataclass` ¶

OutboxMessage(id: str, topic: str, payload: Any, attempts: int, created: float)

A message of the outbox, as deliver gets it.

Parameters:

Name	Type	Description	Default
`id`	`str`	random, unique: what consumers deduplicate by.	required
`topic`	`str`	what it's about, such as "order.placed".	required
`payload`	`Any`	what was written with it.	required
`attempts`	`int`	the deliveries tried before this one.	required
`created`	`float`	when it was written (Unix time).	required

fastapi_alcazaba.background.outbox.OutboxRelay ¶

OutboxRelay(outbox: SQLAlchemyOutbox, deliver: Callable[[OutboxMessage], Awaitable[Any]], *, batch: int = 100, interval: float = 1.0, max_attempts: int = 10, max_delay: float = 3600)

Delivers the outbox's messages, the oldest first, at least once: deliver(message) is awaited, and the message marked as sent when it returns. When it raises, the message is tried again later (1, 2, 4… seconds, up to max_delay), and given up on after max_attempts: kept with its error, never sent. Run it in one process (FastAPILeader).

Parameters:

Name	Type	Description	Default
`outbox`	`SQLAlchemyOutbox`	where the messages are.	required
`deliver`	`Callable[[OutboxMessage], Awaitable[Any]]`	`deliver(message)`, async: enqueue a task, call a webhook, publish.	required
`batch`	`int`	messages read at a time.	`100`
`interval`	`float`	seconds between two reads when there was nothing to deliver.	`1.0`
`max_attempts`	`int`	deliveries tried before giving up on a message.	`10`
`max_delay`	`float`	the longest wait between two tries, in seconds.	`3600`

relay_once `async` ¶

relay_once() -> int

Deliver the messages due now; how many were sent.

run `async` ¶

run() -> None

Relay until cancelled: a batch, and when there was nothing due, a pause.

Background work¶

fastapi_alcazaba.background.FastAPIScheduler ¶

every ¶

cron ¶

next_run ¶

start async ¶

stop async ¶

lifespan async ¶

fastapi_alcazaba.background.FastAPITaskQueue ¶

task ¶

enqueue async ¶

start async ¶

stop async ¶

lifespan async ¶

pending async ¶

dead_letters async ¶

requeue_dead_letters async ¶

discard_dead_letters async ¶

fastapi_alcazaba.background.DeadLetter dataclass ¶

fastapi_alcazaba.background.QueueFull ¶

fastapi_alcazaba.background.FastAPILeader ¶

is_leader property ¶

run ¶

start async ¶

stop async ¶

lifespan async ¶

fastapi_alcazaba.background.Lock ¶

held property ¶

acquire async ¶

release async ¶

fastapi_alcazaba.background.LockNotAcquired ¶

fastapi_alcazaba.background.FastAPIBroadcast ¶

publish async ¶

subscribe async ¶

close async ¶

fastapi_alcazaba.background.Message dataclass ¶

fastapi_alcazaba.background.outbox.SQLAlchemyOutbox ¶

add async ¶

due async ¶

give_up async ¶

fastapi_alcazaba.background.outbox.outbox_table ¶

fastapi_alcazaba.background.outbox.OutboxMessage dataclass ¶

fastapi_alcazaba.background.outbox.OutboxRelay ¶

relay_once async ¶

run async ¶

start `async` ¶

stop `async` ¶

lifespan `async` ¶

enqueue `async` ¶

start `async` ¶

stop `async` ¶

lifespan `async` ¶

pending `async` ¶

dead_letters `async` ¶

requeue_dead_letters `async` ¶

discard_dead_letters `async` ¶

fastapi_alcazaba.background.DeadLetter `dataclass` ¶

is_leader `property` ¶

start `async` ¶

stop `async` ¶

lifespan `async` ¶

held `property` ¶

acquire `async` ¶

release `async` ¶

publish `async` ¶

subscribe `async` ¶

close `async` ¶

fastapi_alcazaba.background.Message `dataclass` ¶

add `async` ¶

due `async` ¶

give_up `async` ¶

fastapi_alcazaba.background.outbox.OutboxMessage `dataclass` ¶

relay_once `async` ¶

run `async` ¶