async-timer

The missing Python async timer.

The problem

The obvious way to run something on an interval in asyncio:

asyncdefrefresh():
whileTrue:
awaitdo_the_thing()
awaitasyncio.sleep(5)

This works until it doesn't:

• It drifts.sleep(5) runs afterdo_the_thing() finishes, so a 2-second call gives you a 7-second period. Drift compounds.
• It has no cancellation story. Cancelling the wrapping task during do_the_thing() may interrupt mid-write; you need explicit shielding and cleanup to handle this safely.
• The first call isn't observable. Want to wait until the cache is populated before serving traffic? You have to bolt on an Event.
• There's nowhere for consumers to subscribe. If something else wants the latest value (or every value), you're hand-rolling a queue or a fanout.
• It doesn't compose. Running ten of these on startup and cancelling them together on shutdown needs a TaskGroup plus bookkeeping.

async-timer is what you actually want: a recurring async call with proper lifecycle, two delivery models, and the operational knobs (fixed-rate vs fixed-delay, jitter, trigger-now, cross-thread control) that real production code ends up needing.

Zero runtime dependencies. Python 3.9+.

Install

pip install async-timer

The 30-second example: FastAPI cache warmup

importcontextlibimporttimeimportuvicornfromfastapiimportFastAPIimportasync_timerDB_CACHE= {"initialised": False}
asyncdefrefresh_db():
DB_CACHE.update(initialised=True, cur_value=time.time())
@contextlib.asynccontextmanagerasyncdeflifespan(_app: FastAPI):
asyncwithasync_timer.Timer(delay=5, target=refresh_db) astimer:
awaittimer.wait(hit_count=1) # block startup until the first refreshyield# serve traffic; timer keeps refreshingapp=FastAPI(lifespan=lifespan)
@app.get("/")asyncdefroot():
return {"db_cache": DB_CACHE}

Two lines do the heavy lifting:

• async with Timer(...) as timer starts on enter, cancels on exit (and awaits cleanup — no orphan task).
• await timer.wait(hit_count=1) gates startup on the first successful refresh.

More recipes: docs/recipes/. Runnable scripts: docs/examples/.

Features

• Zero runtime dependencies.
• Any callable shape. Sync or async functions, generators, async generators, or callables returning any of those.
• Two delivery models.join() / wait() / async for self is single-shot fan-out (latest value, may drop intermediate ticks under slow consumers). subscribe() gives each consumer a buffered queue (every tick, optional maxsize for bounded drop-oldest).
• Scheduling modes.fixed_delay (default; next tick fires delay after the previous one finishes) or fixed_rate (anchored to wall clock; missed slots skipped + logged). Optional initial_delay and jitter.
• Trigger on demand.await timer.trigger() fires now and resumes the schedule.
• Last-value cache.timer.last_result / timer.last_tick_at — no blocking.
• Cancel anytime. Explicit cancel() or constructor cancel_aws (awaitables that stop the timer when they resolve). await cancel() waits for cleanup before returning; safe from inside the target/callbacks.
• Restartable.start() after cancel() works (raises TimerRestartError if cancel_aws was used — those are single-shot).
• Decorator.@async_timer.every(5) wraps a function into a Timer; original on .func.
• Groups.TimerGroup() starts/cancels a set of timers together.
• Named.name="db_refresh" shows in repr() and scopes the logger.
• Test-friendly.mock_async_timer.MockTimer replaces real sleeps with an AsyncMock.

When to use this — and when not to

async-timer is for in-process recurring work driven by asyncio: cache refresh, periodic polling, metrics sampling, heartbeats, fan-out feeds. Its model is one process, one loop, many timers.

You want	Reach for
Periodic work in an asyncio app	async-timer
Cron-style wall-clock scheduling ("every Monday 9am")	APScheduler, aiocron
Background jobs with retries, queues, persistence	arq, dramatiq, Celery + beat
Recurring tasks across many processes	A scheduler + broker (Celery beat, arq cron)
One-shot setTimeout-equivalent	loop.call_later (stdlib)

If you need durability or cross-process coordination, you need a broker. async-timer doesn't try to be that.

More examples

join()

importasyncioimportasync_timerasyncdefmain():
timer=async_timer.Timer(12, target=lambda: 42)
timer.start()
val=awaittimer.join() # 42, after the first tickawaittimer.cancel()
asyncio.run(main())

async for

importasyncio, timeimportasync_timerasyncdefmain():
asyncwithasync_timer.Timer(14, target=time.time) astimer:
asyncfortintimer:
print(t) # current time every 14 secondsasyncio.run(main())

Decorator

importasync_timer@async_timer.every(5, mode="fixed_rate", name="db_refresh")asyncdefrefresh_db():
...
awaitrefresh_db.func() # call the undecorated fn (tests)asyncdefmain():
refresh_db.start()
awaitrefresh_db.join()
awaitrefresh_db.cancel()

TimerGroup

importasync_timerasyncdeflifespan():
asyncwithasync_timer.TimerGroup(name="caches") asgroup:
group.add(async_timer.Timer(5, target=refresh_db))
group.add(async_timer.Timer(60, target=prune_cache))
# Block until every cache has been populated at least once,# then serve traffic.awaitgroup.wait(hit_count=1)
yield# both running; both cancelled on exit

TimerGroup mirrors Timer's surface across a set of timers; each group method fans out to its members (AND-combined) and returns [(timer, rv), ...] in iteration order:

• group.wait(hit_count=...) / wait(hits=...) — block until every member satisfies the condition.
• group.trigger() — fire every member's target now (cache-invalidate-all).
• group.is_running() — True iff active and every member is running.
• group.start() / await group.cancel_all() — explicit lifecycle for use outside async with.
• group.cancel_threadsafe(timeout=5.0) — cancel from a non-loop thread (signal handler, sync REST endpoint, worker thread).

Each of wait() and trigger() accepts timeout= (whole-group wall-clock bound) and return_exceptions=True (per-member errors appear in the result list instead of propagating, mirroring asyncio.gather).

Trigger now

asyncdefforce_refresh(timer):
returnawaittimer.trigger()

Latest value, no blocking

@async_timer.every(5)asyncdefrefresh_db():
returnawaitdb.fetch()
defget_cached():
returnrefresh_db.last_result# None until the first tick

Every-tick delivery via subscribe()

join() / async for self drop ticks under slow consumers (single-shot fan-out). Use subscribe() when you need every tick:

asyncwithtimer.subscribe() asfeed:
asyncforvalueinfeed:
awaitlog_it(value) # never misses a tick from subscribe-timeawaitasyncio.sleep(3.0) # even though the consumer is slow

Bounded queue (drop oldest + log when full):

asyncwithtimer.subscribe(maxsize=10, name="metrics-sink") asfeed:
asyncforvalueinfeed:
awaitslow_export(value)

Multiple subscribers each get an independent copy:

asyncwithtimer.subscribe() asa, timer.subscribe() asb:
...

Consumer-side load shedding:

asyncwithtimer.subscribe() asfeed:
asyncforvalueinfeed:
iffeed.qsize>100:
feed.drop_oldest(feed.qsize-1) # keep only the newestlog.warning("shed %d ticks", feed.dropped_count)
awaitslow_export(value)

drop_oldest() never swallows end-of-stream / exception sentinels. Target exceptions re-raise from the subscriber's iteration.

Tests with MockTimer

mock_async_timer.MockTimer is a drop-in Timer subclass that replaces the real sleep with an AsyncMock, so ticks fire as fast as the loop can schedule them — no wall-clock waits in tests:

frommock_async_timerimportMockTimerasyncdeftest_periodic_refresh():
calls=0deftick():
nonlocalcallscalls+=1asyncwithMockTimer(0.1, tick) ast:
awaitt.wait(hits=3)
assertcalls==3

Same surface as Timer — join, wait, trigger, subscribe, TimerGroup, decorator wrapping all work the same way.

Exceptions

All library-raised errors derive from async_timer.TimerError, which itself inherits from RuntimeError for back-compat with existing except RuntimeError clauses:

Exception	Raised when
TimerAlreadyRunningError	start() called on a running timer
TimerNotRunningError	trigger() / join() on a stopped timer
TimerRestartError	start() after cancel() on a Timer built with cancel_aws (single-shot)
ThreadsafeDispatchError	*_threadsafe called from the bound loop thread, before start, or after the loop closed

Catch TimerError to filter only library-originated errors; catch a specific subclass for finer control.

Thread safety

A Timer runs in a single asyncio event loop. Most state-mutating operations must be called from the loop's thread. The following are explicitly safe to use from any thread:

Read-only attributes (atomic under CPython's GIL):

• timer.last_result, timer.last_tick_at, timer.hit_count
• timer.is_running(), timer.delay, timer.name
• subscription.qsize, subscription.dropped_count

set_delay(new_delay) is a single attribute write — safe from any thread; takes effect on the next sleep.

Cross-thread control methods — marshal the operation back to the timer's loop and block for completion:

# From a sync REST handler, signal handler, worker thread, etc.:timer.cancel_threadsafe(timeout=5.0) # raises TimeoutError if exceededresult=timer.trigger_threadsafe(timeout=5.0)
feed.close_threadsafe()

These raise ThreadsafeDispatchError (a RuntimeError subclass) with a clear message if called from the timer's own loop thread (use await cancel() / await trigger() instead), or if the timer has not been started yet, or if the bound event loop has been closed.

Anything else (subscribe(), awaiting join() / wait(), iterating async for over the timer or a subscription, reading from a subscription queue) must happen on the loop's thread. From other threads, use asyncio.run_coroutine_threadsafe(coro, loop) to dispatch.

License

MIT.