Task Scheduler

A simple scheduler that runs tasks periodically. Use it for lightweight recurring jobs without a full task queue.

Fast and easy: simple decorators to define and schedule tasks with minimal boilerplate.
Interval tasks: run tasks at fixed intervals, locally or across a cluster.
Coordination: control concurrency with distributed primitives (see Coordination primitives).
Dependency injection: use FastDepends to inject dependencies into tasks.
Error handling: errors are caught and logged, so a failing task does not stop the scheduler.

Quick start

Register a Tasks instance with a Grelmicro app, then schedule a task with the every decorator:

from grelmicro import Grelmicro
from grelmicro.task import Tasks

tasks = Tasks()
micro = Grelmicro(uses=[tasks])

@tasks.every(seconds=5)
async def cleanup() -> None:
    ...

async with micro:
    ...

Per-process by default

Tasks runs schedules in the local process only. Every process that boots a Tasks instance runs its own copy of every registered task. To run a task at most once across the fleet, gate it with TaskLock or LeaderElection. Without one of those, a 3-replica deployment runs the same @tasks.every(...) three times per tick.

Note

This is not a replacement for full task queues such as Celery, taskiq, or APScheduler. It is small, simple, and safe for running tasks in a distributed system.

Tasks

The Tasks class is the main entry point to manage tasks. The recommended way to lifecycle it is to register it with a Grelmicro app, as shown in the quick start above.

Grelmicro.use(item) (or the uses= constructor kwarg) accepts any async context manager and lifecycles it with the app. The caller keeps the reference and uses the manager directly.

Choose the entry point by the job:

Need	Use
Simple recurring function	`@tasks.every(...)`
Group tasks across modules	`TaskRouter`
Add an object that already implements the task protocol	`tasks.add_task(...)`
Run at most once across replicas	`@tasks.every(..., lock=TaskLock(...))`
Run only on the leader	`@tasks.every(..., leader=leader_election)`

Start it standalone using the application lifespan:

FastAPIFastStream

from contextlib import asynccontextmanager

from fastapi import FastAPI

from grelmicro.task import Tasks

task = Tasks()


@asynccontextmanager
async def lifespan(app: FastAPI):
    async with task:
        yield


app = FastAPI(lifespan=lifespan)

from contextlib import asynccontextmanager

from faststream import ContextRepo, FastStream
from faststream.redis import RedisBroker

from grelmicro.task import Tasks

task = Tasks()


@asynccontextmanager
async def lifespan(context: ContextRepo):
    async with task:
        yield


broker = RedisBroker()
app = FastStream(broker, lifespan=lifespan)

Interval Task

Use the every decorator to run a task at a fixed interval:

Note

The interval specifies the waiting time between task executions. Ensure that the task execution duration is considered to meet deadlines effectively.

The interval is measured from the end of one run to the start of the next (end-to-start). A run that takes longer than the interval pushes the next attempt back.

Sensitive workflows: pass an explicit name=

When name= is omitted, the task reference is derived from the function's module:qualname. That reference appears in logs, distributed coordination keys (when TaskLock is used), and metric labels. Pass an explicit name="..." for tasks that handle credentials, customer data, or other workflows where the internal module path should not leak through operational surfaces.

TasksTaskRouter

from grelmicro.task import Tasks

task = Tasks()


@task.every(seconds=5)
async def my_task():
    print("Hello, World!")

from grelmicro.task import TaskRouter

task = TaskRouter()


@task.every(seconds=5)
async def my_task():
    print("Hello, World!")

Distributed Lock

Pass a TaskLock via lock to enable distributed locking: the task runs at most once per interval across all workers. The lock keeps its default "default" name, so the task name is used and you never repeat it.

from grelmicro.coordination import TaskLock
from grelmicro.task import Tasks

task = Tasks()


@task.every(seconds=60, lock=TaskLock(lease_duration=300))
async def cleanup():
    print("Running cleanup...")

Parameter	Description
`seconds`	Duration between each scheduling attempt, as a number of seconds or a `timedelta`. Each worker retries every interval, but only one executes per interval.
`lock`	A `TaskLock` for at-most-once scheduling. Its `lease_duration` is the crash-protection TTL and must be >= `seconds`. Its `min_hold_duration` keeps the lock held after completion to prevent re-execution too soon.

The lock is authoritative: its lease_duration, min_hold_duration, backend, and worker are used as set.

Leader Gating

Restrict the task to the leader worker with a Leader Election, so only one worker executes it. Setting leader also enables distributed locking. Without a lock, one is configured with lease_duration of seconds * 5 and min_hold_duration of seconds:

from grelmicro.coordination import LeaderElection
from grelmicro.coordination.memory import MemoryLeaderElectionAdapter
from grelmicro.task import Tasks

leader = LeaderElection("my-service", backend=MemoryLeaderElectionAdapter())
task = Tasks()
task.add_task(leader)


@task.every(seconds=60, leader=leader)
async def cleanup():
    print("Running cleanup...")

Custom Lock Timing

For long-running tasks, customize both lease_duration and min_hold_duration on the TaskLock:

from grelmicro.coordination import TaskLock
from grelmicro.task import Tasks

task = Tasks()


@task.every(
    seconds=60,
    lock=TaskLock(lease_duration=300, min_hold_duration=30),
)
async def long_task():
    print("Running long task...")

Resource Lock

Combine distributed locking with a Lock to synchronize access to a shared resource during task execution. Pass the Lock via the sync parameter:

from grelmicro.coordination import Lock, TaskLock
from grelmicro.task import Tasks

task = Tasks()
resource_lock = Lock("shared-resource")


@task.every(seconds=60, lock=TaskLock(lease_duration=300), sync=resource_lock)
async def cleanup():
    print("Running cleanup...")

How It Works

When the lock is already held, the task skips the execution (logged at DEBUG level) and retries on the next interval.

Node A:  [acquire] → [execute] → [hold for seconds] → [TTL expires]
Node B:  [skip] → ... → [skip] → ... → [acquire] → [execute]

When combining leader gating, distributed locking, and a resource lock, the synchronization primitives are acquired in this order:

Order	Primitive	Purpose
1	`LeaderElection`	Rejects non-leader workers immediately without acquiring any lock, which avoids unnecessary contention.
2	`TaskLock`	Guarantees at-most-once execution per interval. It is acquired after leadership is confirmed so the TTL window stays short.
3	`Lock`	User-provided lock for shared-resource access. It is acquired last so the resource is held only during actual execution.

Each primitive is only acquired if the previous one succeeded. For example, a non-leader worker is rejected at step 1 and never touches the task lock or resource lock.

Cron Task

Use the cron decorator to run a task on a cron schedule:

from grelmicro.task import Tasks

task = Tasks()


@task.cron("0 2 * * *", timezone="Europe/Zurich")
async def nightly_report():
    print("Running the nightly report at 02:00 Zurich time")

The expression has five fields: minute hour day-of-month month day-of-week. The example above runs every day at 02:00 in the Europe/Zurich timezone. The timezone defaults to "UTC".

Each field accepts:

Syntax	Meaning
`*`	Every value
`*/15`	Every 15th value (a step)
`9-17`	A range
`9-17/2`	Every second value in a range
`1,15,45`	A list of values
`5`	A single value

Field ranges are minute 0-59, hour 0-23, day-of-month 1-31, month 1-12, and day-of-week 0-6 where 0 is Sunday. The value 7 also means Sunday.

Day-of-month and day-of-week

When both day-of-month and day-of-week are restricted (neither is *), a day matches if it matches either field. For example, 0 0 15 * 1 runs on the 15th of the month and on every Monday. When only one is restricted, only that one applies.

Distributed cron

With a Coordination component wired, every fire is claimed against durable state, so the task runs at most once across all workers per fire:

@task.cron("*/5 * * * *")
async def sync_data():
    ...

The schedule backend stores the last fire on the provider (Redis, Postgres, and SQLite all ship today). Because that state is durable, a fire missed while every worker was down replays once when a worker comes back. Only the most recent missed fire runs, never a backlog of skipped ones. Without a backend, the task runs on every worker, every fire. Kubernetes is intentionally not provided: use a native Kubernetes CronJob.

Set misfire_grace_seconds to bound how late a missed fire may run:

@task.cron("0 * * * *", misfire_grace_seconds=600)
async def hourly_rollup():
    ...

A fire more than 600 seconds late is dropped instead of replayed. The default is None, which replays any missed fire however late.

Make the body idempotent

The guarantee is at-most-once. A worker that claims a fire and then crashes mid-run does not retry it, because the last-fire state already advanced. Make the body idempotent, or wrap it with @retry, when correctness depends on completion.

Cron in distributed systems

On Kubernetes, when the task is a batch job and you can define manifests, prefer a native Kubernetes CronJob that runs a one-shot command. It is the platform's job and the least code. Grelmicro does not create CronJob resources and should not, since that needs cluster-write permissions an application should not hold.

Use grelmicro @cron when you want the task to run inside the live service with its warm connections and dependencies, or want one scheduling model across Redis, Postgres, SQLite, and bare metal.

Task Introspection

Each task exposes two read-only properties for observability:

next_fire_time: the next scheduled fire as a timezone-aware datetime, or None when the task has not started yet. For interval tasks, this is computed from the last loop instant. For cron tasks, it comes from the parsed expression.
last_fire: a FireInfo with the started_at timestamp, outcome (a FireOutcome enum: SUCCESS, ERROR, or SKIPPED), and duration in seconds. None before the first fire. FireOutcome is a StrEnum, so each member compares equal to its string value (outcome == "success").

Access the task object via tasks.tasks:

from grelmicro.task import FireInfo, Tasks

tasks = Tasks()

@tasks.every(seconds=60)
async def cleanup() -> None:
    ...

# After startup: tasks.tasks holds IntervalTask and CronTask objects.
# The decorator returns the original function unchanged.
task = tasks.tasks[-1]
info: FireInfo | None = task.last_fire
if info is not None:
    print(info.outcome, info.duration)

next_fire = task.next_fire_time  # None until the first loop iteration

Task Router

For bigger applications, use the TaskRouter class to organize tasks across modules:

from grelmicro.task import TaskRouter


router = TaskRouter()


@router.every(seconds=5)
async def my_task():
    print("Hello, World!")

Then include the TaskRouter into the Tasks or other routers:

from grelmicro.task import Tasks

task = Tasks()
task.include_router(router)

Tip

The TaskRouter follows the same philosophy as the APIRouter in FastAPI or the Router in FastStream.

See Coordination primitives for more details.