Basic usage¶

1. Create a broker and app¶

from faststream import FastStream
from faststream_redis_timers import TimersBroker
from redis.asyncio import Redis

client = Redis.from_url("redis://localhost:6379")
broker = TimersBroker(client)
app = FastStream(broker)

2. Define a subscriber¶

Subscribers work like any FastStream subscriber. Decorate a handler with @broker.subscriber(topic):

@broker.subscriber("reminders")
async def handle_reminder(message: str) -> None:
    print(f"Reminder fired: {message}")

3. Publish a timer¶

Use broker.publish() with activate_in (relative delay) or activate_at (absolute time) to schedule delivery. publish() returns the resolved timer_id. Keep it for cancellation or inspection.

from datetime import UTC, datetime, timedelta


@app.after_startup
async def schedule_reminder() -> None:
    # Relative — fire 24 hours from now
    await broker.publish(
        "Call dentist",
        topic="reminders",
        activate_in=timedelta(hours=24),
    )

    # Absolute — fire at a specific moment
    await broker.publish(
        "Quarterly review",
        topic="reminders",
        activate_at=datetime(2026, 6, 1, 9, tzinfo=UTC),
    )

activate_at must be timezone-aware. Passing both activate_in and activate_at raises ValueError.

The full example:

from datetime import timedelta
from faststream import FastStream
from faststream_redis_timers import TimersBroker
from redis.asyncio import Redis

client = Redis.from_url("redis://localhost:6379")
broker = TimersBroker(client)
app = FastStream(broker)


@broker.subscriber("reminders")
async def handle_reminder(message: str) -> None:
    print(f"Reminder fired: {message}")


@app.after_startup
async def schedule_reminder() -> None:
    await broker.publish(
        "Call dentist",
        topic="reminders",
        activate_in=timedelta(hours=24),
    )

Broker options¶

Parameter	Default	Description
`client`	required for production	`redis.asyncio.Redis` client instance; the caller owns its lifecycle (the broker does not close it). The constructor accepts `None`, but a broker without a client raises `IncorrectState` on the first operation; `None` is the test-broker shape (see Testing).
`timeline_key`	`timers_timeline`	Sorted set key prefix; each topic uses `{timeline_key}:{topic}`
`payloads_key`	`timers_payloads`	Hash key prefix; each topic uses `{payloads_key}:{topic}`
`start_timeout`	`3.0`	Seconds to wait for the subscriber's first Redis ping during startup
`graceful_timeout`	`15.0`	Seconds to wait for in-flight timers on shutdown

TimersBroker does not call .aclose() on the supplied client. Wrap it in async with Redis.from_url(...) as client: (or your own try/finally) so the connection is released when the application stops.

Timer IDs¶

Each timer has a unique timer_id. If you don't provide one, a UUID is generated automatically. You can supply your own to make a timer idempotent: publishing the same timer_id twice will overwrite the first silently (no error, no warning). This makes retries of publish() safe, but two unrelated callers that pick the same ID will overwrite each other. Namespace your IDs (e.g., f"invoice-{invoice_id}-due") to avoid accidental collisions.

await broker.publish(
    "INV-001",
    topic="invoices",
    timer_id="invoice-INV-001-due",
    activate_in=timedelta(days=30),
)

Past activation times fire immediately¶

Both activate_in=timedelta(seconds=-5) and activate_at=datetime.now(tz=UTC) - timedelta(seconds=5) produce a timer scheduled in the past, and the next subscriber poll picks it up and fires it immediately. No error is raised; this lets activate_at computations that take "marginally too long" still deliver instead of breaking. If you want strict scheduling, validate at the call site before publishing.

Cancelling timers¶

Cancel a pending timer before it fires using broker.cancel_timer(topic, timer_id). This is a no-op if the timer has already fired or does not exist. The timer_id returned from publish() is what you'd pass:

timer_id = await broker.publish(
    "INV-001",
    topic="invoices",
    activate_in=timedelta(days=30),
)

# Later — invoice was paid early, cancel the reminder
await broker.cancel_timer("invoices", timer_id)

Inspecting pending timers¶

Three broker methods let you inspect or bulk-cancel timers without poking Redis directly:

Method	Returns	Use for
`await broker.has_pending(topic, timer_id)`	`bool`	"is timer X still scheduled?"
`await broker.get_pending_timers(topic, before=None)`	`list[str]`	List pending IDs on topic, optionally filtered to those due by before (a timezone-aware `datetime`)
`await broker.cancel_all(topic)`	`int`	Cancel every timer on topic; returns the number removed

from datetime import UTC, datetime, timedelta

# Quick existence check
if await broker.has_pending("invoices", "invoice-INV-001-due"):
    ...

# All pending timers on a topic
pending = await broker.get_pending_timers("invoices")

# Only those due in the next hour
soon = await broker.get_pending_timers(
    "invoices",
    before=datetime.now(tz=UTC) + timedelta(hours=1),
)

# Wipe a topic's pending timers (e.g., during a maintenance reset)
removed = await broker.cancel_all("invoices")

These methods only inspect/cancel pending timers on the topic. Handlers that have already started running are unaffected.

Leased timers still appear as pending¶

Timers that have been claimed by a worker and are currently being processed have their score pushed forward by lease_ttl seconds (the visibility-timeout pattern; see How it works). They are still in the sorted set, so:

has_pending(topic, timer_id) returns True while the handler is running.
get_pending_timers(topic) (no before) includes them.
get_pending_timers(topic, before=datetime.now(tz=UTC)) excludes them: they are no longer due "by now" because their score is in the future.

If you are polling has_pending to detect "the timer fired and the handler finished", you will get false positives during the lease window. Pass before=datetime.now(tz=UTC) (or treat the timer as gone only after has_pending is False) to avoid this.

`cancel_all` race with executing handlers¶

If cancel_all(topic) runs while a worker is mid-handler for a leased timer on that topic, the handler runs to completion. When it finishes, its commit (the ZREM + HDEL that normally removes the timer) becomes a no-op because cancel_all has already deleted both keys for the topic. Side effects stay; only the Redis cleanup is skipped, so handlers that have side effects (sent emails, written rows) will have already done them. Use cancel_all for topic resets, not for "stop everything in flight."

Debug logging¶

Set log_level=logging.DEBUG on TimersBroker to emit per-timer DEBUG lines: timers fetched per poll cycle, claim contested by another worker, and timer delivered to handler. Useful for diagnosing "my timer didn't fire".