Skip to content

Recipes

A cheat sheet of the patterns that cover most applications, and the mistakes that are easy to make. Each entry links to the tutorial that explains it in full.

Declare a task

A task is an async function taking a single TaskRun argument, decorated with @task.

from fluid.scheduler import TaskRun, task


@task
async def say_hi(ctx: TaskRun) -> None:
    ctx.logger.info("hi")

Parameters are a pydantic model, validated before the run starts:

class Greet(BaseModel):
    name: str = "world"


@task
async def greet(ctx: TaskRun[Greet]) -> None:
    ctx.logger.info("hi %s", ctx.params.name)

Annotate the second type parameter to get typed dependencies as well, TaskRun[Greet, Deps]. Without it ctx.deps is Any. See Tasks and Task Dependencies.

Choose a task manager

Process Class
Queues work, never runs it TaskManager
Runs queued work TaskConsumer
Runs queued work and owns the schedule TaskScheduler

TaskScheduler is the default choice for a single service. See Task Managers.

scheduler = TaskScheduler(deps=deps)
scheduler.register_from_module(my_tasks)

Tasks must be registered on every process that queues them, not only on the one that runs them.

Serve over HTTP

from fluid.scheduler import task_manager_fastapi

app = task_manager_fastapi(scheduler)

POST /tasks/{name} queues a run, GET /tasks lists tasks, GET /tasks-status reports the running managers. Your own routes reach the manager with TaskManagerDep. See Extending the FastAPI App.

Command line entry point

from fluid.scheduler.cli import TaskManagerCLI

if __name__ == "__main__":
    TaskManagerCLI(scheduler_app)()

Gives serve, ls, exec and enable. This is required for applications with CPU bound tasks. See Task Queue App.

Schedule

@task(schedule=every(timedelta(seconds=30)))
async def heartbeat(ctx: TaskRun) -> None: ...


@task(schedule=crontab(hours="*/2"))
async def report(ctx: TaskRun) -> None: ...

Schedules only fire in a process running a TaskScheduler.

Offload CPU bound work

@task(cpu_bound=True, timeout_seconds=600)
async def crunch(ctx: TaskRun) -> None:
    heavy_pandas_work()

The same declaration runs as a subprocess locally and as a Kubernetes Job in-cluster. See CPU bound tasks and K8s Jobs.

Shared resources

@dataclass
class Deps:
    http_client: HttpxClient = field(default_factory=HttpxClient)


deps = Deps()
scheduler = TaskScheduler(deps=deps)
scheduler.add_async_context_manager(deps.http_client)

Anything registered with add_async_context_manager is opened on startup and closed on shutdown. See Task Dependencies.

Retries, limits and timeouts

@task(
    timeout_seconds=300,
    max_concurrency=2,
    retry=RetryPolicy(max_attempts=3, wait=2.0, backoff=2.0),
    rate_limit_retry=RetryPolicy(max_attempts=10, wait=5.0),
)
async def fetch(ctx: TaskRun) -> None: ...

timeout_seconds defaults to 60. max_concurrency=0 means no limit. See Task Retries.

Queue work

await task_manager.queue("greet", name="luca")   # from anywhere
await ctx.queue("next_step", data_id=42)         # from inside a task, records the chain
run = await consumer.queue_and_wait("greet")     # queue and await the result
run = await task_manager.execute("greet")        # run inline, skipping the queue

Prefer chaining with ctx.queue over waiting. See Chaining tasks.

Persist run history

task_manager.with_plugin(TaskDbPlugin(CrudDB.from_env()))

Adds a /tasks-history API backed by Postgres. Requires the db extra. See Plugins.

Common mistakes

  • CPU bound task without a CLI entry point. The subprocess runs the application entry point, so it has to be a TaskManagerCLI. A consumer started any other way raises CpuBoundEntryPointError on startup.
  • Expecting a plain TaskManager to run anything. It queues, it does not consume. Nothing executes until a consumer is running somewhere.
  • Event handlers on a plain TaskManager. register_async_handler is a no-op there, so plugins built on lifecycle events, the database plugin included, record nothing. Attach them to the consumer.
  • More than one scheduler on a broker. Each one keeps its own record of what it last fired, in memory, so every due task is queued twice. Replicate consumers, not schedulers.
  • Expecting a consumer to fire schedules. Only a TaskScheduler watches the clock.
  • Sharing state in memory with a CPU bound task. It runs in another process with its own dependencies. Pass state through the database or the broker.
  • Blocking calls in a normal task. Anything CPU heavy or blocking freezes the event loop for every other task in the process. Mark it cpu_bound=True.
  • Forgetting to register a task on the producer. Queueing by name requires the task in the registry, and a consumer receiving a run for a task it does not know logs an unknown task error.