# Aio Fluid An async task queue for Python backend services. Tasks are async functions declared with a decorator, scheduled with every() or crontab(), and run concurrently on asyncio. A task marked cpu_bound runs in a subprocess, or as a Kubernetes Job in a cluster, so heavy work never blocks the event loop. # Getting started # **An async task queue that offloads CPU-bound work to subprocesses or Kubernetes Jobs** **Documentation**: [fluid.quantmind.com](https://fluid.quantmind.com/) **Source**: [github.com/quantmind/aio-fluid](https://github.com/quantmind/aio-fluid) Declare tasks with a `@task` decorator, schedule them with `every()` or cron, and run them concurrently on asyncio. The part that sets `aio-fluid` apart: mark a task `cpu_bound=True` and it runs in a **fresh subprocess** so heavy CPU work never freezes the event loop. And when your consumer runs inside Kubernetes, the *same task* dispatches as a **Kubernetes Job** instead, with **no code change**. ```python import os from datetime import timedelta from fastapi import FastAPI from pydantic import BaseModel from fluid.scheduler import TaskRun, TaskScheduler, every, task, task_manager_fastapi from fluid.scheduler.cli import TaskManagerCLI class Report(BaseModel): rows: int = 5_000_000 def heavy_pandas_work(rows: int) -> None: """Stand-in for the CPU-heavy work you would do in a real task.""" sum(range(rows)) @task(schedule=every(timedelta(seconds=5))) async def heartbeat(ctx: TaskRun) -> None: """IO-bound task, scheduled every five seconds runs concurrently on the event loop """ ctx.logger.info("still alive") @task( cpu_bound=True, schedule=every(timedelta(seconds=20), delay=timedelta(seconds=5)), timeout_seconds=600, ) async def crunch(ctx: TaskRun[Report]) -> None: """CPU-bound task, scheduled every 20 seconds with an initial delay of 5 seconds Same decorator, one flag. Runs in a subprocess (or a Kubernetes Job in-cluster) so the heavy work never blocks the event loop. Identical code in both places. """ heavy_pandas_work(ctx.params.rows) ctx.logger.info("crunch finished on pid %d", os.getpid()) def scheduler_app() -> FastAPI: scheduler = TaskScheduler() scheduler.register_from_dict(globals()) return task_manager_fastapi(scheduler) if __name__ == "__main__": TaskManagerCLI( scheduler_app, help="Simple Task Manager CLI with default commands", log_config=dict(app_names=("__main__", "fluid")), )() ``` ## Why aio-fluid? Most Python task queues force a choice: async-native runners (`arq`, `taskiq`) that assume your work never blocks the loop, or heavyweight brokers (`Celery`) that predate asyncio. Neither has a clean answer for *"this one task is CPU-heavy"* beyond "spin up a second worker fleet." `aio-fluid` treats CPU-bound work as a first-class task type: - **One decorator, two execution models.** `@task(cpu_bound=True)` runs locally as a subprocess and in-cluster as a Kubernetes Job: the switch is automatic (`KUBERNETES_SERVICE_HOST` + the `k8s` extra). Your task code is identical in both. See [K8s Jobs](https://fluid.quantmind.com/tutorials/task_k8s/). - **Async-native and typed.** Tasks are plain `async def` functions; parameters are [pydantic](https://docs.pydantic.dev/) models, validated on the way in. - **Dependency injection.** A database manager, an HTTP client or any other resource is grouped into a single typed `deps` object every task run can reach, with startup and shutdown handled by the task manager. See [Task Dependencies](https://fluid.quantmind.com/tutorials/task_deps/). - **The scheduling you expect.** `every(timedelta(...))` and `crontab(...)`, per-task `max_concurrency`, priorities, `timeout_seconds`, and retry policies. - **FastAPI-ready.** Drop a task manager into a FastAPI app to queue and inspect runs over HTTP. - **Task lifecycle callbacks.** Every state a run moves through (`queued`, `running`, `success`, `failure`, `aborted`) is dispatched as an event you can subscribe to, with sync or async handlers, so metrics, alerting and bookkeeping hang off the queue instead of your task code. - **Task manager plugins.** Plugins hook into those same events and can mount their own HTTP routes. The bundled database plugin persists every run to Postgres and serves a `/tasks-history` API on top of it. See [Plugins](https://fluid.quantmind.com/tutorials/task_app/#plugins). - **Pluggable broker.** Redis by default; the broker is an interface, not a hard dependency. `Celery` is the mature, battle-tested default with the biggest ecosystem; reach for it when you need that breadth. `aio-fluid` is for async services that want CPU-bound work handled natively and scaled onto Kubernetes without a parallel worker deployment. For a feature-by-feature look at `aio-fluid` next to Celery, RQ, arq and taskiq, backed by download data, see [Python task queues compared](https://fluid.quantmind.com/comparison/). ## Batteries included Alongside the task queue, `aio-fluid` ships the building blocks Quantmind uses to run backend services: - **Async workers**: composable components with a managed start/stop lifecycle; the foundation the task queue is built on. See [Workers](https://fluid.quantmind.com/reference/workers/). - **Async Postgres CRUD**: a typed CRUD layer over `asyncpg` and SQLAlchemy, with pagination and schema migrations. See [Database](https://fluid.quantmind.com/reference/db/). - **Event dispatchers**: sync and async `Dispatcher` types for decoupling event sources from handlers. See [Dispatchers](https://fluid.quantmind.com/reference/dispatchers/). - **HTTP client helpers**: a unified async client wrapping `httpx` and `aiohttp`. See [HTTP Client](https://fluid.quantmind.com/reference/http_client/). - **CLI tooling**: ready-made `click` / `rich` command-line interfaces for task managers and databases. ## Installation This is a python package you can install via pip: ```text pip install aio-fluid ``` To install all the dependencies: ```text pip install aio-fluid[cli, db, http, log, k8s] ``` this includes the following extra dependencies: - `cli` for the command line interface using [click](https://click.palletsprojects.com/) and [rich](https://github.com/Textualize/rich) - `db` for database support with [asyncpg](https://github.com/MagicStack/asyncpg) and [sqlalchemy](https://www.sqlalchemy.org/) - `http` for http client support with [httpx](https://www.python-httpx.org/) and [aiohttp](https://docs.aiohttp.org/en/stable/) - `log` for JSON logging support with [python-json-logger](https://github.com/madzak/python-json-logger) - `k8s` for Kubernetes support for CPU bound tasks ## AI agents The documentation is published in a form coding agents can consume directly, following the [llms.txt](https://llmstxt.org/) convention: - [llms.txt](https://fluid.quantmind.com/llms.txt): an index of every page, each with a one-line description, so an agent can fetch only what it needs. - [llms-full.txt](https://fluid.quantmind.com/llms-full.txt): the whole documentation in a single file, API reference included. - [Recipes](https://fluid.quantmind.com/recipes/): a cheat sheet of the canonical patterns and the mistakes that are easy to make. Every page is also served as markdown, by appending `index.md` to its URL. The package ships `py.typed`, so a type checker resolves every signature in your editor and in your agent's context. For what to put in your agent's instructions file, see [Use with AI agents](https://fluid.quantmind.com/ai-agents/). If you are pointing an agent at this repository to contribute, read [AGENTS.md](https://github.com/quantmind/aio-fluid/blob/main/AGENTS.md). ## Development You can run the examples via ```text uv run python -m examples ``` We use [uv](https://uv.run/) as a development tool to run the examples and tests, but you can also use python directly if that's your preference. ## License This project is licensed under the BSD License - see the [LICENSE](https://github.com/quantmind/aio-fluid/blob/main/LICENSE) file for details. # 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. ```python 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: ```python 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](https://fluid.quantmind.com/tutorials/tasks/index.md) and [Task Dependencies](https://fluid.quantmind.com/tutorials/task_deps/index.md). ## 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](https://fluid.quantmind.com/tutorials/task_managers/index.md). ```python 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 ```python 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](https://fluid.quantmind.com/tutorials/task_fastapi/index.md). ## Command line entry point ```python 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](https://fluid.quantmind.com/tutorials/task_app/index.md). ## Schedule ```python @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 ```python @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](https://fluid.quantmind.com/tutorials/tasks/#cpu-bound-tasks) and [K8s Jobs](https://fluid.quantmind.com/tutorials/task_k8s/index.md). ## Shared resources ```python @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](https://fluid.quantmind.com/tutorials/task_deps/index.md). ## Retries, limits and timeouts ```python @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](https://fluid.quantmind.com/tutorials/task_retry/index.md). ## Queue work ```python 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](https://fluid.quantmind.com/tutorials/tasks/#chaining-tasks). ## Persist run history ```python task_manager.with_plugin(TaskDbPlugin(CrudDB.from_env())) ``` Adds a `/tasks-history` API backed by Postgres. Requires the `db` extra. See [Plugins](https://fluid.quantmind.com/tutorials/task_app/#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. # Use with AI agents Coding agents write a lot of the code that uses this library, so the documentation is published in a form they can consume directly, following the [llms.txt](https://llmstxt.org/) convention. This page is about pointing your agent at it. ## What is published | Resource | Size | When to use it | | ---------------------------------------------------------- | --------------- | ------------------------------------------------------------------------------- | | [recipes](https://fluid.quantmind.com/recipes/index.md) | ~1.5k tokens | Always. The canonical patterns and the mistakes that are easy to make | | [llms.txt](https://fluid.quantmind.com/llms.txt) | ~1k tokens | The index. Every page with a one-line description, to pick the one that matters | | any page, with `index.md` appended to its URL | 2k to 8k tokens | The specific topic at hand | | [llms-full.txt](https://fluid.quantmind.com/llms-full.txt) | ~94k tokens | A broad sweep, or one-shot ingestion into a vector store | The package also ships `py.typed`, so a type checker resolves every signature, which is the cheapest context an agent can have. ## Pointing your agent at it Put this in the instructions file your agent reads, `AGENTS.md`, `CLAUDE.md`, `.cursor/rules` or the equivalent, and adjust the last section to your project: ```markdown ## Background tasks (aio-fluid) This project uses [aio-fluid](https://fluid.quantmind.com/) for its task queue. Before writing or changing task code, read . It is short and covers the canonical patterns and the mistakes that are easy to make. For more detail: - indexes every page, fetch the one you need - any page is available as markdown by appending `index.md` to its URL, for example - is the entire documentation in one file, use it only when you need a broad sweep In this project: - tasks live in `myapp/tasks/`, registered in `myapp/app.py` - the entry point is `myapp/__main__.py`, a `TaskManagerCLI`, which is required because we have `cpu_bound` tasks - the API process runs a plain `TaskManager` (it only queues), the worker deployment runs the `TaskScheduler` ``` That last section is the part that pays off. The documentation cannot know which manager your processes run or where your tasks live, and those are exactly the things an agent will otherwise guess wrong. ## Prefer the recipes page to the full dump Feeding `llms-full.txt` into every session is the tempting mistake. It is a sizeable fraction of a context window spent on content that is mostly unrelated to the task at hand, and it crowds out your own code. The cheaper path is the [recipes](https://fluid.quantmind.com/recipes/index.md) page for the patterns, then [llms.txt](https://fluid.quantmind.com/llms.txt) to fetch the single page that covers whatever came up. ## Agents without web access When the agent cannot fetch URLs, vendor a copy into your repository and point the instructions at the local path: ```bash curl -o docs/vendor/aio-fluid-recipes.md https://fluid.quantmind.com/recipes/index.md ``` Refresh it when you upgrade the library. ## Contributing to aio-fluid The above is for applications that use the library. For an agent working on this repository itself, the conventions are in [AGENTS.md](https://github.com/quantmind/aio-fluid/blob/main/AGENTS.md). # Tutorials # Tasks Tasks are standard python async functions decorated with the @task decorator. ```python from fluid.scheduler import task, TaskRun @task async def say_hi(ctx: TaskRun) -> None: print("Hi!") ``` The TaskRun object is passed to the task function and contains the task metadata, including optional parameters, and the TaskManager. ## Task Parameters It is possible to pass parameters to the task, to do so, create a pydantic model for the task parameters ```python from pydantic import BaseModel class TaskParams(BaseModel): name: str ``` and pass it to the `task` decorator ```python from fluid.scheduler import task, TaskRun @task async def say_hi(ctx: TaskRun[TaskParams]) -> None: print(f"Hi {ctx.params.name}!") ``` ## Task Types There are few types of tasks implemented, lets take a look at them. ### IO Bound Tasks They run concurrently with the TaskConsumer. They must perform non blocking IO operations (no heavy CPU bound operations that blocks the event loop). ```python from fluid.scheduler import task, TaskRun from pydantic import BaseModel class Scrape(BaseModel): url: str = "https://" @task async def fecth_data(ctx: TaskRun[Scrape]) -> None: # fetch data data = await http_cli.get(ctx.params.url) data_id = await datastore_cli.stote(data) # trigger another task ctx.task_manager.queue("heavy_calculation", data_id=data_id) ``` ### CPU bound tasks They normally run on a subprocess and they can be defined by setting the `cpu_bound` flag to `True` in the task decorator. They can perform heavy CPU bound operations without blocking the event loop. ```python from fluid.scheduler import task, TaskRun @task(cpu_bound=True) async def heavy_calculation(ctx: TaskRun) -> None: data = await datastore_cli.get(ctx.params["data_id"]) # perform some heavy calculation ... # trigger another task ctx.task_manager.queue("fetch_data") ``` #### How it works When a CPU bound task is dispatched, the consumer spawns a **fresh Python subprocess** which runs the task through the `exec` command of the application command line client. This keeps the consumer's asyncio event loop completely unblocked while the subprocess runs. The command is derived from the one which started the consumer: the `serve` command is dropped, along with any option that belongs to it, and `exec ` is appended, with the run id and the task params passed as options. A [Kubernetes Job](https://fluid.quantmind.com/tutorials/task_k8s/index.md) derives its command from the consumer deployment in exactly the same way, so a task runs the same locally and in a cluster. It also means the entry point has to be a TaskManagerCLI: a consumer started any other way raises CpuBoundEntryPointError on startup. The subprocess is identified by the `TASK_MANAGER_SPAWN=true` environment variable. Inside it, `@task(cpu_bound=True)` behaves like a plain `@task`, so the executor function runs directly without any extra subprocess indirection. You can check whether your code is running inside a CPU subprocess: ```python from fluid.scheduler.common import is_in_cpu_process if is_in_cpu_process(): # running inside the spawned subprocess ... ``` Stdout and stderr from the subprocess are streamed back to the consumer in real time, so logs produced by the task appear in the consumer's output. A CPU bound task does not run in the consumer process, so it does not share the TaskManager instance the consumer is using. The process that executes the task builds its own task manager first, which is why an application with CPU bound tasks has to be set up with a command line entry point. See [Setup for CPU bound tasks](https://fluid.quantmind.com/tutorials/task_app/#setup-for-cpu-bound-tasks). #### Kubernetes When the consumer is running inside a Kubernetes cluster, CPU bound tasks can be dispatched as Kubernetes Jobs instead of local subprocesses. See [K8s Jobs](https://fluid.quantmind.com/tutorials/task_k8s/index.md) for more details. ### Scheduled Tasks Both IO and CPU bound tasks can be periodically scheduled via the `schedule` keyword argument. There are two types of scheduling, the most common is the every function that takes a `timedelta` object. ```python import asyncio from datetime import timedelta from fluid.scheduler import task, TaskRun, every @task(schedule=every(timedelta(seconds=1))) async def scheduled(ctx: TaskRun) -> None: await asyncio.sleep(0.1) ``` You can also use the crontab function to schedule tasks using cron expressions. ```python import asyncio from fluid.scheduler import task, TaskRun, crontab @task(schedule=crontab(hours='*/2')) async def scheduled(ctx: TaskRun) -> None: await asyncio.sleep(0.1) ``` ## Timeout All tasks, both IO and CPU bound, respect the `timeout_seconds` parameter (default **60 seconds**). The timeout is measured from when the task starts executing. For IO bound tasks, `asyncio` raises a `TimeoutError` if the coroutine has not completed within the timeout, and the task run transitions to the `failure` state. For CPU bound tasks, the subprocess (or Kubernetes Job) is killed and the run likewise transitions to `failure`. ```python from fluid.scheduler import task, TaskRun @task(timeout_seconds=300) async def slow_io_task(ctx: TaskRun) -> None: ... @task(cpu_bound=True, timeout_seconds=300) async def slow_cpu_task(ctx: TaskRun) -> None: ... ``` For long-running tasks make sure to raise `timeout_seconds` to an appropriate value. ## Concurrency control Use `max_concurrency` to limit how many instances of a task can run simultaneously. This applies to both IO and CPU bound tasks, and is useful to avoid overwhelming downstream services or exhausting system resources when many tasks are queued at once. ```python from fluid.scheduler import task, TaskRun @task(max_concurrency=5) async def fetch_data(ctx: TaskRun) -> None: ... @task(cpu_bound=True, max_concurrency=2) async def heavy_calculation(ctx: TaskRun) -> None: ... ``` A value of `0` (the default) means no limit. When the limit is reached the task run transitions to the `rate_limited` state. To automatically retry rate-limited tasks, combine `max_concurrency` with `rate_limit_retry`. See [Task Retry](https://fluid.quantmind.com/tutorials/task_retry/index.md) for details. ## Chaining tasks A task can queue another task with TaskRun.queue, which is how multi-step pipelines are built. Each step ends by queueing the next one and returns: ```python from pydantic import BaseModel from fluid.scheduler import TaskRun, TaskScheduler, task SYMBOLS = ("BTC-USD", "ETH-USD") class Symbol(BaseModel): symbol: str = "BTC-USD" @task async def daily_pipeline(ctx: TaskRun) -> None: """Start one chain per symbol.""" for symbol in SYMBOLS: await ctx.queue(extract, symbol=symbol) @task async def extract(ctx: TaskRun[Symbol]) -> None: """Download the raw data, then hand over to the next step.""" ctx.logger.info("extracting %s", ctx.params.symbol) # queue the next step and return, nothing blocks here. # if this task fails, transform is never queued await ctx.queue(transform, symbol=ctx.params.symbol) @task async def transform(ctx: TaskRun[Symbol]) -> None: """Normalise what extract downloaded.""" ctx.logger.info( "transforming %s, chain started by %s", ctx.params.symbol, ctx.root_run_id ) def task_scheduler() -> TaskScheduler: scheduler = TaskScheduler() scheduler.register_from_dict(globals()) return scheduler ``` Queueing rather than waiting is what you want here. The next step is a durable message in the broker, so a consumer restart does not lose the pipeline, any consumer in the fleet can pick the step up, and the parent does not sit idle occupying a run slot while the rest of the chain executes. Error handling falls out of the shape: a step that fails never reaches its `queue` call, so the chain stops there. Every run queued this way records where it came from: - from_run_id is the run that queued it. - root_run_id is the first run in the chain, shared by every run descending from it, so a whole pipeline can be retrieved in one query. Both are empty for a run that was not queued from another run, such as one started by a schedule or over HTTP. ### Waiting for a result When the caller genuinely needs the outcome, for example an HTTP handler that must return it in the response, use TaskConsumer.queue_and_wait instead. Note that it waits in memory on the local task manager, so the wait (not the queued run) is lost if that process restarts, and the caller holds its run slot for the duration. Prefer chaining for anything that looks like a pipeline. ## Aborting a task Any task, IO or CPU bound, can signal a deliberate, non-error cancellation by calling ctx.abort(): ```python from fluid.scheduler import task, TaskRun @task async def conditional_work(ctx: TaskRun) -> None: if not should_proceed(ctx.params): ctx.abort("precondition not met") ... ``` When this happens the task run transitions to the `aborted` TaskState, which is distinct from `failure`: - the event is logged at **info** level, not as an error - no retry policy is triggered - any registered abort handlers (e.g. the database plugin) are still notified ### CPU-bound tasks For CPU-bound tasks (subprocess or Kubernetes Job) the task function runs in a **separate process**, so the abort signal must be relayed back to the consumer. The mechanism works as follows: 1. The inner process calls `ctx.abort()`, which raises TaskAbortedError. 1. The consumer running *inside* that process catches the error and writes the reason to a short-lived Redis key (60-second TTL). 1. After the subprocess or k8s Job exits, the outer consumer reads the Redis key. If an abort reason is found it re-raises TaskAbortedError, marking the run as `aborted` instead of `success`. This means a CPU-bound task that aborts itself is always correctly reflected as `aborted` in the task run state, regardless of whether it ran locally or as a Kubernetes Job. # Task Managers Three classes can hold your tasks, and they form a chain where each one adds a single capability to the one before it: ```text TaskManager -> TaskConsumer -> TaskScheduler queue + execute + schedule ``` TaskConsumer is a TaskManager that also runs tasks, and TaskScheduler is a TaskConsumer that also fires schedules. Pick the smallest one that covers what the process has to do. | | TaskManager | TaskConsumer | TaskScheduler | | --------------------------------------------------- | ----------- | ------------ | ------------- | | Register tasks | ✅ | ✅ | ✅ | | Queue a task run on the broker | ✅ | ✅ | ✅ | | Execute a task inline, bypassing the queue | ✅ | ✅ | ✅ | | Is a Workers, with a start/stop lifecycle | ❌ | ✅ | ✅ | | Consumes the queue and runs what it finds | ❌ | ✅ | ✅ | | Fires `schedule=` tasks when they are due | ❌ | ❌ | ✅ | | Async event handlers, and the plugins built on them | ❌ | ✅ | ✅ | ## TaskManager The base class owns the pieces every application needs: the registry, the broker connection, the `deps` object and the plugin list. It can put work on the queue with TaskManager.queue, and it can run a task there and then with TaskManager.execute, which skips the queue and awaits the task on the current event loop. What it does not do is run anything in the background. It is not a worker: no coroutine of its own is ever started, so a task queued by a TaskManager sits on the broker until some consumer elsewhere picks it up. One consequence is easy to miss. TaskManager.register_async_handler is a no-op on the base class, because the worker that dispatches those events only exists on a consumer. Plugins built on lifecycle events, the database plugin among them, therefore record nothing when attached to a plain TaskManager. Attach them to the process that consumes the queue. Use it when the process submits work but must never spend its own capacity running it: an HTTP API in front of a separate worker fleet, a cron container that queues one run and exits, a test that drives a task directly. ## TaskConsumer TaskConsumer adds the machinery that executes queued work. It starts max_concurrent_tasks coroutine workers, each pulling one task run at a time from the broker, so that setting is the concurrency ceiling for the process. Around them it starts the worker that dispatches async events, the in-process queue that holds delayed runs until they are due, and a heartbeat that publishes the manager status other processes read through `GET /tasks-status`. max_concurrent_tasks is a field of TaskManagerConfig and defaults to 5. Pass it to the constructor, `TaskConsumer(max_concurrent_tasks=20)`, or set `FLUID_MAX_CONCURRENT_TASKS` to size a deployment without touching the code; an explicit argument wins over the environment variable. sleep_millis behaves the same way, while scheduler_heartbeat_millis is read from Settings alone, with no constructor argument. See [Settings](https://fluid.quantmind.com/reference/settings/index.md) for the full list and the naming rules. Being a worker, it starts and stops with whatever runs it, and TaskConsumer.queue_and_wait becomes available: queue a run and await its result. A consumer does not look at schedules. Registering a task declared with `schedule=every(...)` on a TaskConsumer gives you a task that is ready to run and never triggers, because nothing in the process is watching the clock. Something has to queue it, and that something is a scheduler. Use it for worker deployments you scale horizontally: every replica consumes the same queue, and running ten of them multiplies throughput without any of them duplicating work. ## TaskScheduler TaskScheduler adds one more worker, which ticks on a short heartbeat, asks the broker for the enabled tasks that have a schedule, and evaluates each every or crontab rule against the current time. A task that is due is queued on the broker like any other run. That last detail is what makes the design scale: the scheduler publishes to the shared queue, it does not execute the run itself. Any consumer on the same broker may pick it up, so the process that owns the clock is not the bottleneck. A scheduler consumes the queue as well, since it is a consumer, and a small deployment can be a single TaskScheduler doing both jobs. Run **one** scheduler per broker. Each scheduler keeps its own record of what it last fired, in memory, with no coordination between processes, so two schedulers on one broker queue every due task twice. Consumers are the part you replicate, not the scheduler. The two flags on TaskManagerConfig let you split the roles without changing class: ```python TaskScheduler(schedule_tasks=False) # behaves as a consumer TaskScheduler(consume_tasks=False) # schedules only, runs nothing ``` ## In a FastAPI app task_manager_fastapi accepts any of the three, and the choice decides what the app process does, because the task routes themselves are identical in every case. `POST /tasks/{name}` queues a run whichever manager is behind it. When the manager is a worker, a consumer or a scheduler, it is added to the app workers and starts and stops with the app: serving requests and running tasks happen in the same process. When it is a plain TaskManager, the app only gets startup and shutdown hooks, enough to open and close the resources registered with add_async_context_manager, and nothing consumes the queue. So the same tasks module can be served by two entry points, an API that only produces and a worker that schedules and executes: ```python from fastapi import FastAPI from examples.docs import task_deps from fluid.scheduler import TaskManager, TaskScheduler, task_manager_fastapi def api_app() -> FastAPI: """Frontend app, it queues task runs but never executes them. It registers the tasks because a task must be in the registry to be queued, but it needs no dependencies: nothing runs here. """ task_manager = TaskManager() task_manager.register_from_module(task_deps) return task_manager_fastapi(task_manager, title="Task producer") def worker_app() -> FastAPI: """Worker app, it schedules and executes the same tasks.""" deps = task_deps.Deps() scheduler = TaskScheduler(deps=deps) scheduler.add_async_context_manager(deps.http_client) scheduler.register_from_module(task_deps) return task_manager_fastapi(scheduler, title="Task worker") ``` Both entry points register the tasks, and they have to: a task must be in the registry to be queued by name, and a consumer that receives a run for a task it does not know logs an unknown task error. The dependencies, on the other hand, are only needed where tasks actually run, which is why the producer above builds none. Deploy that pair as one API deployment scaled for traffic and one worker deployment scaled for load, sharing a broker. For a single service doing both, pass a TaskScheduler and be done. See [Extending the FastAPI App](https://fluid.quantmind.com/tutorials/task_fastapi/index.md) for reaching whichever manager you chose from your own routes. ## Choosing - Only queueing work, or executing it inline in tests and scripts, use TaskManager. - Running queued work, with no schedules in the process, use TaskConsumer. - Owning the clock, and usually consuming too, use TaskScheduler. This is the default choice for a single-service application. Whichever you pick, an application with [CPU bound tasks](https://fluid.quantmind.com/tutorials/tasks/#cpu-bound-tasks) needs TaskManagerCLI as its entry point, because the subprocess that runs such a task builds its own task manager from it. See [Setup for CPU bound tasks](https://fluid.quantmind.com/tutorials/task_app/#setup-for-cpu-bound-tasks). # Task Queue App The `fluid.scheduler` module is a simple yet powerful distributed task producer (TaskScheduler) and consumer (TaskConsumer) system for executing tasks. The middleware for distributing tasks can be configured via the TaskBroker interface. A redis task broker is provided for convenience. ## Tasks Consumer Create a task consumer, register tasks from modules, and run the consumer. ```python import asyncio from typing import Any from fluid.scheduler import TaskConsumer import task_module_a, task_module_b def task_consumer(**kwargs: Any) -> TaskConsumer: consumer = TaskConsumer(**kwargs) consumer.register_from_module(task_module_a) consumer.register_from_module(task_module_b) return consumer if __name__ == "__main__": consumer = task_consumer() asyncio.run(consumer.run()) ``` Pass `tags` to register_from_module (or register_from_dict / register_task) to add extra tags to every registered task on top of the tags already declared on each task: ```python consumer.register_from_module(task_module_a, tags=["module-a"]) ``` ## FastAPI Integration A TaskManager can be integrated with FastAPI so that tasks can be queued via HTTP requests. To setup the FastAPI app, use the task_manager_fastapi function: ```python import uvicorn from fluid.scheduler import task_manager_fastapi if __name__ == "__main__": consumer = task_consumer() app = task_manager_fastapi(consumer) uvicorn.run(app) ``` You can test via the example provided ```bash $ python -m examples.simple_fastapi ``` and check the openapi UI at . The app returned is an ordinary FastAPI app: your own routes can be added to it and reach the task manager, its dependencies and its resources. See [Extending the FastAPI App](https://fluid.quantmind.com/tutorials/task_fastapi/index.md). The `GET /tasks` endpoint lists registered tasks and accepts a repeatable `tags` query parameter to only return tasks that have at least one of the given tags: ```text GET /tasks GET /tasks?tags=fast&tags=slow ``` ## Task App Command Line The TaskConsumer or TaskScheduler can be run with the command line tool to allow for an even richer API. ```python from fluid.scheduler.cli import TaskManagerCLI from fluid.scheduler import task_manager_fastapi if __name__ == "__main__": consumer = task_consumer() TaskManagerCLI(task_manager_fastapi(consumer))() ``` This features requires to install the package with the `cli` extra. ```bash $ pip install aio-fluid[cli] ``` ```bash $ python -m examples.simple_cli Usage: python -m examples.simple_cli [OPTIONS] COMMAND [ARGS]... Options: --help Show this message and exit. Commands: enable Enable or disable a task exec Execute a registered task ls List all tasks with their schedules serve Start app server. ``` The command line tool provides a powerful interface to execute tasks, parameters are passed as optional arguments using the standard click interface. ## Setup for CPU bound tasks For an application with [CPU bound tasks](https://fluid.quantmind.com/tutorials/tasks/#cpu-bound-tasks) the command line entry point above is not optional, it is how those tasks are executed. A CPU bound task does not run in the consumer process, so it does not share the TaskManager instance the consumer is using. The process that runs the task builds its own task manager first, through the `exec` command, which means everything the application attaches to the manager, dependencies and plugins in particular, has to be built again there. So build the task manager in the entry point, with the same dependencies and plugins the consumer uses, and expose it through TaskManagerCLI. A task then behaves the same whether it runs on the event loop or in a separate process, and the `cli` extra becomes a requirement rather than an option. The same entry point is what a [Kubernetes Job](https://fluid.quantmind.com/tutorials/task_k8s/index.md) runs when a CPU bound task is dispatched in a cluster, with the Job command derived from the consumer deployment. ## Plugins Plugins extend the task manager with additional behaviour by hooking into task lifecycle events. A plugin implements the TaskManagerPlugin interface and is registered via TaskManager.with_plugin. ### Database Plugin The TaskDbPlugin stores every task run in a database table so you can query task history, audit outcomes, and build dashboards on top of the data. It requires a CrudDB instance and the `db` extra: ```bash pip install aio-fluid[db] ``` Register the plugin when building your task manager: ```python from fluid.scheduler import TaskScheduler, task_manager_fastapi from fluid.scheduler.db import TaskDbPlugin from fluid.db import CrudDB db = CrudDB.from_env() task_manager = TaskScheduler(...) task_manager.with_plugin(TaskDbPlugin(db)) app = task_manager_fastapi(task_manager) ``` The plugin creates a `fluid_tasks` table (configurable via `table_name`) and persists a row for each task run as it moves through its lifecycle states. Tasks tagged with `skip_db` are excluded from persistence. The plugin mounts a `/tasks-history` router on the app with two endpoints: | Method | Path | Description | | ------ | ------------------------- | ------------------------------------------- | | `GET` | `/tasks-history` | List task run history with optional filters | | `GET` | `/tasks-history/{run_id}` | Fetch a single task run by ID | The list endpoint accepts the following query parameters: | Parameter | Type | Description | | --------- | ----------- | ------------------------------------------------ | | `name` | `string` | Filter by task name | | `state` | `TaskState` | Filter by task state (e.g. `success`, `failure`) | | `start` | `datetime` | Only runs queued at or after this time | | `end` | `datetime` | Only runs queued at or before this time | Example requests: ```bash # All history, most recent first GET /history # Only successful runs of the "add" task GET /history?name=add&state=success # Runs queued in a specific time window GET /history?start=2024-01-01T00:00:00Z&end=2024-01-02T00:00:00Z # Fetch a specific run by ID GET /history/abc123 ``` ### Custom Plugins To create your own plugin, subclass TaskManagerPlugin and implement the `register` method. Use TaskManager.register_async_handler to subscribe to task lifecycle events: ```python from fluid.scheduler import TaskManagerPlugin, TaskManager, TaskState from fluid.utils.dispatcher import Event class MyPlugin(TaskManagerPlugin): def register(self, task_manager: TaskManager) -> None: task_manager.register_async_handler( Event(TaskState.success, "my_plugin"), self._on_success, ) async def _on_success(self, task_run) -> None: print(f"Task {task_run.name} succeeded") ``` # Task Dependencies Production tasks rarely run in isolation: they need a database manager, an HTTP client, a cache. The TaskManager carries a single `deps` object for exactly this, and every task run can reach it. ## Passing dependencies Group what your tasks need into one object and pass it to the task manager. Any object will do, a dataclass is a good fit: ```python from dataclasses import dataclass, field from pydantic import BaseModel from fluid.scheduler import TaskRun, TaskScheduler, task from fluid.utils.http_client import HttpxClient @dataclass class Deps: """Dependencies shared by every task run.""" http_client: HttpxClient = field(default_factory=HttpxClient) class Quote(BaseModel): symbol: str = "BTC-USD" @task async def fetch_quote(ctx: TaskRun[Quote, Deps]) -> None: """Fetch a quote with the shared HTTP client.""" data = await ctx.deps.http_client.get( f"https://api.example.com/quotes/{ctx.params.symbol}" ) ctx.logger.info("got %s", data) def task_scheduler() -> TaskScheduler: deps = Deps() scheduler = TaskScheduler(deps=deps) # the client is opened on startup and closed on shutdown scheduler.add_async_context_manager(deps.http_client) scheduler.register_from_dict(globals()) return scheduler ``` Inside a task the dependencies are available as TaskRun.deps. Annotate the second type parameter of TaskRun to have them typed, as in `TaskRun[Quote, Deps]` above: without it `deps` is typed as `Any` and you get no completion or type checking. Both the params and the deps parameters are optional, so `TaskRun`, `TaskRun[Quote]` and `TaskRun[Quote, Deps]` are all valid annotations. ## Resource lifecycle Dependencies that hold a resource needing a startup and a shutdown, a connection pool for instance, should not be opened by each task. Register them with TaskManager.add_async_context_manager and the task manager enters them when it starts and exits them when it stops: ```python scheduler.add_async_context_manager(deps.http_client) ``` Routes served by the same task manager app share those resources with the task runs, see [Extending the FastAPI App](https://fluid.quantmind.com/tutorials/task_fastapi/index.md). ## Dependencies are not shared with CPU bound tasks A task declared with `cpu_bound=True` does not run in the consumer process. It is executed by a separate process, or by a Kubernetes Job in a cluster, which builds its own task manager from the command line entry point. Its dependencies are therefore constructed again, in that process, and nothing is shared with the consumer. Two consequences worth keeping in mind: - Dependencies must be cheap to construct, because the cost is paid on every run of a CPU bound task. - Anything held in memory by a dependency, a cache or an open connection, is not visible to a CPU bound task. Use the database or the broker to pass state across the process boundary. See [K8s Jobs](https://fluid.quantmind.com/tutorials/task_k8s/index.md) for how CPU bound tasks are dispatched. ## Defaults and plugins When no `deps` is passed the task manager creates an empty [State](https://www.starlette.io/applications/#storing-state-on-the-app-instance), the same namespace object starlette uses for `app.state`, so attributes can be set on it after construction: ```python scheduler = TaskScheduler() scheduler.deps.db_manager = db_manager ``` This works but is untyped and offers no protection against two components choosing the same attribute name. Prefer passing a typed object. `deps` belongs to your application. A separate `state` namespace, also a starlette [State](https://www.starlette.io/applications/#storing-state-on-the-app-instance), is reserved for plugins, which use it to store their own data on the task manager without colliding with your dependencies. See [Plugins](https://fluid.quantmind.com/tutorials/task_app/#plugins). # Extending the FastAPI App task_manager_fastapi returns a plain FastAPI app with the task routes mounted on it. It is meant to be extended: your own routes can live in the same app and reach the same TaskManager, with the same dependencies and the same resources the tasks use. ## One app for tasks and routes Pass your own app with the `app` argument and the task routes are added to it, rather than to a new one: ```python from typing import Annotated, cast from fastapi import APIRouter, Depends, FastAPI from examples.docs.task_deps import Deps, task_scheduler from fluid.scheduler import task_manager_fastapi from fluid.scheduler.endpoints import TaskManagerDep from fluid.utils.http_client import ResponseType def get_deps(task_manager: TaskManagerDep) -> Deps: """Typed access to the task manager dependencies.""" return cast(Deps, task_manager.deps) DepsDep = Annotated[Deps, Depends(get_deps)] router = APIRouter() @router.get("/quotes/{symbol}") async def get_quote(symbol: str, deps: DepsDep) -> ResponseType: """Fetch a quote with the same HTTP client the tasks use.""" return await deps.http_client.get(f"https://api.example.com/quotes/{symbol}") def scheduler_app() -> FastAPI: app = FastAPI(title="Quotes API") app.include_router(router) return task_manager_fastapi(task_scheduler(), app=app) ``` Building the app in a factory function keeps it usable from TaskManagerCLI, which is what runs the app and, for [CPU bound tasks](https://fluid.quantmind.com/tutorials/task_app/#setup-for-cpu-bound-tasks), what a task subprocess runs too. ## Reaching the task manager The task manager is stored on the app state as `app.state.task_manager`. Rather than reading the attribute, use the accessors in `fluid.scheduler.endpoints`: ```python from fluid.scheduler.endpoints import TaskManagerDep, get_task_manager ``` TaskManagerDep is an annotated FastAPI dependency, so a route asks for the task manager by declaring it as an argument: ```python @router.get("/queue-size") async def queue_size(task_manager: TaskManagerDep) -> dict[str, int]: return await task_manager.broker.queue_length() ``` get_task_manager(app) does the same outside a request, where there is an app but no request to depend on, in a test fixture or a startup hook for instance. ## Injecting the dependencies TaskManager.deps is typed as `Any`, because the library does not know what your application puts in it. Wrap the cast in a dependency of your own once, and every route gets the dependencies fully typed: ```python def get_deps(task_manager: TaskManagerDep) -> Deps: return cast(Deps, task_manager.deps) DepsDep = Annotated[Deps, Depends(get_deps)] ``` A route then declares `deps: DepsDep` and works with a `Deps` object, not with `Any`. See [Task Dependencies](https://fluid.quantmind.com/tutorials/task_deps/index.md) for how `deps` is built and passed to the task manager. The resources are shared, not merely visible. A route and a task run hold the same HTTP client instance, the same connection pool, the same cache. Anything registered with TaskManager.add_async_context_manager is entered when the app starts and exited when it stops, so by the time a route runs the client is already open, and there is nothing to open or close per request. This holds only for routes in the same process as the task manager. A `cpu_bound=True` task builds its own dependencies in its own process and shares nothing with the app, as described in [Dependencies are not shared with CPU bound tasks](https://fluid.quantmind.com/tutorials/task_deps/#dependencies-are-not-shared-with-cpu-bound-tasks). ## Plugin state Plugins keep their data in a separate namespace, TaskManager.state, so they never collide with the `deps` your application owns. A plugin that serves routes reads it back through the same task manager dependency. The database plugin does exactly this: ```python from fluid.scheduler.db import TaskDbPluginDep ``` so its `/tasks-history` routes reach the plugin without the application having to wire anything. See [Plugins](https://fluid.quantmind.com/tutorials/task_app/#plugins). ## Workers When the task manager is a worker, a TaskConsumer or a TaskScheduler, task_manager_fastapi adds it to the app workers, which start and stop with the app. `WorkersDep` from `fluid.tools_fastapi` gives a route access to that worker set, to report their status for a health endpoint for instance: ```python from fluid.tools_fastapi import WorkersDep ``` # Task Retries Tasks can be configured to retry automatically when they fail or when they cannot run due to concurrency limits. Both behaviours are controlled by a RetryPolicy attached to the task via the @task decorator. ## Retrying on failure Use the `retry` parameter to re-queue a task after an execution error. ```python from fluid.scheduler import RetryPolicy, task, TaskRun @task(retry=RetryPolicy(max_attempts=3, wait=2.0, backoff=2.0)) async def fetch_data(ctx: TaskRun) -> None: """Fetch data from an external API.""" response = await ctx.deps.http_client.get("https://api.example.com/data") ctx.logger.info("fetched %d bytes", len(response)) ``` If `fetch_data` raises, the TaskConsumer re-queues it with a delay and moves on. With `backoff=2.0` the wait times grow exponentially: `2s → 4s → 8s`. After 3 failed retries the TaskRun ends in the failure state. ### Limiting which exceptions trigger a retry By default all exceptions trigger a retry. Pass `exceptions` to be more selective: ```python @task(retry=RetryPolicy(max_attempts=5, wait=1.0, exceptions=(IOError, TimeoutError))) async def fetch_data(ctx: TaskRun) -> None: ... ``` `ValueError` or other programming errors will not be retried and the task fails immediately. ## Retrying when rate limited When max_concurrency is set, a task that cannot start because the limit is already reached ends in the rate_limited state by default. Set `rate_limit_retry` to re-queue it instead: ```python @task( max_concurrency=1, rate_limit_retry=RetryPolicy(max_attempts=10, wait=5.0), ) async def exclusive_job(ctx: TaskRun) -> None: """Only one instance of this task can run at a time.""" ... ``` The task will be re-queued up to 10 times, waiting 5 seconds between each attempt. If the slot is still occupied after all attempts, the run ends in rate_limited. ## Combining both policies A task can have both policies simultaneously: ```python @task( max_concurrency=2, retry=RetryPolicy(max_attempts=3, wait=2.0, backoff=2.0, max_wait=30.0), rate_limit_retry=RetryPolicy(max_attempts=5, wait=10.0), ) async def resilient_task(ctx: TaskRun) -> None: ... ``` ## How retries work under the hood Both retry paths use the same mechanism — no workers are blocked waiting: 1. The TaskConsumer detects the failure or concurrency limit. 1. It creates a fresh copy of the TaskRun with `execute_after` set to `now + delay`. 1. The copy is pushed back onto the Redis queue via the TaskBroker and the worker moves on immediately. 1. When a worker dequeues the copy and `execute_after` is still in the future, it schedules re-queuing via `call_later` and returns — still without blocking. 1. Once the delay has elapsed the task enters the queue normally and is executed. # K8s Jobs When the TaskConsumer runs inside a Kubernetes cluster, [CPU bound tasks](https://fluid.quantmind.com/tutorials/tasks/#cpu-bound-tasks) can be dispatched as [Kubernetes Jobs](https://kubernetes.io/docs/concepts/workloads/controllers/job/) instead of local subprocesses. This offloads heavy computation to dedicated pods and keeps the consumer event loop free. ## How it works The switch is automatic. When `KUBERNETES_SERVICE_HOST` is set (which Kubernetes injects into every pod) and the `k8s` extra is installed, any task declared with `cpu_bound=True` will spawn a Kubernetes Job instead of a subprocess. No code change is required in the task itself. The Job pod template is derived from the **task consumer deployment**. The implementation reads the deployment, locates the target container, and builds a Job spec from it. This means the Job inherits most of the container's configuration from the deployment — image, image pull policy, volume mounts, security context, and everything else — while only overriding the fields necessary to run the task. **Inherited from the deployment container (unchanged):** - Container image and image pull policy - Volume mounts (and pod-level volumes) - Environment variables (the task's env vars are appended, never replaced) - Security context - Everything else not listed below **Overridden or cleared:** | Field | Value in the Job | | ----------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `command` | Same as the deployment, but any trailing `serve` token is removed | | `args` | `exec --log --run-id --params ` | | `env` | Inherited from the deployment, then `TASK_MANAGER_SPAWN=true` appended, then any task-level `env` vars appended (see [Injecting environment variables](#injecting-environment-variables)) | | `resources` | Inherited from the deployment unless overridden via [`K8sConfig.resources`](#configuration) | | `liveness_probe` | Cleared — probes are not meaningful for Job pods and would prematurely kill long-running tasks | | `readiness_probe` | Cleared | **Other containers** (sidecars) from the deployment are dropped — only the target container runs in the Job pod. **Pod-level init containers** and **volumes** are preserved, so any setup performed at pod startup (e.g. installing TLS certificates) is reproduced in the Job pod. `TASK_MANAGER_SPAWN=true` signals to the process inside the Job that it is a CPU-bound worker rather than a long-lived consumer. The Job is created in the same namespace as the consumer with: - `backoff_limit: 0` — a failed pod is never retried; the error is propagated back to the task consumer instead - `ttlSecondsAfterFinished` — set from [`K8sConfig.job_ttl`](#configuration), the Job and its pods are cleaned up automatically after completion (default 300 s) - `restartPolicy: Never` on the pod template The job name is derived from the task name and the first 7 characters of the run ID, slugified and capped at 63 characters to comply with Kubernetes DNS label requirements: ```text task-- ``` Once the Job is created, the consumer polls its status every [`K8sConfig.sleep`](#configuration) seconds until it either succeeds or fails. ## Installation It requires both the `cli` and `k8s` extras: ```bash pip install aio-fluid[cli,k8s] ``` The `cli` extra is not optional here. The Job runs the deployment's own command with `exec ` as its arguments, so the container entry point must be a TaskManagerCLI. That CLI is what builds the application TaskManager, with its dependencies and plugins, inside the Job pod. See [Setup for CPU bound tasks](https://fluid.quantmind.com/tutorials/task_app/#setup-for-cpu-bound-tasks). ## Defining a CPU bound task ```python from fluid.scheduler import task, TaskRun @task(cpu_bound=True) async def heavy_calculation(ctx: TaskRun) -> None: # heavy CPU work here — runs in a k8s Job when inside a cluster, # or in a local subprocess when running outside one ... ``` ## Configuration K8s behaviour can be tuned per-task via the `k8s_config` argument, which accepts a K8sConfig object: ```python from fluid.scheduler import task, TaskRun, K8sConfig @task( cpu_bound=True, k8s_config=K8sConfig( namespace="workers", # namespace where the Job is created deployment="fluid-task", # deployment to copy the container spec from container="main", # container name inside the deployment job_ttl=600, # seconds to keep the Job after completion (default 300) sleep=2.0, # polling interval while waiting for the Job (default 2.0) resources={ # override the container's resource spec (default: inherited from deployment) "limits": {"cpu": "2", "memory": "4Gi"}, "requests": {"cpu": "1", "memory": "2Gi"}, }, ), ) async def heavy_calculation(ctx: TaskRun) -> None: ... ``` ### K8sConfig fields | Field | Type | Default | Description | | ------------ | --------------------------------- | ------------------------------------------------------ | -------------------------------------------------------------------------------------------------------------------- | | `namespace` | `str` | `FLUID_TASK_CONSUMER_K8S_NAMESPACE` or `"default"` | Kubernetes namespace where the Job is created | | `deployment` | `str` | `FLUID_TASK_CONSUMER_K8S_DEPLOYMENT` or `"fluid-task"` | Deployment to read the container spec from | | `container` | `str` | `FLUID_TASK_CONSUMER_K8S_CONTAINER` or `"main"` | Container name within the deployment | | `resources` | `K8sResourceRequirements \| None` | `None` | Resource limits/requests for the Job container. If `None`, the deployment's existing resource spec is used unchanged | | `job_ttl` | `int` | `FLUID_TASK_CONSUMER_K8S_JOB_TTL` or `300` | Seconds to retain the Job after completion before automatic cleanup | | `sleep` | `float` | `FLUID_TASK_CONSUMER_K8S_SLEEP` or `2.0` | Polling interval in seconds while waiting for the Job to finish | All `K8sConfig` fields have defaults drawn from environment variables, so a minimal deployment only needs to set those variables rather than hard-coding values per task. If `k8s_config` is omitted entirely, a K8sConfig instance with all defaults is used. ### Resource overrides The `resources` field accepts a K8sResourceRequirements dict with optional `limits` and `requests` keys: ```python resources={ "limits": {"cpu": "4", "memory": "8Gi"}, "requests": {"cpu": "500m", "memory": "1Gi"}, } ``` When not provided (the default), the Job container inherits the resource spec from the deployment container unchanged. This is useful for tasks that need more CPU or memory than the consumer pod is allocated. ## Injecting environment variables Extra environment variables can be injected into the Job (or subprocess, when running outside a cluster) using the `env` argument on the @task decorator: ```python from fluid.scheduler import task, TaskRun @task( cpu_bound=True, env={"MODEL_PATH": "/mnt/models/v2", "LOG_LEVEL": "DEBUG"}, ) async def heavy_calculation(ctx: TaskRun) -> None: import os model_path = os.environ["MODEL_PATH"] ... ``` These variables are appended to the environment after the deployment's existing env vars and `TASK_MANAGER_SPAWN=true`, so they can override anything set in the deployment if needed. For subprocess execution (outside a cluster), they are merged into the spawned process's environment the same way, making task definitions portable across both runtimes without any conditional logic. ## Required RBAC permissions The pod running the TaskConsumer or the TaskScheduler needs permission to read the deployment and create/read jobs. Assuming the consumer/scheduler runs in the `workers` namespace, a minimal `Role` and `RoleBinding` can be defined as follows: ```yaml apiVersion: rbac.authorization.k8s.io/v1 kind: Role metadata: name: execute-jobs-role namespace: workers rules: - apiGroups: - apps resources: - deployments verbs: - get - list - watch - apiGroups: - batch resources: - jobs - cronjobs - jobs/status verbs: - create - get - list - watch - delete - patch - update --- apiVersion: v1 kind: ServiceAccount metadata: name: tasks-consumer-sa namespace: workers --- apiVersion: rbac.authorization.k8s.io/v1 kind: RoleBinding metadata: name: tasks-consumer-rb namespace: workers subjects: - kind: ServiceAccount name: tasks-consumer-sa roleRef: kind: Role name: execute-jobs-role apiGroup: rbac.authorization.k8s.io ``` The `tasks-consumer-sa` ServiceAccount should be used by the consumer/scheduler deployment. # Task Broker A TaskBroker is responsible for queuing task runs and storing task information. A subclass has to implement every abstract method below, grouped here by the concern each one serves. ```python # queues @property @abstractmethod def task_queue_names(self) -> tuple[str, ...]: """Names of the task queues""" @abstractmethod async def queue_task(self, task_run: TaskRun) -> None: """Queue a task run""" @abstractmethod async def get_task_run(self, task_manager: TaskManager) -> TaskRun | None: """Get a Task run from the task queue""" @abstractmethod async def queue_length(self) -> dict[str, int]: """Length of task queues""" @abstractmethod async def clear_queue(self, *priorities: TaskPriority) -> dict[str, int]: """Clear task queues, returns number of removed items per priority""" # task information @abstractmethod async def get_tasks_info(self, *task_names: str) -> list[TaskInfo]: """List of TaskInfo objects""" @abstractmethod async def update_task(self, task: Task, params: dict[str, Any]) -> TaskInfo: """Update a task dynamic parameters""" # in-flight task runs, used to enforce max_concurrency @abstractmethod async def add_task_run(self, task_run: TaskRun) -> None: """Add a task run to the broker""" @abstractmethod async def remove_task_run(self, task_run: TaskRun) -> None: """Remove a task run from the broker""" @abstractmethod async def current_task_runs(self, task_name: str) -> int: """The number of current task runs for a given task_name""" # aborts @abstractmethod async def set_task_aborted(self, run_id: str, reason: str) -> None: """Signal that a task run was aborted, storing the reason""" @abstractmethod async def get_task_aborted(self, run_id: str) -> str | None: """Return the abort reason for a task run, or None if not aborted""" # task manager status @abstractmethod async def set_manager_status(self, manager_id: str, data: dict, ttl: int) -> None: """Store the status of a running task manager""" @abstractmethod async def get_all_manager_statuses(self) -> TaskManagersStatus: """Get statuses of all running task managers""" # locking and shutdown @abstractmethod def lock(self, name: str, timeout: float | None = None) -> Lock: """Create a lock""" @abstractmethod async def close(self) -> None: """Close the broker on shutdown""" ``` The return type of `lock` is currently the lock class of the `redis` client, so a broker built on a different backend has to return an object satisfying that type. Only the async context manager protocol is used at runtime. The library ships a Redis broker for convenience. ```python from fluid.scheduler import TaskBroker redis_broker = TaskBroker.from_url("redis://localhost:6379") ``` By default the broker uses the url provided in the `FLUID_BROKER_URL` environment variable and falls back to `redis://localhost:6379`. ```python broker = TaskBroker.from_url() str(broker.url) == "redis://localhost:6379" ``` # Workers Workers are the main building block for asynchronous programming with `aio-fluid`. They are responsible for running tasks and managing their lifecycle. All workers implemented derive from the base abstract class Worker where the main method to implement is the Worker.run method. ## Worker Lifecycle The lifecycle of a worker is managed by the WorkerState class which provides a set of states that a worker can be in. The worker starts in an inital state and than it can be started and stopped. ### Startup To start a worker one uses the async method Worker.startup which create the task running the worker. The task will transition the worker from WorkerState.INIT to the WorkerState.RUNNING state. The worker will then run the Worker.on_startup coroutine method (which by default is a no-op) follow by the main worker coroutine method Worker.run method until it is stopped. This is a very simple example of a worker that prints a message every second until it is stopped: ```python import asyncio from fluid.utils.worker import Worker class SimpleWorker(Worker): async def run(self): while self.is_running(): self.print_message() await asyncio.sleep(1) def print_message(self): print(f"Hello from {self.worker_name} in state {self.worker_state}") async def main(): worker = SimpleWorker() worker.print_message() await worker.startup() asyncio.get_event_loop().call_later(5, worker.gracefully_stop) await worker.wait_for_shutdown() worker.print_message() if __name__ == "__main__": asyncio.run(main()) ``` ### Shutdown To shut down a worker there are few possibilities. - Direct call to the async Worker.shutdown method which will trigger the graceful shutdown and wait for the worker to finish its work. - Call the Worker.gracefully_stop method which will trigger the graceful shutdown. Importantly, this method does not wait for the worker to finish its work, ti simply transition from the WorkerState.RUNNING to WorkerState.STOPPING state. To wait for the worker exit one should call the async Worker.wait_for_shutdown method (as in the example above) ## Async Context Manager Worker implements the async context manager protocol. Entering the context calls Worker.startup and exiting it calls Worker.shutdown, so the `async with` pattern is the most concise way to manage the full lifecycle: ```python async with MyWorker() as worker: # worker is running here ... # worker is fully shut down here ``` Resources that the worker needs for its entire lifetime can be opened and closed inside Worker.run using normal `async with` statements — no subclassing of lifecycle hooks is required. # Async Database The `fluid.db` module provides a simple asynchronous interface to interact with postgres databases. It is built on top of the [sqlalchemy](https://www.sqlalchemy.org/) and [asyncpg](https://github.com/MagicStack/asyncpg) libraries. ## Installation To use the database module, you need to install the `db` extra, and optionally the `cli` extra if you want to use the command line interface for managing database migrations: ```bash pip install aio-fluid[db,cli] ``` ## Database There are two database classes: - Database — provides connection management, transactions, and migrations. - CrudDB — extends Database with CRUD helpers for common query patterns. Most applications should use CrudDB directly: ```python from fluid.db import CrudDB db = CrudDB("postgresql+asyncpg://postgres:postgres@localhost:5432/mydb") ``` Note the use of the `postgresql+asyncpg` driver in the connection string. This is required for the async engine. You can also load the DSN from an environment variable (defaults to `DATABASE`): ```python db = CrudDB.from_env() ``` ## Register a Table Register tables against the database's `metadata` so that migrations and CRUD helpers can discover them: ```python import sqlalchemy as sa articles = sa.Table( "articles", db.metadata, sa.Column("id", sa.Integer, primary_key=True, autoincrement=True), sa.Column("title", sa.String(200), nullable=False), sa.Column("author", sa.String(100)), sa.Column("score", sa.Integer), sa.Column("published_at", sa.DateTime(timezone=True)), ) ``` ## Migrations The Migration object wraps Alembic and is obtained from the database: ```python mig = db.migration() ``` ### Create the database ```python # creates the database if it doesn't exist yet; returns False if it already existed mig.db_create() ``` ### Apply migrations ```python # upgrade to the latest revision (equivalent to `alembic upgrade heads`) mig.upgrade("heads") ``` For quick setup in tests or development, you can create all tables directly from metadata without Alembic revision files: ```python mig.create_all() ``` ### Generate a new revision ```python # auto-generate a revision by diffing metadata against the current schema mig.revision("add score column", autogenerate=True) ``` ## Connections and Transactions Use `connection()` when you only need to read data: ```python async with db.connection() as conn: result = await conn.execute(sa.text("SELECT 1")) ``` Use `transaction()` when you need to write data — it commits on exit and rolls back on exception: ```python async with db.transaction() as conn: await conn.execute(articles.insert().values(title="Hello")) ``` The `ensure_connection` and `ensure_transaction` variants are useful in functions that may receive an existing connection from the caller, avoiding nested connections: ```python async def insert_article(data: dict, conn=None): async with db.ensure_transaction(conn) as conn: await conn.execute(articles.insert().values(**data)) ``` ## CRUD Operations CrudDB provides async helpers that cover the most common patterns. ### Insert ```python # single row row = (await db.db_insert(articles, {"title": "Hello", "score": 10})).one() # multiple rows — missing columns are filled with None automatically rows = (await db.db_insert(articles, [ {"title": "Hello", "score": 10}, {"title": "World"}, ])).fetchall() ``` All insert operations return the inserted rows via `RETURNING *`. ### Select ```python rows = (await db.db_select(articles, {"author": "alice"})).fetchall() ``` Pass `order_by` to sort results. Prefix a field name with `-` for descending order: ```python rows = (await db.db_select(articles, {}, order_by=("-published_at",))).fetchall() ``` #### Filter operators Filters use a `"field:op"` key syntax. The default operator is `eq`: | Key | Meaning | | ------------------------- | ---------------- | | `"score"` or `"score:eq"` | `score = value` | | `"score:ne"` | `score != value` | | `"score:gt"` | `score > value` | | `"score:ge"` | `score >= value` | | `"score:lt"` | `score < value` | | `"score:le"` | `score <= value` | Pass a list as the value to use `IN` / `NOT IN`: ```python # WHERE score IN (5, 10, 15) rows = (await db.db_select(articles, {"score": [5, 10, 15]})).fetchall() ``` ### Update ```python rows = (await db.db_update(articles, {"author": "alice"}, {"score": 99})).fetchall() ``` Returns all updated rows via `RETURNING *`. ### Upsert `db_upsert` updates a single record if it exists, or inserts it if it does not: ```python # update score if the row exists, otherwise insert it row = await db.db_upsert( articles, {"title": "Hello"}, # lookup key {"score": 42}, # data to set ) ``` ### Delete ```python deleted = (await db.db_delete(articles, {"author": "alice"})).fetchall() ``` Returns all deleted rows via `RETURNING *`. ### Count ```python n = await db.db_count(articles, {"author": "alice"}) ``` ## Pagination Pagination implements cursor-based pagination on top of CrudDB. It fetches one extra row beyond the requested limit to determine whether a next page exists, then encodes a cursor that the client returns with the next request. ```python from fluid.db import Pagination # first page rows, cursor = await Pagination.create( "published_at", "id", limit=20, filters={"author": "alice"}, desc=True, ).execute(db, articles) # next page — filters and limit are embedded in the cursor if cursor: rows, cursor = await Pagination.create( "published_at", "id", cursor=cursor, desc=True, ).execute(db, articles) ``` When `cursor` is provided, the `filters` and `limit` arguments are ignored — they are decoded from the cursor itself, ensuring consistent pages even if the caller changes them between requests. ### The ordering must be unique The ordering fields must uniquely identify a row. The cursor encodes only the values of those fields for the first row of the next page, and the next query resumes from that position, so ties are ambiguous: rows sharing the same ordering values can appear on two consecutive pages, and rows can be skipped altogether, because the database is free to order tied rows differently between the two queries. `published_at` alone is not unique — two articles can be published at the same instant — which is why the examples above order by `("published_at", "id")`. Appending the primary key makes the ordering total and the cursor unambiguous. The same rule applies to the ordering used with full-text search below. ### Full-text search Combine pagination with a full-text search across multiple columns: ```python from fluid.db import Pagination from fluid.db.pagination import Search rows, cursor = await Pagination.create( "published_at", "id", limit=20, search=Search(search_fields=("title", "author"), search_text="fluid"), desc=True, ).execute(db, articles) ``` # Event Dispatchers [Event dispatchers](https://fluid.quantmind.com/reference/dispatchers/index.md) are a way to decouple the event source from the event handler. This is useful when you want to have multiple handlers for the same event, or when you want to have a single handler for multiple events. ## A Simple Dispatcher In this example we will create a simple dispatcher that will dispatch strings to a list of handlers. The only requirement for the implementation of a Dispatcher is to implement the `event_type` method. ```python from fluid.utils.dispatcher import Dispatcher class SimpleDispatcher(Dispatcher[str]): def event_type(self, message: str) -> str: return "*" simple = SimpleDispatcher() assert simple.dispatch("you can dispatch strings to this dispatcher") == 0 def count_words(x: str) -> None: words = [x.strip() for w in x.split(" ") if w.strip()] print(f"number of words {len(words)}") simple.register_handler("*", count_words) assert simple.dispatch("you can dispatch strings to this dispatcher") == 1 def count_letters(x: str) -> None: letters = set(x) print(f"number of letters {len(letters)}") simple.register_handler("*.count_letters", count_letters) assert simple.dispatch("you can dispatch strings to this dispatcher") == 2 ``` In this example we have a simple dispatcher that will dispatch strings to a list of handlers. The `event_type` method returns the type of the event, in this case always a "\*" string. The registration of multiple handlers is done via the use of tags (see the `count_letters` registration). ## A Data Dispatcher In this example we will create a dispatcher that will dispatch data to a list of handlers. The event type of the message is given by the type of the data. ```python from typing import Any from fluid.utils.dispatcher import Dispatcher class MessageDispatcher(Dispatcher[Any]): def event_type(self, data: Any) -> str: return type(data).__name__ dispatcher = MessageDispatcher() assert dispatcher.dispatch({}) == 0 def count_keys(data: Any) -> None: print(f"number of keys {len(data)}") dispatcher.register_handler("dict", count_keys) assert dispatcher.dispatch(dict(a=1, b=2)) == 1 ``` # API reference # Reference Complete API reference for all public classes, functions, and parameters in Aio Fluid. See the [home page](https://fluid.quantmind.com/index.md) for installation instructions. ## Workers [Workers](https://fluid.quantmind.com/reference/workers/index.md) — base async worker types with start/stop lifecycle (`Workers`, `WorkerFunction`, `AsyncConsumer`). ## Task Scheduler - [Task](https://fluid.quantmind.com/reference/task/index.md) — the `@task` decorator, `Task`, `TaskPriority`, `TaskState`, and `K8sConfig`. - [Task Run](https://fluid.quantmind.com/reference/task_run/index.md) — `TaskRun`, the context object passed to every task executor. - [Task Retry](https://fluid.quantmind.com/reference/task_retry/index.md) — `RetryPolicy` for failure retries and rate-limit retries. - [Task Scheduling](https://fluid.quantmind.com/reference/task_scheduling/index.md) — `every()` and `crontab()` schedule helpers. - [Task Manager](https://fluid.quantmind.com/reference/task_manager/index.md) — `TaskManager`, the base class for running and queuing tasks. - [Task Consumer](https://fluid.quantmind.com/reference/task_consumer/index.md) — `TaskConsumer`, the worker that dequeues and executes tasks. - [Task Scheduler](https://fluid.quantmind.com/reference/task_scheduler/index.md) — `TaskScheduler`, combines consumer and scheduler. - [Task Broker](https://fluid.quantmind.com/reference/task_broker/index.md) — `TaskBroker` interface and the Redis implementation. - [Task Manager Plugins](https://fluid.quantmind.com/reference/task_plugin/index.md) — extend `TaskManager` with lifecycle hooks. - [Task Registry](https://fluid.quantmind.com/reference/task_registry/index.md) — internal registry that maps task names to `Task` objects. - [Task Manager CLI](https://fluid.quantmind.com/reference/task_cli/index.md) — command-line tools for `TaskManager` applications. ## Database - [Database](https://fluid.quantmind.com/reference/db/index.md) — async Postgres connection and query interface. - [CrudDB](https://fluid.quantmind.com/reference/db_crud/index.md) — CRUD operations on top of `Database`. - [DB Migration](https://fluid.quantmind.com/reference/db_migrations/index.md) — schema migration management. - [DB Pagination](https://fluid.quantmind.com/reference/db_pagination/index.md) — paginated query results. - [DB CLI](https://fluid.quantmind.com/reference/db_cli/index.md) — command-line tools for database management. ## Utilities - [Event Dispatchers](https://fluid.quantmind.com/reference/dispatchers/index.md) — `Dispatcher` and `AsyncDispatcher` for decoupled event handling. - [HTTP Client](https://fluid.quantmind.com/reference/http_client/index.md) — unified async HTTP client wrappers for `aiohttp` and `httpx`. - [Errors](https://fluid.quantmind.com/reference/errors/index.md) — error hierarchies for utilities and the task scheduler. - [Settings](https://fluid.quantmind.com/reference/settings/index.md), environment variables configuring the task consumer, broker, database and HTTP client. - [Utils](https://fluid.quantmind.com/reference/utils/index.md) — miscellaneous helpers. # Database It can be imported from `fluid.db`: It requires the `db` extra to be installed: ```bash pip install aio-fluid[db] ``` ```python from fluid.db import Database ``` ## fluid.db.Database ```python Database( dsn, echo=(lambda: DBECHO)(), pool_size=(lambda: DBPOOL_MAX_SIZE)(), max_overflow=(lambda: DBPOOL_MAX_OVERFLOW)(), metadata=MetaData(), migration_path="", app_name=(lambda: APP_NAME)(), _engine=None, ) ``` A container for tables in a database and a manager of asynchronous connections to a postgresql database ### dsn ```python dsn ``` data source name, aka connection string Example: `postgresql+asyncpg://user:password@localhost/dbname` Note that the `+asyncpg` part is important for the async engine. Currently, only `postgresql+asyncpg` is supported, but other databases may be supported in the future. ### echo ```python echo = field(default_factory=lambda: settings.DBECHO) ``` Echo SQL queries to stdout It defaults to the `DBECHO` setting in the settings module ### pool_size ```python pool_size = field( default_factory=lambda: settings.DBPOOL_MAX_SIZE ) ``` ### max_overflow ```python max_overflow = field( default_factory=lambda: settings.DBPOOL_MAX_OVERFLOW ) ``` ### metadata ```python metadata = field(default_factory=sa.MetaData) ``` ### migration_path ```python migration_path = '' ``` Path to the directory containing migration files. If empty, migrations will be stored in the default location `migrations` in the current working directory. ### app_name ```python app_name = field(default_factory=lambda: settings.APP_NAME) ``` ### tables ```python tables ``` A dictionary of tables in the database ### engine ```python engine ``` The :class:`sqlalchemy.ext.asyncio.AsyncEngine` creating connection and transactions ### url ```python url ``` The SQLAlchemy URL object for the database ### sync_engine ```python sync_engine ``` The sqlalchemy Engine object for synchrouns operations ### from_env ```python from_env( *, dsn=None, schema=None, migration_path=None, app_name=None, max_overflow=None, pool_size=None, db_name=None ) ``` Create a new database container from environment variables as defaults Source code in `fluid/db/container.py` ```python @classmethod def from_env( cls, *, dsn: str | None = None, schema: str | None = None, migration_path: str | Path | None = None, app_name: str | None = None, max_overflow: int | None = None, pool_size: int | None = None, db_name: str | None = None, ) -> Self: """Create a new database container from environment variables as defaults""" if dsn is None: dsn = settings.DATABASE if schema is None: schema = settings.DATABASE_SCHEMA kwargs = compact_dict( migration_path=migration_path, app_name=app_name, max_overflow=max_overflow, pool_size=pool_size, ) if db_name: dsn = ( make_url(dsn) .set(database=db_name) .render_as_string(hide_password=False) ) return cls(dsn=dsn, metadata=sa.MetaData(schema=schema), **kwargs) ``` ### connection ```python connection() ``` Context manager for obtaining an asynchronous connection Source code in `fluid/db/container.py` ```python @asynccontextmanager async def connection(self) -> AsyncIterator[AsyncConnection]: """Context manager for obtaining an asynchronous connection""" async with self.engine.connect() as conn: yield conn ``` ### ensure_connection ```python ensure_connection(conn=None) ``` Context manager for obtaining an asynchronous connection Source code in `fluid/db/container.py` ```python @asynccontextmanager async def ensure_connection( self, conn: AsyncConnection | None = None, ) -> AsyncIterator[AsyncConnection]: """Context manager for obtaining an asynchronous connection""" if conn: yield conn else: async with self.engine.connect() as conn: yield conn ``` ### transaction ```python transaction() ``` Context manager for initializing an asynchronous database transaction Source code in `fluid/db/container.py` ```python @asynccontextmanager async def transaction(self) -> AsyncIterator[AsyncConnection]: """Context manager for initializing an asynchronous database transaction""" async with self.engine.begin() as conn: yield conn ``` ### ensure_transaction ```python ensure_transaction(conn=None) ``` Context manager for ensuring we a connection has initialized a database transaction Source code in `fluid/db/container.py` ```python @asynccontextmanager async def ensure_transaction( self, conn: AsyncConnection | None = None, ) -> AsyncIterator[AsyncConnection]: """Context manager for ensuring we a connection has initialized a database transaction""" if conn: if not conn.in_transaction(): async with conn.begin(): yield conn else: yield conn else: async with self.transaction() as conn: yield conn ``` ### close ```python close() ``` Close the asynchronous db engine if opened Source code in `fluid/db/container.py` ```python async def close(self) -> None: """Close the asynchronous db engine if opened""" if self._engine is not None: engine, self._engine = self._engine, None await engine.dispose() ``` ### ping ```python ping() ``` Ping the database Source code in `fluid/db/container.py` ```python async def ping(self) -> str: """Ping the database""" # TODO: we need a custom ping query async with self.connection() as conn: await conn.execute(sa.text("SELECT 1")) return "ok" ``` ### migration ```python migration() ``` The migration manager for this database Source code in `fluid/db/container.py` ```python def migration(self) -> Migration: """The migration manager for this database""" return Migration(db=self) ``` # DB CLI The Database command line interface (CLI) is a tool for managing database migrations and other database operations. It requires to install the additional `cli` extra: ```bash pip install aio-fluid[db,cli] ``` It can be imported from `fluid.db.cli`: ```python from fluid.db.cli import DbGroup ``` ## fluid.db.cli.DbGroup ```python DbGroup( db, name="db", help="Manage database and migrations", **kwargs ) ``` Bases: `Group` A click group for database commands This class provides a CLI for a Database Application. It requires to install the `cli` extra dependencies. Source code in `fluid/db/cli.py` ```python def __init__( self, db: Database, name: str = "db", help: str = "Manage database and migrations", # noqa: A002 **kwargs: Any, ) -> None: super().__init__(name=name, help=help, **kwargs) self.db = db for command in _db.commands.values(): self.add_command(command) ``` ### db ```python db = db ``` ### get_command ```python get_command(ctx, name) ``` Source code in `fluid/db/cli.py` ```python def get_command(self, ctx: click.Context, name: str) -> Optional[click.Command]: ctx.obj = {"db": self.db} return super().get_command(ctx, name) ``` ### list_commands ```python list_commands(ctx) ``` Source code in `fluid/db/cli.py` ```python def list_commands(self, ctx: click.Context) -> list[str]: ctx.obj = {"db": self.db} return super().list_commands(ctx) ``` # CrudDB The CrudDB class inherits from Database to provide standard CRUD operations for a database table. It requires the `db` extra to be installed: ```bash pip install aio-fluid[db] ``` It can be imported from `fluid.db`: ```python from fluid.db import CrudDB ``` ## fluid.db.CrudDB ```python CrudDB( dsn, echo=(lambda: DBECHO)(), pool_size=(lambda: DBPOOL_MAX_SIZE)(), max_overflow=(lambda: DBPOOL_MAX_OVERFLOW)(), metadata=MetaData(), migration_path="", app_name=(lambda: APP_NAME)(), _engine=None, ) ``` Bases: `Database` A Database with additional methods for CRUD operations ### dsn ```python dsn ``` data source name, aka connection string Example: `postgresql+asyncpg://user:password@localhost/dbname` Note that the `+asyncpg` part is important for the async engine. Currently, only `postgresql+asyncpg` is supported, but other databases may be supported in the future. ### echo ```python echo = field(default_factory=lambda: settings.DBECHO) ``` Echo SQL queries to stdout It defaults to the `DBECHO` setting in the settings module ### pool_size ```python pool_size = field( default_factory=lambda: settings.DBPOOL_MAX_SIZE ) ``` ### max_overflow ```python max_overflow = field( default_factory=lambda: settings.DBPOOL_MAX_OVERFLOW ) ``` ### metadata ```python metadata = field(default_factory=sa.MetaData) ``` ### migration_path ```python migration_path = '' ``` Path to the directory containing migration files. If empty, migrations will be stored in the default location `migrations` in the current working directory. ### app_name ```python app_name = field(default_factory=lambda: settings.APP_NAME) ``` ### tables ```python tables ``` A dictionary of tables in the database ### engine ```python engine ``` The :class:`sqlalchemy.ext.asyncio.AsyncEngine` creating connection and transactions ### url ```python url ``` The SQLAlchemy URL object for the database ### sync_engine ```python sync_engine ``` The sqlalchemy Engine object for synchrouns operations ### from_env ```python from_env( *, dsn=None, schema=None, migration_path=None, app_name=None, max_overflow=None, pool_size=None, db_name=None ) ``` Create a new database container from environment variables as defaults Source code in `fluid/db/container.py` ```python @classmethod def from_env( cls, *, dsn: str | None = None, schema: str | None = None, migration_path: str | Path | None = None, app_name: str | None = None, max_overflow: int | None = None, pool_size: int | None = None, db_name: str | None = None, ) -> Self: """Create a new database container from environment variables as defaults""" if dsn is None: dsn = settings.DATABASE if schema is None: schema = settings.DATABASE_SCHEMA kwargs = compact_dict( migration_path=migration_path, app_name=app_name, max_overflow=max_overflow, pool_size=pool_size, ) if db_name: dsn = ( make_url(dsn) .set(database=db_name) .render_as_string(hide_password=False) ) return cls(dsn=dsn, metadata=sa.MetaData(schema=schema), **kwargs) ``` ### connection ```python connection() ``` Context manager for obtaining an asynchronous connection Source code in `fluid/db/container.py` ```python @asynccontextmanager async def connection(self) -> AsyncIterator[AsyncConnection]: """Context manager for obtaining an asynchronous connection""" async with self.engine.connect() as conn: yield conn ``` ### ensure_connection ```python ensure_connection(conn=None) ``` Context manager for obtaining an asynchronous connection Source code in `fluid/db/container.py` ```python @asynccontextmanager async def ensure_connection( self, conn: AsyncConnection | None = None, ) -> AsyncIterator[AsyncConnection]: """Context manager for obtaining an asynchronous connection""" if conn: yield conn else: async with self.engine.connect() as conn: yield conn ``` ### transaction ```python transaction() ``` Context manager for initializing an asynchronous database transaction Source code in `fluid/db/container.py` ```python @asynccontextmanager async def transaction(self) -> AsyncIterator[AsyncConnection]: """Context manager for initializing an asynchronous database transaction""" async with self.engine.begin() as conn: yield conn ``` ### ensure_transaction ```python ensure_transaction(conn=None) ``` Context manager for ensuring we a connection has initialized a database transaction Source code in `fluid/db/container.py` ```python @asynccontextmanager async def ensure_transaction( self, conn: AsyncConnection | None = None, ) -> AsyncIterator[AsyncConnection]: """Context manager for ensuring we a connection has initialized a database transaction""" if conn: if not conn.in_transaction(): async with conn.begin(): yield conn else: yield conn else: async with self.transaction() as conn: yield conn ``` ### close ```python close() ``` Close the asynchronous db engine if opened Source code in `fluid/db/container.py` ```python async def close(self) -> None: """Close the asynchronous db engine if opened""" if self._engine is not None: engine, self._engine = self._engine, None await engine.dispose() ``` ### ping ```python ping() ``` Ping the database Source code in `fluid/db/container.py` ```python async def ping(self) -> str: """Ping the database""" # TODO: we need a custom ping query async with self.connection() as conn: await conn.execute(sa.text("SELECT 1")) return "ok" ``` ### migration ```python migration() ``` The migration manager for this database Source code in `fluid/db/container.py` ```python def migration(self) -> Migration: """The migration manager for this database""" return Migration(db=self) ``` ### db_select ```python db_select(table, filters, *, order_by=None, conn=None) ``` Select rows from a given table | PARAMETER | DESCRIPTION | | ---------- | ---------------------------------------------------------------------------------------------------------------------- | | `table` | The table to select from **TYPE:** `FromClause` | | `filters` | Key-value pairs for filtering rows; supports 'field:op' syntax for operators (eq, ne, gt, ge, lt, le) **TYPE:** `dict` | | `order_by` | Column names to order by; prefix with '-' for descending **TYPE:** \`tuple[str, ...] | | `conn` | Optional existing connection to reuse **TYPE:** \`AsyncConnection | Source code in `fluid/db/crud.py` ```python async def db_select( self, table: Annotated[FromClause, Doc("The table to select from")], filters: Annotated[ dict, Doc( "Key-value pairs for filtering rows; supports 'field:op' syntax " "for operators (eq, ne, gt, ge, lt, le)" ), ], *, order_by: Annotated[ tuple[str, ...] | None, Doc("Column names to order by; prefix with '-' for descending"), ] = None, conn: Annotated[ AsyncConnection | None, Doc("Optional existing connection to reuse") ] = None, ) -> CursorResult: """Select rows from a given table""" sql_query = self.get_query(table, Select(table), params=filters) if order_by: sql_query = self.order_by_query(table, cast(Select, sql_query), order_by) async with self.ensure_transaction(conn) as conn: return await conn.execute(sql_query) ``` ### db_insert ```python db_insert(table, data, *, conn=None) ``` Insert one or more rows into a table, returning the inserted rows | PARAMETER | DESCRIPTION | | --------- | --------------------------------------------------------------------------------------------------------------------------- | | `table` | The table to insert into **TYPE:** `Table` | | `data` | A single row dict or a list of row dicts; missing columns in a multi-row insert are filled with None **TYPE:** \`list[dict] | | `conn` | Optional existing connection to reuse **TYPE:** \`AsyncConnection | Source code in `fluid/db/crud.py` ```python async def db_insert( self, table: Annotated[Table, Doc("The table to insert into")], data: Annotated[ list[dict] | dict, Doc( "A single row dict or a list of row dicts; missing columns in " "a multi-row insert are filled with None" ), ], *, conn: Annotated[ AsyncConnection | None, Doc("Optional existing connection to reuse") ] = None, ) -> CursorResult: """Insert one or more rows into a table, returning the inserted rows""" async with self.ensure_transaction(conn) as conn: sql_query = self.insert_query(table, data) return await conn.execute(sql_query) ``` ### db_update ```python db_update(table, filters, data, *, conn=None) ``` Update rows matching the filters, returning all updated rows | PARAMETER | DESCRIPTION | | --------- | --------------------------------------------------------------------------------------- | | `table` | The table to update **TYPE:** `Table` | | `filters` | Key-value pairs identifying rows to update; supports 'field:op' syntax **TYPE:** `dict` | | `data` | Column values to set on the matching rows **TYPE:** `dict` | | `conn` | Optional existing connection to reuse **TYPE:** \`AsyncConnection | Source code in `fluid/db/crud.py` ```python async def db_update( self, table: Annotated[Table, Doc("The table to update")], filters: Annotated[ dict, Doc( "Key-value pairs identifying rows to update; supports 'field:op' syntax" ), ], data: Annotated[dict, Doc("Column values to set on the matching rows")], *, conn: Annotated[ AsyncConnection | None, Doc("Optional existing connection to reuse") ] = None, ) -> CursorResult: """Update rows matching the filters, returning all updated rows""" update = ( cast( Update, self.get_query(table, table.update(), params=filters), ) .values(**data) .returning(*table.columns) ) async with self.ensure_transaction(conn) as conn: return await conn.execute(update) ``` ### db_upsert ```python db_upsert(table, filters, data=None, *, conn=None) ``` Update a single row if it exists, otherwise insert it, returning the row | PARAMETER | DESCRIPTION | | --------- | ----------------------------------------------------------------------------------------------------- | | `table` | The table to upsert into **TYPE:** `Table` | | `filters` | Key-value pairs used to look up the existing row **TYPE:** `dict` | | `data` | Column values to set; if None, the row is fetched or inserted using only the filters **TYPE:** \`dict | | `conn` | Optional existing connection to reuse **TYPE:** \`AsyncConnection | Source code in `fluid/db/crud.py` ```python async def db_upsert( self, table: Annotated[Table, Doc("The table to upsert into")], filters: Annotated[ dict, Doc("Key-value pairs used to look up the existing row") ], data: Annotated[ dict | None, Doc( "Column values to set; if None, the row is fetched or inserted " "using only the filters" ), ] = None, *, conn: Annotated[ AsyncConnection | None, Doc("Optional existing connection to reuse") ] = None, ) -> Row: """Update a single row if it exists, otherwise insert it, returning the row""" if data: result = await self.db_update(table, filters, data, conn=conn) else: result = await self.db_select(table, filters, conn=conn) record = result.one_or_none() if record is None: insert_data = data.copy() if data else {} insert_data.update(filters) result = await self.db_insert(table, insert_data, conn=conn) record = result.one() return record ``` ### db_delete ```python db_delete(table, filters, *, conn=None) ``` Delete rows matching the filters, returning the deleted rows | PARAMETER | DESCRIPTION | | --------- | --------------------------------------------------------------------------------------- | | `table` | The table to delete from **TYPE:** `Table` | | `filters` | Key-value pairs identifying rows to delete; supports 'field:op' syntax **TYPE:** `dict` | | `conn` | Optional existing connection to reuse **TYPE:** \`AsyncConnection | Source code in `fluid/db/crud.py` ```python async def db_delete( self, table: Annotated[Table, Doc("The table to delete from")], filters: Annotated[ dict, Doc( "Key-value pairs identifying rows to delete; supports 'field:op' syntax" ), ], *, conn: Annotated[ AsyncConnection | None, Doc("Optional existing connection to reuse") ] = None, ) -> CursorResult: """Delete rows matching the filters, returning the deleted rows""" sql_query = self.get_query( table, table.delete().returning(*table.columns), params=filters, ) async with self.ensure_transaction(conn) as conn: return await conn.execute(sql_query) ``` ### db_count ```python db_count(table, filters, *, conn=None) ``` Count rows in a table matching the given filters | PARAMETER | DESCRIPTION | | --------- | ------------------------------------------------------------------------------- | | `table` | The table to count rows in **TYPE:** `FromClause` | | `filters` | Key-value pairs for filtering rows; supports 'field:op' syntax **TYPE:** `dict` | | `conn` | Optional existing connection to reuse **TYPE:** \`AsyncConnection | Source code in `fluid/db/crud.py` ```python async def db_count( self, table: Annotated[FromClause, Doc("The table to count rows in")], filters: Annotated[ dict, Doc("Key-value pairs for filtering rows; supports 'field:op' syntax") ], *, conn: Annotated[ AsyncConnection | None, Doc("Optional existing connection to reuse") ] = None, ) -> int: """Count rows in a table matching the given filters""" count_query = self.db_count_query( cast( Select, self.get_query( table, table.select(), params=filters, ), ), ) async with self.ensure_connection(conn) as conn: result: CursorResult = await conn.execute(count_query) return cast(int, result.scalar()) ``` ### insert_query ```python insert_query(table, records) ``` | PARAMETER | DESCRIPTION | | --------- | --------------------------------------------------------------- | | `table` | The table to insert into **TYPE:** `Table` | | `records` | A single row dict or a list of row dicts **TYPE:** \`list[dict] | Source code in `fluid/db/crud.py` ```python def insert_query( self, table: Annotated[Table, Doc("The table to insert into")], records: Annotated[ list[dict] | dict, Doc("A single row dict or a list of row dicts") ], ) -> Insert: if isinstance(records, dict): records = [records] else: cols: Set[str] = set() for record in records: cols.update(record) new_records = [] for record in records: if len(record) < len(cols): record = record.copy() missing = cols.difference(record) for col in missing: record[col] = None new_records.append(record) records = new_records return insert(table).values(records).returning(*table.columns) ``` ### get_query ```python get_query(table, sql_query, *, params=None) ``` Apply filters from params to a SQLAlchemy query and return it | PARAMETER | DESCRIPTION | | ----------- | ----------------------------------------------------------------------- | | `table` | The table the query targets **TYPE:** `FromClause` | | `sql_query` | The base SQLAlchemy query to apply filters to **TYPE:** `QueryType` | | `params` | Key-value filter pairs; keys may use 'field:op' syntax **TYPE:** \`dict | Source code in `fluid/db/crud.py` ```python def get_query( self, table: Annotated[FromClause, Doc("The table the query targets")], sql_query: Annotated[ QueryType, Doc("The base SQLAlchemy query to apply filters to") ], *, params: Annotated[ dict | None, Doc("Key-value filter pairs; keys may use 'field:op' syntax") ] = None, ) -> QueryType: """Apply filters from params to a SQLAlchemy query and return it""" filters: list = [] columns = table.c params = params or {} for key, value in params.items(): bits = key.split(":") field = bits[0] op = bits[1] if len(bits) == 2 else "eq" field = getattr(columns, field) result = self.default_filter_column(field, op, value) if result is not None: if not isinstance(result, (list, tuple)): result = (result,) filters.extend(result) if filters: whereclause = and_(*filters) if len(filters) > 1 else filters[0] sql_query = cast(Select, sql_query).where(whereclause) return sql_query ``` ### db_count_query ```python db_count_query(sql_query) ``` | PARAMETER | DESCRIPTION | | ----------- | --------------------------------------------------------------- | | `sql_query` | The filtered SELECT query to wrap in a COUNT **TYPE:** `Select` | Source code in `fluid/db/crud.py` ```python def db_count_query( self, sql_query: Annotated[ Select, Doc("The filtered SELECT query to wrap in a COUNT") ], ) -> Select: return select(func.count()).select_from(sql_query.alias("inner")) ``` ### order_by_query ```python order_by_query(table, sql_query, order_by) ``` Apply ordering to a SELECT query | PARAMETER | DESCRIPTION | | ----------- | ------------------------------------------------------------------------------------ | | `table` | The table the query targets **TYPE:** `FromClause` | | `sql_query` | The SELECT query to add ordering to **TYPE:** `Select` | | `order_by` | Column names to order by; prefix with '-' for descending **TYPE:** `tuple[str, ...]` | Source code in `fluid/db/crud.py` ```python def order_by_query( self, table: Annotated[FromClause, Doc("The table the query targets")], sql_query: Annotated[Select, Doc("The SELECT query to add ordering to")], order_by: Annotated[ tuple[str, ...], Doc("Column names to order by; prefix with '-' for descending"), ], ) -> Select: """Apply ordering to a SELECT query""" return sql_query.order_by(*self.order_by_columns(table, order_by)) ``` ### order_by_columns ```python order_by_columns(table, order_by) ``` Return a list of SQLAlchemy column expressions for the given order_by fields | PARAMETER | DESCRIPTION | | ---------- | ------------------------------------------------------------------------------------ | | `table` | The table whose columns are referenced **TYPE:** `FromClause` | | `order_by` | Column names to order by; prefix with '-' for descending **TYPE:** `tuple[str, ...]` | Source code in `fluid/db/crud.py` ```python def order_by_columns( self, table: Annotated[FromClause, Doc("The table whose columns are referenced")], order_by: Annotated[ tuple[str, ...], Doc("Column names to order by; prefix with '-' for descending"), ], ) -> list[Column]: """Return a list of SQLAlchemy column expressions for the given order_by fields""" columns = [] for name in order_by: if name.startswith("-"): order_by_column = getattr(table.c, name[1:], None) if order_by_column is not None: columns.append(order_by_column.desc()) else: order_by_column = getattr(table.c, name, None) if order_by_column is not None: columns.append(order_by_column) return columns ``` ### search_query ```python search_query(table, sql_query, search_fields, search) ``` Apply a case-insensitive substring search across the given columns | PARAMETER | DESCRIPTION | | --------------- | --------------------------------------------------------------------- | | `table` | The table whose columns are searched **TYPE:** `FromClause` | | `sql_query` | The SELECT query to add the search filter to **TYPE:** `Select` | | `search_fields` | Column names to search across using ILIKE **TYPE:** `tuple[str, ...]` | | `search` | Search text; empty string is a no-op **TYPE:** `str` | Source code in `fluid/db/crud.py` ```python def search_query( self, table: Annotated[FromClause, Doc("The table whose columns are searched")], sql_query: Annotated[ Select, Doc("The SELECT query to add the search filter to") ], search_fields: Annotated[ tuple[str, ...], Doc("Column names to search across using ILIKE") ], search: Annotated[str, Doc("Search text; empty string is a no-op")], ) -> Select: """Apply a case-insensitive substring search across the given columns""" if search and search_fields: columns = [getattr(table.c, col) for col in search_fields] return sql_query.where(or_(*(col.ilike(f"%{search}%") for col in columns))) return sql_query ``` ### default_filter_column ```python default_filter_column(column, op, value) ``` Build a SQLAlchemy WHERE clause expression for a single column filter | PARAMETER | DESCRIPTION | | --------- | ------------------------------------------------------------------------- | | `column` | The SQLAlchemy column to filter on **TYPE:** `Column` | | `op` | Comparison operator: eq, ne, gt, ge, lt, or le **TYPE:** `str` | | `value` | Comparison value; a list triggers IN / NOT IN for eq / ne **TYPE:** `Any` | Source code in `fluid/db/crud.py` ```python def default_filter_column( self, column: Annotated[Column, Doc("The SQLAlchemy column to filter on")], op: Annotated[ str, Doc("Comparison operator: `eq`, `ne`, `gt`, `ge`, `lt`, or `le`") ], value: Annotated[ Any, Doc("Comparison value; a list triggers IN / NOT IN for `eq` / `ne`") ], ) -> Any: """Build a SQLAlchemy WHERE clause expression for a single column filter""" if isinstance(column.type, JSONB) and isinstance(value, dict): if op == "eq": return column.contains(value) return None if multiple := isinstance(value, (list, tuple)): value = tuple(column_value_to_python(column, v) for v in value) else: value = column_value_to_python(column, value) if multiple and op in ("eq", "ne"): if op == "eq": return column.in_(value) elif op == "ne": return ~column.in_(value) else: if multiple: assert len(value) > 0 value = value[0] if op == "eq": return column == value elif op == "ne": return column != value elif op == "gt": return column > value elif op == "ge": return column >= value elif op == "lt": return column < value elif op == "le": return column <= value ``` # DB Migration The migration object is accessed via Database.migration or CrudDB and is used to create and manage database migrations. It requires the `db` extra to be installed: ```bash pip install aio-fluid[db] ``` ## fluid.db.Migration ```python Migration(db) ``` A wrapper around Alembic commands to perform database migrations ### db ```python db ``` ### cfg ```python cfg = field(init=False, repr=False) ``` ### metadata ```python metadata ``` ### sync_engine ```python sync_engine ``` ### init ```python init() ``` Source code in `fluid/db/migration.py` ```python def init(self) -> str: dirname = self.cfg.get_main_option("script_location") or "" alembic_cmd.init(self.cfg, dirname) _wire_metadata(Path(dirname) / "env.py") return self.message() ``` ### show ```python show(revision) ``` Source code in `fluid/db/migration.py` ```python def show(self, revision: str) -> str: alembic_cmd.show(self.cfg, revision) return self.message() ``` ### history ```python history() ``` Source code in `fluid/db/migration.py` ```python def history(self) -> str: alembic_cmd.history(self.cfg) return self.message() ``` ### revision ```python revision(message, autogenerate=False, branch_label=None) ``` Source code in `fluid/db/migration.py` ```python def revision( self, message: str, autogenerate: bool = False, branch_label: str | None = None, ) -> str: alembic_cmd.revision( self.cfg, autogenerate=autogenerate, message=message, branch_label=branch_label, ) return self.message() ``` ### upgrade ```python upgrade(revision) ``` Source code in `fluid/db/migration.py` ```python def upgrade(self, revision: str) -> str: alembic_cmd.upgrade(self.cfg, revision) return self.message() ``` ### downgrade ```python downgrade(revision) ``` Source code in `fluid/db/migration.py` ```python def downgrade(self, revision: str) -> str: alembic_cmd.downgrade(self.cfg, revision) return self.message() ``` ### current ```python current(verbose=False) ``` Source code in `fluid/db/migration.py` ```python def current(self, verbose: bool = False) -> str: alembic_cmd.current(self.cfg, verbose=verbose) return self.message() ``` ### message ```python message() ``` Source code in `fluid/db/migration.py` ```python def message(self) -> str: msg = cast(StringIO, self.cfg.stdout).getvalue() self.cfg.stdout.seek(0) self.cfg.stdout.truncate() return msg ``` ### db_exists ```python db_exists(dbname='') ``` Source code in `fluid/db/migration.py` ```python def db_exists(self, dbname: str = "") -> bool: url = self.sync_engine.url if dbname: url = url.set(database=dbname) return database_exists(url) ``` ### db_create ```python db_create(dbname='') ``` Creates a new database if it does not exist Source code in `fluid/db/migration.py` ```python def db_create(self, dbname: str = "") -> bool: """Creates a new database if it does not exist""" url = self.sync_engine.url if dbname: url = url.set(database=dbname) if database_exists(url): return False create_database(url) return True ``` ### db_drop ```python db_drop(dbname='') ``` Source code in `fluid/db/migration.py` ```python def db_drop(self, dbname: str = "") -> bool: url = self.sync_engine.url if dbname: url = url.set(database=dbname) if database_exists(url): drop_database(url) return True return False ``` ### create_all ```python create_all() ``` Create all tables from :attr:`metadata` in database Source code in `fluid/db/migration.py` ```python def create_all(self) -> None: """Create all tables from :attr:`metadata` in database""" with self.sync_engine.begin() as conn: self.metadata.create_all(conn) ``` ### truncate ```python truncate(table, *, cascade=False) ``` Truncate a specific table in the database Source code in `fluid/db/migration.py` ```python def truncate(self, table: str, *, cascade: bool = False) -> None: """Truncate a specific table in the database""" cascade_sql = " cascade" if cascade else "" with self.sync_engine.begin() as conn: conn.execute(sa.text(f"truncate table {table}{cascade_sql}")) ``` ### truncate_all ```python truncate_all() ``` Truncate all tables in the database Source code in `fluid/db/migration.py` ```python def truncate_all(self) -> None: """Truncate all tables in the database""" with self.sync_engine.begin() as conn: conn.execute(sa.text(f'truncate {", ".join(self.metadata.tables)}')) ``` ### drop_all_schemas ```python drop_all_schemas() ``` Drop all schema in database Source code in `fluid/db/migration.py` ```python def drop_all_schemas(self) -> None: """Drop all schema in database""" with self.sync_engine.begin() as conn: conn.execute(sa.text("DROP SCHEMA IF EXISTS public CASCADE")) conn.execute(sa.text("CREATE SCHEMA IF NOT EXISTS public")) ``` ### create_ro_user ```python create_ro_user( username, password, role="", schema="public" ) ``` Creates a read-only user Source code in `fluid/db/migration.py` ```python def create_ro_user( self, username: str, password: str, role: str = "", schema: str = "public", ) -> bool: """Creates a read-only user""" engine = self.sync_engine role = role or f"{engine.url.username}_ro" database = engine.url.database created = True with engine.begin() as conn: try: conn.execute(sa.text(f"CREATE ROLE {role};")) except sa.exc.ProgrammingError: created = False with engine.begin() as conn: conn.execute( sa.text( f"GRANT CONNECT ON DATABASE {database} TO {role};" f"GRANT USAGE ON SCHEMA {schema} TO {role};" f"GRANT SELECT ON ALL TABLES IN SCHEMA {schema} TO {role};" f"GRANT SELECT ON ALL SEQUENCES IN SCHEMA {schema} TO {role};", ), ) conn.execute( sa.text( f"ALTER DEFAULT PRIVILEGES IN SCHEMA {schema} " f"GRANT SELECT ON TABLES TO {role};", ), ) with engine.begin() as conn: try: conn.execute( sa.text( f"CREATE USER {username} WITH PASSWORD '{password}';" f"GRANT {role} TO {username};", ), ) except sa.exc.ProgrammingError: created = False return created ``` ### drop_role ```python drop_role(role) ``` Drop a role Source code in `fluid/db/migration.py` ```python def drop_role( self, role: str, ) -> bool: """Drop a role""" try: with self.sync_engine.begin() as conn: conn.execute(sa.text(f"DROP OWNED BY {role};")) with self.sync_engine.begin() as conn: conn.execute(sa.text(f"DROP ROLE IF EXISTS {role};")) except sa.exc.ProgrammingError as exc: if f'role "{role}" does not exist' not in str(exc): raise return False return True ``` # DB Pagination The Pagination class is a tool for managing paginated rows from the database. It requires the `db` extra to be installed: ```bash pip install aio-fluid[db] ``` It can be imported from `fluid.db`: ```python from fluid.db import Pagination, Search ``` ## fluid.db.Pagination Bases: `NamedTuple` Cursor-based pagination over a database table. The `order_by_fields` must uniquely identify a row: the values of those fields, taken together, must be distinct for every row matched by the query. The cursor stores nothing but those values, so it can only resume from an unambiguous position. Add a unique column, usually the primary key, as the last ordering field when the other fields can tie: ```python Pagination.create("published_at", "id", limit=20) ``` When the ordering is not unique, rows sharing the same ordering values can be repeated across consecutive pages, and rows can be missed entirely because the database is free to order tied rows differently between the two queries. ### order_by_fields ```python order_by_fields ``` Fields to order results by ### limit ```python limit ``` Maximum number of results per page ### filters ```python filters ``` Filters applied to the query ### search ```python search ``` Full-text search configuration ### cursor ```python cursor ``` Decoded pagination cursor ### desc ```python desc = False ``` Order results in descending order ### order_by_fields_sign ```python order_by_fields_sign ``` ### create ```python create( *order_by_fields, cursor="", limit=None, filters=None, search=None, desc=False ) ``` Factory method to create a Pagination instance, decoding the cursor if provided. If the cursor is provided, filters, limit and search are extracted from it, and the provided values for these parameters are ignored. The `order_by_fields` must uniquely identify a row, otherwise pages can repeat or miss rows. | PARAMETER | DESCRIPTION | | ------------------ | ------------------------------------------------------------------------------------------------- | | `*order_by_fields` | Fields to order results by **TYPE:** `str` **DEFAULT:** `()` | | `cursor` | Encoded pagination cursor from a previous response **TYPE:** `str` **DEFAULT:** `''` | | `limit` | Maximum number of results per page; defaults to settings.DEFAULT_PAGINATION_LIMIT **TYPE:** \`int | | `filters` | Filters to apply to the query **TYPE:** \`dict[str, Any] | | `search` | Full-text search configuration **TYPE:** \`Search | | `desc` | Order results in descending order **TYPE:** `bool` **DEFAULT:** `False` | Source code in `fluid/db/pagination.py` ```python @classmethod def create( cls, *order_by_fields: Annotated[str, Doc("Fields to order results by")], cursor: Annotated[ str, Doc("Encoded pagination cursor from a previous response"), ] = "", limit: Annotated[ int | None, Doc( "Maximum number of results per page; " "defaults to settings.DEFAULT_PAGINATION_LIMIT" ), ] = None, filters: Annotated[ dict[str, Any] | None, Doc("Filters to apply to the query"), ] = None, search: Annotated[ Search | None, Doc("Full-text search configuration"), ] = None, desc: Annotated[ bool, Doc("Order results in descending order"), ] = False, ) -> Self: """Factory method to create a Pagination instance, decoding the cursor if provided. If the cursor is provided, filters, limit and search are extracted from it, and the provided values for these parameters are ignored. The `order_by_fields` must uniquely identify a row, otherwise pages can repeat or miss rows. """ if cursor: decoded_cursor = Cursor.decode(cursor, order_by_fields) limit = decoded_cursor.limit filters = decoded_cursor.filters if search: search = search._replace(search_text=decoded_cursor.search_text) else: decoded_cursor = None limit = limit or settings.DEFAULT_PAGINATION_LIMIT return cls( order_by_fields=order_by_fields, cursor=decoded_cursor, limit=limit, filters=filters or {}, search=search, desc=desc, ) ``` ### execute ```python execute(db, table, *, conn=None) ``` Execute the paginated query and return the results along with the next cursor. | PARAMETER | DESCRIPTION | | --------- | ----------------------------------------------------------------- | | `db` | Database instance to execute the query on **TYPE:** `CrudDB` | | `table` | SQLAlchemy table to query **TYPE:** `FromClause` | | `conn` | Optional existing connection to reuse **TYPE:** \`AsyncConnection | Source code in `fluid/db/pagination.py` ```python async def execute( self, db: Annotated[CrudDB, Doc("Database instance to execute the query on")], table: Annotated[FromClause, Doc("SQLAlchemy table to query")], *, conn: Annotated[ AsyncConnection | None, Doc("Optional existing connection to reuse"), ] = None, ) -> tuple[Sequence[Row], str]: """Execute the paginated query and return the results along with the next cursor. """ sql_query = self.query(db, table) async with db.ensure_connection(conn) as conn: result = await conn.execute(sql_query) data = result.all() cursor = "" if self.limit > 0 and len(data) > self.limit: cursor = self._encode_cursor(data[-1]) data = data[:-1] return data, cursor ``` ### query ```python query(db, table) ``` Source code in `fluid/db/pagination.py` ```python def query(self, db: CrudDB, table: FromClause) -> Select: sql_query = cast( Select, db.get_query(table, table.select(), params=self.filters), ) if self.search: sql_query = db.search_query( table, sql_query, self.search.search_fields, self.search.search_text, ) start_clause = self._start_clause(table) if start_clause is not None: sql_query = sql_query.where(start_clause) columns = db.order_by_columns(table, self.order_by_fields_sign) ordered = sql_query.order_by(*columns) return ordered.limit(self.limit + 1) if self.limit > 0 else ordered ``` ## fluid.db.Search Bases: `NamedTuple` ### search_fields ```python search_fields ``` Fields to search in ### search_text ```python search_text ``` Text to search for # Event Dispatchers A set of classes for dispatching events, they can be imported from `fluid.utils.dispatcher`: ```python from fluid.utils.dispatcher import Dispatcher ``` ## fluid.utils.dispatcher.Event Bases: `NamedTuple` ### type ```python type ``` The event type ### tag ```python tag = '' ``` The event tag - for registering multiple handlers for a given event type ### from_string_or_event ```python from_string_or_event(event) ``` | PARAMETER | DESCRIPTION | | --------- | ------------------------------------------------------------------------------------------ | | `event` | The Event or a string of the form {event_type} or {event_type}.{event_tag} **TYPE:** \`str | Source code in `fluid/utils/dispatcher.py` ```python @classmethod def from_string_or_event( cls, event: Annotated[ str | Self, Doc( "The [Event][fluid.utils.dispatcher.Event] or a string of" " the form `{event_type}` or `{event_type}.{event_tag}`" ), ], ) -> Self: if isinstance(event, str): return cls.from_string(event) return event ``` ### from_string ```python from_string(event) ``` | PARAMETER | DESCRIPTION | | --------- | -------------------------------------------------------------------------------------- | | `event` | The event string has the form {event_type} or {event_type}.{event_tag} **TYPE:** `str` | Source code in `fluid/utils/dispatcher.py` ```python @classmethod def from_string( cls, event: Annotated[ str, Doc( "The event string has the form `{event_type}` " "or `{event_type}.{event_tag}`" ), ], ) -> Self: bits = event.split(".") return cls(bits[0], bits[1] if len(bits) > 1 else "") ``` ## fluid.utils.dispatcher.BaseDispatcher ```python BaseDispatcher() ``` Bases: `Generic[MessageType, MessageHandlerType]`, `ABC` Base generic abstract class for dispatchers Source code in `fluid/utils/dispatcher.py` ```python def __init__(self) -> None: self._msg_handlers: defaultdict[str, dict[str, MessageHandlerType]] = ( defaultdict( dict, ) ) ``` ### register_handler ```python register_handler(event, handler) ``` Register a handler for the given event It is possible to register multiple handlers for the same event type by providing a different tag for each handler. For example, to register two handlers for the event type `foo`: ```python dispatcher.register_handler("foo.first", handler1) dispatcher.register_handler("foo.second", handler2) ``` | PARAMETER | DESCRIPTION | | --------- | ------------------------------------------------------- | | `event` | The event to register the handler for **TYPE:** \`Event | | `handler` | The handler to register **TYPE:** `MessageHandlerType` | Source code in `fluid/utils/dispatcher.py` ````python def register_handler( self, event: Annotated[Event | str, Doc("The event to register the handler for")], handler: Annotated[MessageHandlerType, Doc("The handler to register")], ) -> MessageHandlerType | None: """Register a handler for the given event It is possible to register multiple handlers for the same event type by providing a different tag for each handler. For example, to register two handlers for the event type `foo`: ```python dispatcher.register_handler("foo.first", handler1) dispatcher.register_handler("foo.second", handler2) ``` """ event = Event.from_string_or_event(event) previous = self._msg_handlers[event.type].get(event.tag) self._msg_handlers[event.type][event.tag] = handler return previous ```` ### unregister_handler ```python unregister_handler(event) ``` Unregister a handler for the given event It returns the handler that was unregistered or `None` if no handler was registered for the given event. | PARAMETER | DESCRIPTION | | --------- | ----------------------------------------------------- | | `event` | The event to unregister the handler **TYPE:** \`Event | Source code in `fluid/utils/dispatcher.py` ```python def unregister_handler( self, event: Annotated[Event | str, Doc("The event to unregister the handler")] ) -> MessageHandlerType | None: """Unregister a handler for the given event It returns the handler that was unregistered or `None` if no handler was registered for the given event. """ event = Event.from_string_or_event(event) return self._msg_handlers[event.type].pop(event.tag, None) ``` ### get_handlers ```python get_handlers(message) ``` Get all event handlers for the given message This method returns a dictionary of all handlers registered for the given message type. If no handlers are registered for the message type, it returns `None`. | PARAMETER | DESCRIPTION | | --------- | ----------------------------------------------------------- | | `message` | The message to get the handlers for **TYPE:** `MessageType` | Source code in `fluid/utils/dispatcher.py` ```python def get_handlers( self, message: Annotated[MessageType, Doc("The message to get the handlers for")], ) -> dict[str, MessageHandlerType] | None: """Get all event handlers for the given message This method returns a dictionary of all handlers registered for the given message type. If no handlers are registered for the message type, it returns `None`. """ event_type = self.event_type(message) return self._msg_handlers.get(event_type) ``` ### event_type ```python event_type(message) ``` return the event type as string Source code in `fluid/utils/dispatcher.py` ```python @abstractmethod def event_type(self, message: MessageType) -> str: """return the event type as string""" ``` ## fluid.utils.dispatcher.Dispatcher ```python Dispatcher() ``` Bases: `BaseDispatcher[MessageType, Callable[[MessageType], None]]` Dispatcher for sync handlers Source code in `fluid/utils/dispatcher.py` ```python def __init__(self) -> None: self._msg_handlers: defaultdict[str, dict[str, MessageHandlerType]] = ( defaultdict( dict, ) ) ``` ### dispatch ```python dispatch(message) ``` dispatch the message to all handlers It returns the number of handlers that were called Source code in `fluid/utils/dispatcher.py` ```python def dispatch(self, message: MessageType) -> int: """dispatch the message to all handlers It returns the number of handlers that were called """ handlers = self.get_handlers(message) if handlers: for handler in handlers.values(): handler(message) return len(handlers or ()) ``` ### register_handler ```python register_handler(event, handler) ``` Register a handler for the given event It is possible to register multiple handlers for the same event type by providing a different tag for each handler. For example, to register two handlers for the event type `foo`: ```python dispatcher.register_handler("foo.first", handler1) dispatcher.register_handler("foo.second", handler2) ``` | PARAMETER | DESCRIPTION | | --------- | ------------------------------------------------------- | | `event` | The event to register the handler for **TYPE:** \`Event | | `handler` | The handler to register **TYPE:** `MessageHandlerType` | Source code in `fluid/utils/dispatcher.py` ````python def register_handler( self, event: Annotated[Event | str, Doc("The event to register the handler for")], handler: Annotated[MessageHandlerType, Doc("The handler to register")], ) -> MessageHandlerType | None: """Register a handler for the given event It is possible to register multiple handlers for the same event type by providing a different tag for each handler. For example, to register two handlers for the event type `foo`: ```python dispatcher.register_handler("foo.first", handler1) dispatcher.register_handler("foo.second", handler2) ``` """ event = Event.from_string_or_event(event) previous = self._msg_handlers[event.type].get(event.tag) self._msg_handlers[event.type][event.tag] = handler return previous ```` ### unregister_handler ```python unregister_handler(event) ``` Unregister a handler for the given event It returns the handler that was unregistered or `None` if no handler was registered for the given event. | PARAMETER | DESCRIPTION | | --------- | ----------------------------------------------------- | | `event` | The event to unregister the handler **TYPE:** \`Event | Source code in `fluid/utils/dispatcher.py` ```python def unregister_handler( self, event: Annotated[Event | str, Doc("The event to unregister the handler")] ) -> MessageHandlerType | None: """Unregister a handler for the given event It returns the handler that was unregistered or `None` if no handler was registered for the given event. """ event = Event.from_string_or_event(event) return self._msg_handlers[event.type].pop(event.tag, None) ``` ### get_handlers ```python get_handlers(message) ``` Get all event handlers for the given message This method returns a dictionary of all handlers registered for the given message type. If no handlers are registered for the message type, it returns `None`. | PARAMETER | DESCRIPTION | | --------- | ----------------------------------------------------------- | | `message` | The message to get the handlers for **TYPE:** `MessageType` | Source code in `fluid/utils/dispatcher.py` ```python def get_handlers( self, message: Annotated[MessageType, Doc("The message to get the handlers for")], ) -> dict[str, MessageHandlerType] | None: """Get all event handlers for the given message This method returns a dictionary of all handlers registered for the given message type. If no handlers are registered for the message type, it returns `None`. """ event_type = self.event_type(message) return self._msg_handlers.get(event_type) ``` ### event_type ```python event_type(message) ``` return the event type as string Source code in `fluid/utils/dispatcher.py` ```python @abstractmethod def event_type(self, message: MessageType) -> str: """return the event type as string""" ``` ## fluid.utils.dispatcher.AsyncDispatcher ```python AsyncDispatcher() ``` Bases: `BaseDispatcher[MessageType, Callable[[MessageType], Awaitable[None]]]` Dispatcher for async handlers Source code in `fluid/utils/dispatcher.py` ```python def __init__(self) -> None: self._msg_handlers: defaultdict[str, dict[str, MessageHandlerType]] = ( defaultdict( dict, ) ) ``` ### dispatch ```python dispatch(message) ``` Dispatch the message and wait for all handlers to complete It returns the number of handlers that were called Source code in `fluid/utils/dispatcher.py` ```python async def dispatch(self, message: MessageType) -> int: """Dispatch the message and wait for all handlers to complete It returns the number of handlers that were called """ handlers = self.get_handlers(message) if handlers: await asyncio.gather(*[handler(message) for handler in handlers.values()]) return len(handlers or ()) ``` ### register_handler ```python register_handler(event, handler) ``` Register a handler for the given event It is possible to register multiple handlers for the same event type by providing a different tag for each handler. For example, to register two handlers for the event type `foo`: ```python dispatcher.register_handler("foo.first", handler1) dispatcher.register_handler("foo.second", handler2) ``` | PARAMETER | DESCRIPTION | | --------- | ------------------------------------------------------- | | `event` | The event to register the handler for **TYPE:** \`Event | | `handler` | The handler to register **TYPE:** `MessageHandlerType` | Source code in `fluid/utils/dispatcher.py` ````python def register_handler( self, event: Annotated[Event | str, Doc("The event to register the handler for")], handler: Annotated[MessageHandlerType, Doc("The handler to register")], ) -> MessageHandlerType | None: """Register a handler for the given event It is possible to register multiple handlers for the same event type by providing a different tag for each handler. For example, to register two handlers for the event type `foo`: ```python dispatcher.register_handler("foo.first", handler1) dispatcher.register_handler("foo.second", handler2) ``` """ event = Event.from_string_or_event(event) previous = self._msg_handlers[event.type].get(event.tag) self._msg_handlers[event.type][event.tag] = handler return previous ```` ### unregister_handler ```python unregister_handler(event) ``` Unregister a handler for the given event It returns the handler that was unregistered or `None` if no handler was registered for the given event. | PARAMETER | DESCRIPTION | | --------- | ----------------------------------------------------- | | `event` | The event to unregister the handler **TYPE:** \`Event | Source code in `fluid/utils/dispatcher.py` ```python def unregister_handler( self, event: Annotated[Event | str, Doc("The event to unregister the handler")] ) -> MessageHandlerType | None: """Unregister a handler for the given event It returns the handler that was unregistered or `None` if no handler was registered for the given event. """ event = Event.from_string_or_event(event) return self._msg_handlers[event.type].pop(event.tag, None) ``` ### get_handlers ```python get_handlers(message) ``` Get all event handlers for the given message This method returns a dictionary of all handlers registered for the given message type. If no handlers are registered for the message type, it returns `None`. | PARAMETER | DESCRIPTION | | --------- | ----------------------------------------------------------- | | `message` | The message to get the handlers for **TYPE:** `MessageType` | Source code in `fluid/utils/dispatcher.py` ```python def get_handlers( self, message: Annotated[MessageType, Doc("The message to get the handlers for")], ) -> dict[str, MessageHandlerType] | None: """Get all event handlers for the given message This method returns a dictionary of all handlers registered for the given message type. If no handlers are registered for the message type, it returns `None`. """ event_type = self.event_type(message) return self._msg_handlers.get(event_type) ``` ### event_type ```python event_type(message) ``` return the event type as string Source code in `fluid/utils/dispatcher.py` ```python @abstractmethod def event_type(self, message: MessageType) -> str: """return the event type as string""" ``` # Errors `aio-fluid` defines two error hierarchies: one for general utilities and one for the task scheduler. ## Utility errors ```python from fluid.utils.errors import FluidError, FluidValueError, ValidationError, WorkerStartError ``` ## fluid.utils.errors.FluidError Bases: `Exception` Base class for all Fluid Trading errors. ## fluid.utils.errors.FluidValueError Bases: `ValueError`, `FluidError` Quant Value Error ## fluid.utils.errors.ValidationError ```python ValidationError( field, msg="", field_type="string", value=None ) ``` Bases: `FluidValueError` Validation Error Source code in `fluid/utils/errors.py` ```python def __init__( self, field: str, msg: str = "", field_type: str = "string", value: Any = None ): self.field = field self.field_type = field_type self.msg = msg or "validation error" self.value = value ``` ### field ```python field = field ``` ### field_type ```python field_type = field_type ``` ### msg ```python msg = msg or 'validation error' ``` ### value ```python value = value ``` ## fluid.utils.errors.WorkerStartError Bases: `FluidError` Worker start error ## fluid.utils.errors.FlamegraphError Bases: `FluidError` Raised when flamegraph generation fails. ## Task scheduler errors ```python from fluid.scheduler.errors import TaskError, UnknownTaskError, DisabledTaskError ``` ## fluid.scheduler.errors.TaskError Bases: `RuntimeError` Base class for all task scheduler errors. ## fluid.scheduler.errors.UnknownTaskError Bases: `TaskError` Raised when a task name is not registered in the task registry. ## fluid.scheduler.errors.DisabledTaskError Bases: `TaskError` Raised when attempting to queue or run a disabled task. ## fluid.scheduler.errors.TaskParamsError ```python TaskParamsError(task_run, message) ``` Bases: `TaskError` Raised when task run parameters fail validation when consumed from the broker. It carries the task run, created with unvalidated params, so the consumer can mark it as failed and log the error. Source code in `fluid/scheduler/errors.py` ```python def __init__(self, task_run: TaskRun, message: str) -> None: super().__init__(message) self.task_run = task_run ``` ### task_run ```python task_run = task_run ``` ## fluid.scheduler.errors.TaskRunError Bases: `TaskError` Raised when a task run fails during execution. This is an internal error used to signal a failure during task execution, and is not intended to be raised by user code. ## fluid.scheduler.errors.TaskAbortedError Bases: `TaskError` Raised when a task run is aborted before completion. If a task needs to abort itself it can raise this error, which will be caught by the consumer and treated as a soft-failure and therefore logged as info and not trigger any retry policy if configured. ## fluid.scheduler.errors.TaskDecoratorError Bases: `TaskError` Raised when a task is incorrectly decorated or configured. ## fluid.scheduler.errors.CpuBoundEntryPointError Bases: `TaskError` Raised when a cpu bound task cannot be executed by the entry point. Cpu bound tasks run in a separate process, started by the `exec` command of the task manager command line client. An application which does not expose one cannot execute them. # HTTP Client `aio-fluid` provides async HTTP client wrappers around [aiohttp](https://docs.aiohttp.org/) and [httpx](https://www.python-httpx.org/) with a unified interface for making requests, handling errors, and monitoring calls. ```python from fluid.utils.http_client import AioHttpClient, HttpxClient ``` Both clients implement the same `HttpClient` base class, so they can be used interchangeably. ## Usage ```python async with AioHttpClient() as client: data = await client.get("https://api.example.com/items") async with HttpxClient() as client: data = await client.post("https://api.example.com/items", json={"name": "foo"}) ``` ## Response ## fluid.utils.http_client.HttpResponse Bases: `ABC` ### url ```python url ``` ### status_code ```python status_code ``` ### method ```python method ``` ### headers ```python headers ``` ### json ```python json() ``` Source code in `fluid/utils/http_client.py` ```python @abstractmethod async def json(self) -> ResponseType: ... ``` ### text ```python text() ``` Source code in `fluid/utils/http_client.py` ```python @abstractmethod async def text(self) -> str: ... ``` ### bytes ```python bytes() ``` Source code in `fluid/utils/http_client.py` ```python @abstractmethod async def bytes(self) -> bytes: ... ``` ## fluid.utils.http_client.HttpResponseError ```python HttpResponseError(response, data) ``` Bases: `RuntimeError` Source code in `fluid/utils/http_client.py` ```python def __init__(self, response: HttpResponse, data: ResponseType) -> None: self.response = response self.data = { "response": data, "request_url": response.url, "request_method": response.method, "response_status": self.status_code, } ``` ### response ```python response = response ``` ### data ```python data = { "response": data, "request_url": response.url, "request_method": response.method, "response_status": self.status_code, } ``` ### status_code ```python status_code ``` ## Clients ## fluid.utils.http_client.HttpClient ```python HttpClient( session=None, content_type="application/json", session_owner=False, ResponseError=HttpResponseError, ok_status=frozenset((200, 201)), default_headers=( lambda: {"user-agent": HTTP_USER_AGENT} )(), ) ``` Bases: `Generic[S, R]`, `ABC` Base class for Http clients ### session ```python session = None ``` ### content_type ```python content_type = 'application/json' ``` ### session_owner ```python session_owner = False ``` ### ResponseError ```python ResponseError = field(default=HttpResponseError, repr=False) ``` ### ok_status ```python ok_status = field(default=frozenset((200, 201)), repr=False) ``` ### default_headers ```python default_headers = field( default_factory=lambda: { "user-agent": settings.HTTP_USER_AGENT } ) ``` ### new_session ```python new_session(**kwargs) ``` Source code in `fluid/utils/http_client.py` ```python @abstractmethod def new_session(self, **kwargs: Any) -> S: ... ``` ### new_response ```python new_response(response) ``` Source code in `fluid/utils/http_client.py` ```python @abstractmethod def new_response(self, response: R) -> GenericHttpResponse[R]: ... ``` ### close ```python close() ``` Source code in `fluid/utils/http_client.py` ```python @abstractmethod async def close(self) -> None: ... ``` ### get_session ```python get_session() ``` Source code in `fluid/utils/http_client.py` ```python def get_session(self) -> S: if not self.session: self.session_owner = True self.session = self.new_session() return self.session ``` ### get ```python get(url, **kwargs) ``` Source code in `fluid/utils/http_client.py` ```python async def get(self, url: str, **kwargs: Any) -> ResponseType: return await self.request("GET", url, **kwargs) ``` ### patch ```python patch(url, **kwargs) ``` Source code in `fluid/utils/http_client.py` ```python async def patch(self, url: str, **kwargs: Any) -> ResponseType: return await self.request("PATCH", url, **kwargs) ``` ### post ```python post(url, **kwargs) ``` Source code in `fluid/utils/http_client.py` ```python async def post(self, url: str, **kwargs: Any) -> ResponseType: return await self.request("POST", url, **kwargs) ``` ### put ```python put(url, **kwargs) ``` Source code in `fluid/utils/http_client.py` ```python async def put(self, url: str, **kwargs: Any) -> ResponseType: return await self.request("PUT", url, **kwargs) ``` ### delete ```python delete(url, **kwargs) ``` Source code in `fluid/utils/http_client.py` ```python async def delete(self, url: str, **kwargs: Any) -> ResponseType: return await self.request("DELETE", url, **kwargs) ``` ### request ```python request( method, url, *, headers=None, callback=None, monitor_http=None, **kw ) ``` Source code in `fluid/utils/http_client.py` ```python async def request( self, method: str, url: str, *, headers: dict | None = None, callback: Callable | bool | None = None, monitor_http: HttpPathFn | None = None, **kw: Any, ) -> ResponseType: session = self.get_session() _headers = self.get_default_headers() _headers.update(headers or ()) method = method or "GET" start = time.monotonic() inner: R = await session.request( method, url, headers=_headers, **kw, ) # type: ignore response = self.new_response(inner) if monitor_http: monitor_http_call( response, time.monotonic() - start, sanitization_fn=monitor_http, ) if callback: if callback is True: return response else: return await callback(response) if self.ok(response): data = await self.response_data(response) elif response.status_code == 204: data = {} else: await self.response_error(response) return data ``` ### ok ```python ok(response) ``` Source code in `fluid/utils/http_client.py` ```python def ok(self, response: HttpResponse) -> bool: return response.status_code in self.ok_status ``` ### get_default_headers ```python get_default_headers() ``` Source code in `fluid/utils/http_client.py` ```python def get_default_headers(self) -> dict[str, str]: headers = self.default_headers.copy() if self.content_type: headers["accept"] = self.content_type return headers ``` ### response_error ```python response_error(response) ``` Source code in `fluid/utils/http_client.py` ```python @classmethod async def response_error(cls, response: HttpResponse) -> None: try: data = await cls.response_data(response) except Exception: data = {"message": await response.text()} raise cls.ResponseError(response, data) ``` ### response_data ```python response_data(response) ``` Source code in `fluid/utils/http_client.py` ```python @classmethod async def response_data(cls, response: HttpResponse) -> ResponseType: content_type = response.headers.get("content-type", "") if "json" in content_type: return await response.json() elif "text" in content_type: return await response.text() return await response.bytes() ``` ## fluid.utils.http_client.AioHttpClient ```python AioHttpClient( session=None, content_type="application/json", session_owner=False, ResponseError=HttpResponseError, ok_status=frozenset((200, 201)), default_headers=( lambda: {"user-agent": HTTP_USER_AGENT} )(), ) ``` Bases: `HttpClient[ClientSession, ClientResponse]` ### session ```python session = None ``` ### content_type ```python content_type = 'application/json' ``` ### session_owner ```python session_owner = False ``` ### ResponseError ```python ResponseError = field(default=HttpResponseError, repr=False) ``` ### ok_status ```python ok_status = field(default=frozenset((200, 201)), repr=False) ``` ### default_headers ```python default_headers = field( default_factory=lambda: { "user-agent": settings.HTTP_USER_AGENT } ) ``` ### new_session ```python new_session(**kwargs) ``` Source code in `fluid/utils/http_client.py` ```python def new_session(self, **kwargs: Any) -> client.ClientSession: return client.ClientSession(**kwargs) ``` ### new_response ```python new_response(response) ``` Source code in `fluid/utils/http_client.py` ```python def new_response( self, response: client.ClientResponse ) -> GenericHttpResponse[client.ClientResponse]: return AioHttpResponse(response) ``` ### close ```python close() ``` Source code in `fluid/utils/http_client.py` ```python async def close(self) -> None: if self.session and self.session_owner: await self.session.close() self.session = None ``` ### get_session ```python get_session() ``` Source code in `fluid/utils/http_client.py` ```python def get_session(self) -> S: if not self.session: self.session_owner = True self.session = self.new_session() return self.session ``` ### get ```python get(url, **kwargs) ``` Source code in `fluid/utils/http_client.py` ```python async def get(self, url: str, **kwargs: Any) -> ResponseType: return await self.request("GET", url, **kwargs) ``` ### patch ```python patch(url, **kwargs) ``` Source code in `fluid/utils/http_client.py` ```python async def patch(self, url: str, **kwargs: Any) -> ResponseType: return await self.request("PATCH", url, **kwargs) ``` ### post ```python post(url, **kwargs) ``` Source code in `fluid/utils/http_client.py` ```python async def post(self, url: str, **kwargs: Any) -> ResponseType: return await self.request("POST", url, **kwargs) ``` ### put ```python put(url, **kwargs) ``` Source code in `fluid/utils/http_client.py` ```python async def put(self, url: str, **kwargs: Any) -> ResponseType: return await self.request("PUT", url, **kwargs) ``` ### delete ```python delete(url, **kwargs) ``` Source code in `fluid/utils/http_client.py` ```python async def delete(self, url: str, **kwargs: Any) -> ResponseType: return await self.request("DELETE", url, **kwargs) ``` ### request ```python request( method, url, *, headers=None, callback=None, monitor_http=None, **kw ) ``` Source code in `fluid/utils/http_client.py` ```python async def request( self, method: str, url: str, *, headers: dict | None = None, callback: Callable | bool | None = None, monitor_http: HttpPathFn | None = None, **kw: Any, ) -> ResponseType: session = self.get_session() _headers = self.get_default_headers() _headers.update(headers or ()) method = method or "GET" start = time.monotonic() inner: R = await session.request( method, url, headers=_headers, **kw, ) # type: ignore response = self.new_response(inner) if monitor_http: monitor_http_call( response, time.monotonic() - start, sanitization_fn=monitor_http, ) if callback: if callback is True: return response else: return await callback(response) if self.ok(response): data = await self.response_data(response) elif response.status_code == 204: data = {} else: await self.response_error(response) return data ``` ### ok ```python ok(response) ``` Source code in `fluid/utils/http_client.py` ```python def ok(self, response: HttpResponse) -> bool: return response.status_code in self.ok_status ``` ### get_default_headers ```python get_default_headers() ``` Source code in `fluid/utils/http_client.py` ```python def get_default_headers(self) -> dict[str, str]: headers = self.default_headers.copy() if self.content_type: headers["accept"] = self.content_type return headers ``` ### response_error ```python response_error(response) ``` Source code in `fluid/utils/http_client.py` ```python @classmethod async def response_error(cls, response: HttpResponse) -> None: try: data = await cls.response_data(response) except Exception: data = {"message": await response.text()} raise cls.ResponseError(response, data) ``` ### response_data ```python response_data(response) ``` Source code in `fluid/utils/http_client.py` ```python @classmethod async def response_data(cls, response: HttpResponse) -> ResponseType: content_type = response.headers.get("content-type", "") if "json" in content_type: return await response.json() elif "text" in content_type: return await response.text() return await response.bytes() ``` ## fluid.utils.http_client.HttpxClient ```python HttpxClient( session=None, content_type="application/json", session_owner=False, ResponseError=HttpResponseError, ok_status=frozenset((200, 201)), default_headers=( lambda: {"user-agent": HTTP_USER_AGENT} )(), ) ``` Bases: `HttpClient[AsyncClient, Response]` ### session ```python session = None ``` ### content_type ```python content_type = 'application/json' ``` ### session_owner ```python session_owner = False ``` ### ResponseError ```python ResponseError = field(default=HttpResponseError, repr=False) ``` ### ok_status ```python ok_status = field(default=frozenset((200, 201)), repr=False) ``` ### default_headers ```python default_headers = field( default_factory=lambda: { "user-agent": settings.HTTP_USER_AGENT } ) ``` ### new_session ```python new_session(**kwargs) ``` Source code in `fluid/utils/http_client.py` ```python def new_session(self, **kwargs: Any) -> httpx.AsyncClient: return httpx.AsyncClient(**kwargs) ``` ### new_response ```python new_response(response) ``` Source code in `fluid/utils/http_client.py` ```python def new_response( self, response: httpx.Response ) -> GenericHttpResponse[httpx.Response]: return HttpxResponse(response) ``` ### close ```python close() ``` Source code in `fluid/utils/http_client.py` ```python async def close(self) -> None: if self.session and self.session_owner: await self.session.aclose() self.session = None ``` ### get_session ```python get_session() ``` Source code in `fluid/utils/http_client.py` ```python def get_session(self) -> S: if not self.session: self.session_owner = True self.session = self.new_session() return self.session ``` ### get ```python get(url, **kwargs) ``` Source code in `fluid/utils/http_client.py` ```python async def get(self, url: str, **kwargs: Any) -> ResponseType: return await self.request("GET", url, **kwargs) ``` ### patch ```python patch(url, **kwargs) ``` Source code in `fluid/utils/http_client.py` ```python async def patch(self, url: str, **kwargs: Any) -> ResponseType: return await self.request("PATCH", url, **kwargs) ``` ### post ```python post(url, **kwargs) ``` Source code in `fluid/utils/http_client.py` ```python async def post(self, url: str, **kwargs: Any) -> ResponseType: return await self.request("POST", url, **kwargs) ``` ### put ```python put(url, **kwargs) ``` Source code in `fluid/utils/http_client.py` ```python async def put(self, url: str, **kwargs: Any) -> ResponseType: return await self.request("PUT", url, **kwargs) ``` ### delete ```python delete(url, **kwargs) ``` Source code in `fluid/utils/http_client.py` ```python async def delete(self, url: str, **kwargs: Any) -> ResponseType: return await self.request("DELETE", url, **kwargs) ``` ### request ```python request( method, url, *, headers=None, callback=None, monitor_http=None, **kw ) ``` Source code in `fluid/utils/http_client.py` ```python async def request( self, method: str, url: str, *, headers: dict | None = None, callback: Callable | bool | None = None, monitor_http: HttpPathFn | None = None, **kw: Any, ) -> ResponseType: session = self.get_session() _headers = self.get_default_headers() _headers.update(headers or ()) method = method or "GET" start = time.monotonic() inner: R = await session.request( method, url, headers=_headers, **kw, ) # type: ignore response = self.new_response(inner) if monitor_http: monitor_http_call( response, time.monotonic() - start, sanitization_fn=monitor_http, ) if callback: if callback is True: return response else: return await callback(response) if self.ok(response): data = await self.response_data(response) elif response.status_code == 204: data = {} else: await self.response_error(response) return data ``` ### ok ```python ok(response) ``` Source code in `fluid/utils/http_client.py` ```python def ok(self, response: HttpResponse) -> bool: return response.status_code in self.ok_status ``` ### get_default_headers ```python get_default_headers() ``` Source code in `fluid/utils/http_client.py` ```python def get_default_headers(self) -> dict[str, str]: headers = self.default_headers.copy() if self.content_type: headers["accept"] = self.content_type return headers ``` ### response_error ```python response_error(response) ``` Source code in `fluid/utils/http_client.py` ```python @classmethod async def response_error(cls, response: HttpResponse) -> None: try: data = await cls.response_data(response) except Exception: data = {"message": await response.text()} raise cls.ResponseError(response, data) ``` ### response_data ```python response_data(response) ``` Source code in `fluid/utils/http_client.py` ```python @classmethod async def response_data(cls, response: HttpResponse) -> ResponseType: content_type = response.headers.get("content-type", "") if "json" in content_type: return await response.json() elif "text" in content_type: return await response.text() return await response.bytes() ``` # Settings Process-wide configuration, sourced from environment variables. Settings cover the task consumer concurrency and timings, the broker and database connections, the HTTP client user agent, pagination defaults, the console backdoor and the stack sampler. It can be imported from `fluid.settings`: ```python from fluid.settings import get_settings settings = get_settings() settings.max_concurrent_tasks ``` The environment is read the first time get_settings is called, not at import time, so an application can populate the environment before the first access. The instance is then cached for the lifetime of the process. ## Environment variable names Most fields are read from `FLUID_`, and names are case insensitive, so `FLUID_MAX_CONCURRENT_TASKS` and `fluid_max_concurrent_tasks` both set `max_concurrent_tasks`. A few fields keep a conventional external name instead, with no prefix: | Field | Environment variable | | ----------------------- | ----------------------- | | `app_name` | `APP_NAME` | | `env` | `PYTHON_ENV` | | `log_level` | `LOG_LEVEL` | | `log_handler` | `LOG_HANDLER` | | `python_log_format` | `PYTHON_LOG_FORMAT` | | `database` | `DATABASE` | | `redis_default_url` | `REDIS_DEFAULT_URL` | | `redis_max_connections` | `MAX_REDIS_CONNECTIONS` | Warning The prefixed form does not work for the fields in the table above. Setting `FLUID_APP_NAME` has no effect, the value is read from `APP_NAME` only. The prefix itself can be changed with `FLUID_ENV_PREFIX`, which is read when `fluid.settings` is imported, so it has to be set before the first import of the library: ```bash FLUID_ENV_PREFIX=svc_ SVC_MAX_CONCURRENT_TASKS=10 python -m myapp serve ``` ## Derived defaults Three values are computed after the environment is read, when they are not set explicitly: - `broker_url` falls back to `redis_default_url`, so pointing `REDIS_DEFAULT_URL` at a Redis instance is enough to move the task queue with it. - `http_user_agent` falls back to `python/{app_name}`. - `log_level` is upper cased, so `LOG_LEVEL=info` and `LOG_LEVEL=INFO` are equivalent. ## Reading settings in tests get_settings caches its result, so a test that changes the environment has to clear the cache for the change to take effect: ```python import os from fluid.settings import get_settings os.environ["FLUID_MAX_CONCURRENT_TASKS"] = "1" get_settings.cache_clear() ``` ## API reference ## fluid.settings.Settings Bases: `BaseSettings` Lazy application settings sourced from environment variables. Settings are read from the environment the first time get_settings is called, not at import time. Access the resolved values either via the cached instance or, for backwards compatibility, via upper-case module attributes (`settings.APP_NAME`), both of which resolve lazily. ### model_config ```python model_config = SettingsConfigDict( case_sensitive=False, extra="ignore", env_prefix=ENV_PREFIX, ) ``` ### app_name ```python app_name = Field( default="fluid", validation_alias="APP_NAME" ) ``` ### env ```python env = Field(default='dev', validation_alias='PYTHON_ENV') ``` ### log_level ```python log_level = Field( default="info", validation_alias="LOG_LEVEL" ) ``` ### log_handler ```python log_handler = Field( default="plain", validation_alias="LOG_HANDLER" ) ``` ### python_log_format ```python python_log_format = Field( default="%(asctime)s %(levelname)s %(name)s %(message)s", validation_alias="PYTHON_LOG_FORMAT", ) ``` ### database ```python database = Field( default="postgresql+asyncpg://postgres:postgres@localhost:5432/fluid", validation_alias="DATABASE", ) ``` ### redis_default_url ```python redis_default_url = Field( default="redis://localhost:6379", validation_alias="REDIS_DEFAULT_URL", ) ``` ### stopping_grace_period ```python stopping_grace_period = 10 ``` ### max_concurrent_tasks ```python max_concurrent_tasks = Field( default=5, description="Maximum number of concurrent tasks per TaskConsumer", ) ``` ### sleep_millis ```python sleep_millis = 1000 ``` ### scheduler_heartbeat_millis ```python scheduler_heartbeat_millis = 100 ``` ### broker_url ```python broker_url = '' ``` ### redis_max_connections ```python redis_max_connections = Field( default=5, validation_alias="MAX_REDIS_CONNECTIONS" ) ``` ### database_schema ```python database_schema = None ``` ### dbpool_max_size ```python dbpool_max_size = 10 ``` ### dbpool_max_overflow ```python dbpool_max_overflow = 10 ``` ### dbecho ```python dbecho = False ``` ### http_user_agent ```python http_user_agent = '' ``` ### default_pagination_limit ```python default_pagination_limit = 250 ``` ### default_pagination_max_limit ```python default_pagination_max_limit = 500 ``` ### backdoor_port ```python backdoor_port = 8087 ``` ### flamegraph_executable ```python flamegraph_executable = 'flamegraph.pl' ``` ### stack_sampler_period_seconds ```python stack_sampler_period_seconds = 1 ``` ## fluid.settings.get_settings ```python get_settings() ``` Return the process-wide Settings instance. The instance is built on first call (reading the environment then) and cached for the lifetime of the process. Call `get_settings.cache_clear()` to force a re-read, which is mostly useful in tests. Source code in `fluid/settings.py` ```python @lru_cache(maxsize=1) def get_settings() -> Settings: """Return the process-wide [Settings][fluid.settings.Settings] instance. The instance is built on first call (reading the environment then) and cached for the lifetime of the process. Call ``get_settings.cache_clear()`` to force a re-read, which is mostly useful in tests. """ return Settings() ``` # Task A Task defines the implementation of a given operation, the inputs required and the scheduling metadata. Usually, a Task is not created directly, but rather through the use of the @task decorator. ## Example A task function is decorated via the @task decorator and must accept the TaskRun object as its first and only argument. ```python from fluid.scheduler import task, TaskRun @task async def hello(ctx: TaskRun) -> None: print("Hello, world!") ``` For retry configuration (`retry`, `rate_limit_retry`) see [Task Retry](https://fluid.quantmind.com/reference/task_retry/index.md). ## fluid.scheduler.task ```python task(executor: TaskExecutor) -> Task ``` ```python task( *, name: str | None = None, schedule: Scheduler | None = None, short_description: str | None = None, description: str | None = None, randomize: RandomizeType | None = None, max_concurrency: int | None = None, priority: TaskPriority | None = None, cpu_bound: bool | None = None, k8s_config: K8sConfig | None = None, timeout_seconds: int | None = None, tags: Sequence[str] | None = None, retry: RetryPolicy | None = None, rate_limit_retry: RetryPolicy | None = None, env: dict[str, str] | None = None ) -> TaskConstructor ``` ```python task( executor=None, *, name=None, schedule=None, short_description=None, description=None, randomize=None, max_concurrency=None, priority=None, cpu_bound=None, k8s_config=None, timeout_seconds=None, tags=None, retry=None, rate_limit_retry=None, env=None ) ``` Decorator to create a Task from a function and optional parameters. This decorator can be used in two ways: - As a simple decorator of the executor function - As a function with keyword arguments for greater control over the task configuration | PARAMETER | DESCRIPTION | | ------------------- | --------------------------------------------------------------------------------------------------------------------------------- | | `executor` | The executor function for the task **TYPE:** \`TaskExecutor | | `name` | The name of the task. If None, the name will be derived from the executor function **TYPE:** \`str | | `schedule` | The schedule for the tas. If None, the task will not be scheduled **TYPE:** \`Scheduler | | `short_description` | A short description of the task. If not provided it will be extracted from the task function docstring first line **TYPE:** \`str | | `description` | A detailed description of the task. If not provided it will be extracted from the task function docstring **TYPE:** \`str | | `randomize` | Randomization settings for the task **TYPE:** \`RandomizeType | | `max_concurrency` | The maximum number of concurrent executions of the task **TYPE:** \`int | | `priority` | The priority of the task such as high, medium, low **TYPE:** \`TaskPriority | | `cpu_bound` | Whether the task is CPU bound **TYPE:** \`bool | | `k8s_config` | Kubernetes configuration - None means use the default configuration **TYPE:** \`K8sConfig | | `timeout_seconds` | Task timeout in seconds - how long the task can run before being aborted **TYPE:** \`int | | `tags` | Task tags - used for categorization and filtering of tasks **TYPE:** \`Sequence[str] | | `retry` | Retry policy for execution failures **TYPE:** \`RetryPolicy | | `rate_limit_retry` | Retry policy when the task is rate limited by max_concurrency **TYPE:** \`RetryPolicy | | `env` | Extra environment variables injected into the subprocess or k8s job **TYPE:** \`dict[str, str] | Source code in `fluid/scheduler/models.py` ```python def task( executor: Annotated[ TaskExecutor | None, Doc("The executor function for the task"), ] = None, *, name: Annotated[ str | None, Doc( ( "The name of the task. If None, the name will be derived " "from the executor function" ) ), ] = None, schedule: Annotated[ Scheduler | None, Doc("The schedule for the tas. If None, the task will not be scheduled"), ] = None, short_description: Annotated[ str | None, Doc( ( "A short description of the task. " "If not provided it will be extracted from the task function docstring " "first line" ) ), ] = None, description: Annotated[ str | None, Doc( ( "A detailed description of the task. " "If not provided it will be extracted from the task function docstring" ) ), ] = None, randomize: Annotated[ RandomizeType | None, Doc("Randomization settings for the task"), ] = None, max_concurrency: Annotated[ int | None, Doc(("The maximum number of concurrent executions of the task")), ] = None, priority: Annotated[ TaskPriority | None, Doc("The priority of the task such as high, medium, low"), ] = None, cpu_bound: Annotated[ bool | None, Doc("Whether the task is CPU bound"), ] = None, k8s_config: Annotated[ K8sConfig | None, Doc("Kubernetes configuration - None means use the default configuration"), ] = None, timeout_seconds: Annotated[ int | None, Doc("Task timeout in seconds - how long the task can run before being aborted"), ] = None, tags: Annotated[ Sequence[str] | None, Doc("Task tags - used for categorization and filtering of tasks"), ] = None, retry: Annotated[ RetryPolicy | None, Doc("Retry policy for execution failures"), ] = None, rate_limit_retry: Annotated[ RetryPolicy | None, Doc("Retry policy when the task is rate limited by max_concurrency"), ] = None, env: Annotated[ dict[str, str] | None, Doc("Extra environment variables injected into the subprocess or k8s job"), ] = None, ) -> Task | TaskConstructor: """Decorator to create a [Task][fluid.scheduler.Task] from a function and optional parameters. This decorator can be used in two ways: - As a simple decorator of the executor function - As a function with keyword arguments for greater control over the task configuration """ kwargs = compact_dict( name=name, schedule=schedule, short_description=short_description, description=description, randomize=randomize, max_concurrency=max_concurrency, priority=priority, cpu_bound=cpu_bound, k8s_config=k8s_config, timeout_seconds=timeout_seconds, tags=frozenset(tags) if tags is not None else None, retry=retry, rate_limit_retry=rate_limit_retry, env=env, ) if kwargs and executor: raise TaskDecoratorError("cannot use positional parameters") elif kwargs: return TaskConstructor(**kwargs) elif not executor: raise TaskDecoratorError("this is a decorator cannot be invoked in this way") else: return TaskConstructor()(executor) ``` ## fluid.scheduler.Task Bases: `NamedTuple`, `Generic[TP]` A Task configuration. This is not created directly, but rather through the use of the @task decorator. Executes any time it is invoked ### name ```python name ``` Task name - unique identifier ### executor ```python executor ``` Task executor function ### params_model ```python params_model ``` Pydantic model for task parameters ### logger ```python logger ``` Task logger ### module ```python module = '' ``` Task python module ### short_description ```python short_description = '' ``` Short task description - one line ### description ```python description = '' ``` Task description - obtained from the executor docstring if not provided ### schedule ```python schedule = None ``` Task schedule - None means the task is not scheduled ### randomize ```python randomize = None ``` Randomize function for task schedule ### max_concurrency ```python max_concurrency = 0 ``` how many tasks can be run concurrently - 0 means no limit ### timeout_seconds ```python timeout_seconds = 60 ``` Task timeout in seconds - how long the task can run before being aborted ### priority ```python priority = TaskPriority.medium ``` Task priority - high, medium, low ### k8s_config ```python k8s_config = None ``` Kubernetes configuration for tasks run on Kubernetes cluster. ### tags ```python tags = frozenset() ``` Task tags - used for categorization and filtering of tasks ### retry ```python retry = None ``` Retry policy for general execution failures. ### rate_limit_retry ```python rate_limit_retry = None ``` Retry policy when the executor raises `RateLimitError`. ### env ```python env = {} ``` Extra environment variables injected into the subprocess or k8s job. ### cpu_bound ```python cpu_bound ``` True if the task is CPU bound ### get_k8s_config ```python get_k8s_config() ``` Get Kubernetes configuration for this task Source code in `fluid/scheduler/models.py` ```python def get_k8s_config(self) -> K8sConfig: """Get Kubernetes configuration for this task""" return self.k8s_config or K8sConfig() ``` ### info ```python info(**params) ``` Return task info object Source code in `fluid/scheduler/models.py` ```python def info(self, **params: Any) -> TaskInfo: """Return task info object""" params.update( name=self.name, description=self.description, module=self.module, priority=self.priority, schedule=str(self.schedule) if self.schedule else None, tags=self.tags, ) return TaskInfo(**compact_dict(params)) ``` ## fluid.scheduler.TaskPriority Bases: `StrEnum` Priority level for task execution ordering. ### high ```python high = enum.auto() ``` Execute before medium and low priority tasks. ### medium ```python medium = enum.auto() ``` Default priority level. ### low ```python low = enum.auto() ``` Execute after high and medium priority tasks. ## fluid.scheduler.TaskState Bases: `StrEnum` Lifecycle state of a task run. ### init ```python init = enum.auto() ``` Task has been created but not yet queued. ### queued ```python queued = enum.auto() ``` Task is waiting in the queue to be picked up by a worker. ### running ```python running = enum.auto() ``` Task is currently being executed. ### success ```python success = enum.auto() ``` Task completed successfully. ### failure ```python failure = enum.auto() ``` Task raised an exception during execution. ### aborted ```python aborted = enum.auto() ``` Task was cancelled before completion. ### rate_limited ```python rate_limited = enum.auto() ``` Task execution was deferred due to rate limiting. ### interrupted ```python interrupted = enum.auto() ``` Task was interrupted by a worker shutdown before it could complete. ### is_failure ```python is_failure ``` Return True if this state is a failure state ### is_done ```python is_done ``` Return True if this state is a finished state ## fluid.scheduler.K8sConfig Bases: `BaseModel` Kubernetes configuration for tasks run on Kubernetes cluster. This configuration is used by the task consumer to run tasks on Kubernetes Jobs. ```python from fluid.scheduler import K8sConfig ``` This is used when the task consumer runs inside a Kubernetes cluster and the task is marked as CPU bound. Fields: - `namespace` (`str`) - `deployment` (`str`) - `container` (`str`) - `resources` (`K8sResourceRequirements | None`) - `job_ttl` (`int`) - `sleep` (`float`) ### namespace ```python namespace ``` Kubernetes namespace where the task consumer deployment run ### deployment ```python deployment ``` Kubernetes deployment of the task consumer ### container ```python container ``` Kubernetes container ### resources ```python resources = None ``` Kubernetes resource limits and requests for the container ### job_ttl ```python job_ttl ``` Time to live for k8s Job after completion ### sleep ```python sleep ``` Amount to async sleep while waiting for completion of k8s Job ## fluid.scheduler.K8sResourceRequirements Bases: `TypedDict` CPU and memory limits/requests for a Kubernetes container. ### limits ```python limits ``` ### requests ```python requests ``` ## fluid.scheduler.is_in_cpu_process ```python is_in_cpu_process() ``` Check if the current process is a CPU process. A CPU process is a process that is spawned by the task manager to run a cpu-bound task. It is identified by the environment variable `TASK_MANAGER_SPAWN` being set to "true". Source code in `fluid/scheduler/common.py` ```python def is_in_cpu_process() -> bool: """Check if the current process is a CPU process. A CPU process is a process that is spawned by the task manager to run a cpu-bound task. It is identified by the environment variable `TASK_MANAGER_SPAWN` being set to "true". """ return os.getenv("TASK_MANAGER_SPAWN") == "true" ``` # Task Broker It can be imported from `fluid.scheduler`: ```python from fluid.scheduler import TaskBroker ``` ## fluid.scheduler.TaskBroker ```python TaskBroker(url) ``` Bases: `ABC` Abstract base class for task brokers A TaskBroker is responsible for queuing tasks & storing task information Source code in `fluid/scheduler/broker.py` ```python def __init__(self, url: URL) -> None: self.url: Annotated[URL, Doc("Broker URL")] = url self.registry: Annotated[TaskRegistry, Doc("Task registry")] = TaskRegistry() ``` ### url ```python url = url ``` Broker URL ### registry ```python registry = TaskRegistry() ``` Task registry ### task_queue_names ```python task_queue_names ``` Names of the task queues ### queue_task ```python queue_task(task_run) ``` Queue a task run This method is called by the TaskManager when a task run is ready to be executed. The broker is responsible for adding the task run to the appropriate queue based on its priority. | PARAMETER | DESCRIPTION | | ---------- | ---------------------------- | | `task_run` | Task run **TYPE:** `TaskRun` | Source code in `fluid/scheduler/broker.py` ```python @abstractmethod async def queue_task(self, task_run: Annotated[TaskRun, Doc("Task run")]) -> None: """Queue a task run This method is called by the [TaskManager][fluid.scheduler.TaskManager] when a task run is ready to be executed. The broker is responsible for adding the task run to the appropriate queue based on its priority. """ ``` ### get_task_run ```python get_task_run(task_manager) ``` Get a Task run from the task queue Source code in `fluid/scheduler/broker.py` ```python @abstractmethod async def get_task_run(self, task_manager: TaskManager) -> TaskRun | None: """Get a Task run from the task queue""" ``` ### queue_length ```python queue_length() ``` Length of task queues Source code in `fluid/scheduler/broker.py` ```python @abstractmethod async def queue_length(self) -> dict[str, int]: """Length of task queues""" ``` ### clear_queue ```python clear_queue(*priorities) ``` Clear task queues, returns number of removed items per priority Source code in `fluid/scheduler/broker.py` ```python @abstractmethod async def clear_queue(self, *priorities: TaskPriority) -> dict[str, int]: """Clear task queues, returns number of removed items per priority""" ``` ### set_manager_status ```python set_manager_status(manager_id, data, ttl) ``` Store the status of a running task manager Source code in `fluid/scheduler/broker.py` ```python @abstractmethod async def set_manager_status(self, manager_id: str, data: dict, ttl: int) -> None: """Store the status of a running task manager""" ``` ### get_all_manager_statuses ```python get_all_manager_statuses() ``` Get statuses of all running task managers Source code in `fluid/scheduler/broker.py` ```python @abstractmethod async def get_all_manager_statuses(self) -> TaskManagersStatus: """Get statuses of all running task managers""" ``` ### get_tasks_info ```python get_tasks_info(*task_names) ``` List of TaskInfo objects Source code in `fluid/scheduler/broker.py` ```python @abstractmethod async def get_tasks_info(self, *task_names: str) -> list[TaskInfo]: """List of TaskInfo objects""" ``` ### update_task ```python update_task(task, params) ``` Update a task dynamic parameters Source code in `fluid/scheduler/broker.py` ```python @abstractmethod async def update_task(self, task: Task, params: dict[str, Any]) -> TaskInfo: """Update a task dynamic parameters""" ``` ### add_task_run ```python add_task_run(task_run) ``` Add a task run to the broker Source code in `fluid/scheduler/broker.py` ```python @abstractmethod async def add_task_run(self, task_run: TaskRun) -> None: """Add a task run to the broker""" ``` ### remove_task_run ```python remove_task_run(task_run) ``` Remove a task run from the broker Source code in `fluid/scheduler/broker.py` ```python @abstractmethod async def remove_task_run(self, task_run: TaskRun) -> None: """Remove a task run from the broker""" ``` ### current_task_runs ```python current_task_runs(task_name) ``` The number of current task runs for a given task_name Source code in `fluid/scheduler/broker.py` ```python @abstractmethod async def current_task_runs(self, task_name: str) -> int: """The number of current task runs for a given task_name""" ``` ### set_task_aborted ```python set_task_aborted(run_id, reason) ``` Signal that a task run was aborted, storing the reason Source code in `fluid/scheduler/broker.py` ```python @abstractmethod async def set_task_aborted(self, run_id: str, reason: str) -> None: """Signal that a task run was aborted, storing the reason""" ``` ### get_task_aborted ```python get_task_aborted(run_id) ``` Return the abort reason for a task run, or None if not aborted Source code in `fluid/scheduler/broker.py` ```python @abstractmethod async def get_task_aborted(self, run_id: str) -> str | None: """Return the abort reason for a task run, or None if not aborted""" ``` ### close ```python close() ``` Close the broker on shutdown Source code in `fluid/scheduler/broker.py` ```python @abstractmethod async def close(self) -> None: """Close the broker on shutdown""" ``` ### lock ```python lock(name, timeout=None) ``` Create a lock Source code in `fluid/scheduler/broker.py` ```python @abstractmethod def lock(self, name: str, timeout: float | None = None) -> Lock: """Create a lock""" ``` ### new_uuid ```python new_uuid() ``` Source code in `fluid/scheduler/broker.py` ```python def new_uuid(self) -> str: return uuid4().hex ``` ### filter_tasks ```python filter_tasks(scheduled=None, enabled=None) ``` Source code in `fluid/scheduler/broker.py` ```python async def filter_tasks( self, scheduled: bool | None = None, enabled: bool | None = None, ) -> list[Task]: task_info = await self.get_tasks_info() task_map = {info.name: info for info in task_info} tasks = [] for task in self.registry.values(): if scheduled is not None and bool(task.schedule) is not scheduled: continue if enabled is not None and task_map[task.name].enabled is not enabled: continue tasks.append(task) return tasks ``` ### task_from_registry ```python task_from_registry(task) ``` Source code in `fluid/scheduler/broker.py` ```python def task_from_registry(self, task: str | Task) -> Task: if isinstance(task, Task): self.register_task(task) return task else: if task_ := self.registry.get(task): return task_ raise UnknownTaskError(task) ``` ### register_task ```python register_task(task) ``` Source code in `fluid/scheduler/broker.py` ```python def register_task(self, task: Task) -> None: self.registry[task.name] = task ``` ### enable_task ```python enable_task(task, enable=True) ``` Enable or disable a registered task Source code in `fluid/scheduler/broker.py` ```python async def enable_task(self, task: str | Task, enable: bool = True) -> TaskInfo: """Enable or disable a registered task""" task_ = self.task_from_registry(task) return await self.update_task(task_, dict(enabled=enable)) ``` ### from_url ```python from_url(url='') ``` Source code in `fluid/scheduler/broker.py` ```python @classmethod def from_url(cls, url: str = "") -> TaskBroker: p = URL(url or broker_url_from_env()) if factory := _brokers.get(p.scheme): return factory(p) raise RuntimeError(f"Invalid broker {p}") ``` ### register_broker ```python register_broker(name, factory) ``` Source code in `fluid/scheduler/broker.py` ```python @classmethod def register_broker(cls, name: str, factory: type[TaskBroker]) -> None: _brokers[name] = factory ``` # Task Manager CLI Command line tools for TaskManager applications. This module requires the `cli` extra to be installed: ```bash pip install aio-fluid[cli] ``` ## Setup Wrap your FastAPI app with TaskManagerCLI and call it as the entry point: ```python from fluid.scheduler.cli import TaskManagerCLI task_manager_cli = TaskManagerCLI("examples.tasks:task_app") if __name__ == "__main__": task_manager_cli() ``` `task_manager_app` can be a FastAPI instance, a callable that returns one, or a dotted-import string (resolved at call time so the CLI stays importable without loading the full app). ## Commands ### `ls` — list registered tasks Prints a table of all registered tasks with their schedule, priority, timeout, CPU-bound flag, and tags. ```bash python -m myapp ls ``` Pass `--tags`/`-t` (repeatable) to only list tasks that have at least one of the given tags: ```bash python -m myapp ls --tags fast --tags slow ``` ### `serve` — start the HTTP server ```bash python -m myapp serve --host 0.0.0.0 --port 8080 ``` Delegates to `uvicorn`. Pass `--reload` for development auto-reload. ### `enable` — enable or disable a task ```bash python -m myapp enable # enable python -m myapp enable --disable # disable ``` ### `exec` — execute a task Runs a registered task synchronously and prints the result table. ```bash python -m myapp exec [OPTIONS] ``` Each task exposes its params model fields as CLI options (via [pydanclick](https://github.com/felix-martel/pydanclick)): ```bash python -m myapp exec add --a 5 --b 3 ``` #### Passing parameters as JSON All task parameters can also be supplied together as a JSON string with `--params`. Values in `--params` **always take priority** over individual CLI option defaults: ```bash python -m myapp exec add --params '{"a": 5, "b": 3}' ``` This is useful when parameters are complex types or when scripting task execution. If both `--params` and individual options are provided, the JSON values win for any overlapping keys. ```bash # error=true from --params overrides the model default error=false python -m myapp exec fast --params '{"error": true}' ``` #### `--run-id` Pass a custom run ID to correlate the task run with external systems: ```bash python -m myapp exec add --run-id my-custom-id --params '{"a": 1, "b": 2}' ``` #### `--log` Enable structured logging output during execution: ```bash python -m myapp exec add --log --params '{"a": 1, "b": 2}' ``` ## API reference ## fluid.scheduler.cli.TaskManagerCLI ```python TaskManagerCLI(task_manager_app, log_config=None, **kwargs) ``` Bases: `LazyGroup` CLI for TaskManager This class provides a CLI for a TaskManager Application. It requires to install the `cli` extra dependencies. | PARAMETER | DESCRIPTION | | ------------------ | ---------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `task_manager_app` | Task manager application. This can be a FastAPI app, a callable that returns a FastAPI app, or a string import path to a FastAPI app. **TYPE:** `TaskManagerApp` | | `log_config` | Log configuration parameters. These parameters are passed to the log.config function when configuring logging. **TYPE:** \`dict | Source code in `fluid/scheduler/cli.py` ```python def __init__( self, task_manager_app: Annotated[ TaskManagerApp, Doc(""" Task manager application. This can be a FastAPI app, a callable that returns a FastAPI app, or a string import path to a FastAPI app. """), ], log_config: Annotated[ dict | None, Doc(""" Log configuration parameters. These parameters are passed to the [log.config][fluid.utils.log.config] function when configuring logging. """), ] = None, **kwargs: Any, ): kwargs.setdefault("commands", DEFAULT_COMMANDS) super().__init__(**kwargs) self.task_manager_app = task_manager_app self.log_config = log_config or {} ``` ### task_manager_app ```python task_manager_app = task_manager_app ``` ### log_config ```python log_config = log_config or {} ``` ### lazy_subcommands ```python lazy_subcommands = lazy_subcommands or {} ``` ### get_task_manager_app ```python get_task_manager_app() ``` Get the FastAPI app for the TaskManager. Source code in `fluid/scheduler/cli.py` ```python def get_task_manager_app(self) -> FastAPI: """Get the FastAPI app for the TaskManager.""" if isinstance(self.task_manager_app, str): return import_from_string(self.task_manager_app)() elif isinstance(self.task_manager_app, FastAPI): return self.task_manager_app else: return self.task_manager_app() ``` ### get_task_manager ```python get_task_manager() ``` Get the TaskManager instance from the app state. Source code in `fluid/scheduler/cli.py` ```python def get_task_manager(self) -> TaskManager: """Get the TaskManager instance from the app state.""" app = self.get_task_manager_app() if not hasattr(app.state, "task_manager"): raise RuntimeError("The app does not have a task_manager in its state") return app.state.task_manager ``` ### list_commands ```python list_commands(ctx) ``` Source code in `fluid/utils/lazy.py` ```python def list_commands(self, ctx: click.Context) -> list[str]: commands = super().list_commands(ctx) commands.extend(self.lazy_subcommands) return sorted(commands) ``` ### get_command ```python get_command(ctx, cmd_name) ``` Source code in `fluid/utils/lazy.py` ```python def get_command(self, ctx: click.Context, cmd_name: str) -> click.Command | None: if cmd_name in self.lazy_subcommands: return self._lazy_load(cmd_name) return super().get_command(ctx, cmd_name) ``` # Task Consumer The task consumer is a TaskManager which is also a Workers that consumes tasks from the task queue and executes them. It can be imported from `fluid.scheduler`: ```python from fluid.scheduler import TaskConsumer ``` ## fluid.scheduler.TaskConsumer ```python TaskConsumer( *, deps=None, config=None, name="", stopping_grace_period=None, **kwargs ) ``` Bases: `TaskManager`, `Workers` The Task Consumer is a Task Manager responsible for consuming tasks from a task queue Create the task consumer and the workers it runs. Workers added via `add_workers` are started by `startup`, therefore they only run in a process which consumes the task queue. The async dispatcher is added as an async context manager instead, so it is started by `__aenter__` and runs in every process which uses the task manager. A cpu bound task runs in a process of its own, which executes a single task and never starts the consumer workers, and it still needs to dispatch the lifecycle events its plugins and handlers subscribe to. | PARAMETER | DESCRIPTION | | ----------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `deps` | Application dependencies available to every task run. See the Task Dependencies tutorial. **TYPE:** `Any` **DEFAULT:** `None` | | `config` | Task manager configuration. Built from the extra keyword arguments when not provided. **TYPE:** \`TaskManagerConfig | | `name` | Worker's name, if not provided it is evaluated from the class name **TYPE:** `str` **DEFAULT:** `''` | | `stopping_grace_period` | Grace period in seconds to wait for workers to stop running when this worker is shutdown. It defaults to the FLUID_STOPPING_GRACE_PERIOD environment variable or 10 seconds. **TYPE:** \`float | | `**kwargs` | Configuration fields, used when config is not provided. **TYPE:** `Any` **DEFAULT:** `{}` | Source code in `fluid/scheduler/consumer.py` ```python def __init__( self, *, deps: Annotated[ Any, Doc(""" Application dependencies available to every task run. See the [Task Dependencies](../tutorials/task_deps.md) tutorial. """), ] = None, config: Annotated[ TaskManagerConfig | None, Doc(""" Task manager configuration. Built from the extra keyword arguments when not provided. """), ] = None, name: Annotated[ str, Doc("Worker's name, if not provided it is evaluated from the class name"), ] = "", stopping_grace_period: Annotated[ float | None, Doc( "Grace period in seconds to wait for workers to stop running " "when this worker is shutdown. " "It defaults to the `FLUID_STOPPING_GRACE_PERIOD` " "environment variable or 10 seconds." ), ] = None, **kwargs: Annotated[ Any, Doc("Configuration fields, used when `config` is not provided."), ], ) -> None: """Create the task consumer and the workers it runs. Workers added via `add_workers` are started by `startup`, therefore they only run in a process which consumes the task queue. The async dispatcher is added as an async context manager instead, so it is started by `__aenter__` and runs in every process which uses the task manager. A cpu bound task runs in a process of its own, which executes a single task and never starts the consumer workers, and it still needs to dispatch the lifecycle events its plugins and handlers subscribe to. """ super().__init__(deps=deps, config=config, **kwargs) if stopping_grace_period is None: stopping_grace_period = settings.STOPPING_GRACE_PERIOD Workers.__init__( self, name=name, stopping_grace_period=2 * stopping_grace_period ) self._async_dispatcher_worker = AsyncConsumer( AsyncTaskDispatcher(), stopping_grace_period=stopping_grace_period ) self._in_process_queue = InProcessTaskQueue( self, stopping_grace_period=stopping_grace_period ) self.add_async_context_manager(self._async_dispatcher_worker) self.add_workers(self._in_process_queue) for i in range(self.config.max_concurrent_tasks): worker_name = f"task-worker-{i+1}" self.add_workers( WorkerFunction( partial(self._consume_tasks, worker_name), name=worker_name, stopping_grace_period=stopping_grace_period, ) ) self.add_workers( WorkerFunction( self._ping_status, heartbeat=1.0, name="manager-status", stopping_grace_period=stopping_grace_period, ) ) ``` ### worker_state ```python worker_state ``` The running state of the worker ### worker_name ```python worker_name ``` The name of the worker ### num_workers ```python num_workers ``` ### deps ```python deps = deps if deps is not None else State() ``` Dependencies for the task manager. Production applications requires global dependencies to be available to all tasks. This can be achieved by setting the `deps` attribute of the task manager to an object with the required dependencies. Each task can cast the dependencies to the required type. ### state ```python state = State() ``` State for the task manager. This can be used by plugins to store state in the task manager. ### config ```python config = config or TaskManagerConfig(**kwargs) ``` Task manager configuration ### dispatcher ```python dispatcher = TaskDispatcher() ``` A dispatcher of TaskRun events. Application can register handlers to listen for events happening during the lifecycle of a task run. ### broker ```python broker = TaskBroker.from_url(self.config.broker_url) ``` ### manager_id ```python manager_id = self.broker.new_uuid() ``` ### registry ```python registry ``` The task registry ### type ```python type ``` The type of the task manager ### has_started ```python has_started() ``` Source code in `fluid/utils/worker.py` ```python def has_started(self) -> bool: return self._worker_state != WorkerState.INIT ``` ### is_running ```python is_running() ``` Source code in `fluid/utils/worker.py` ```python def is_running(self) -> bool: return self._worker_state == WorkerState.RUNNING ``` ### is_stopping ```python is_stopping() ``` Source code in `fluid/utils/worker.py` ```python def is_stopping(self) -> bool: return self._worker_state == WorkerState.STOPPING ``` ### is_stopped ```python is_stopped() ``` Source code in `fluid/utils/worker.py` ```python def is_stopped(self) -> bool: return self._worker_state in (WorkerState.STOPPED, WorkerState.FORCE_STOPPED) ``` ### gracefully_stop ```python gracefully_stop() ``` Try to gracefully stop the workers and this worker Source code in `fluid/utils/worker.py` ```python def gracefully_stop(self) -> None: """Try to gracefully stop the workers and this worker""" super().gracefully_stop() for worker in self._workers: worker.gracefully_stop() ``` ### after_shutdown ```python after_shutdown(reason, code) ``` Called after shutdown of worker By default it does nothing, but can be overriden to do something such as exit the process. Source code in `fluid/utils/worker.py` ```python def after_shutdown(self, reason: str, code: int) -> None: # noqa: B027 """Called after shutdown of worker By default it does nothing, but can be overriden to do something such as exit the process. """ ``` ### status ```python status() ``` Source code in `fluid/utils/worker.py` ```python async def status(self) -> dict: status_workers = await asyncio.gather( *[worker.status() for worker in self._workers], ) return { worker.worker_name: status for worker, status in zip(self._workers, status_workers, strict=False) } ``` ### on_startup ```python on_startup() ``` Source code in `fluid/scheduler/consumer.py` ```python async def on_startup(self) -> None: await self.__aenter__() ``` ### on_shutdown ```python on_shutdown() ``` Source code in `fluid/scheduler/consumer.py` ```python async def on_shutdown(self) -> None: await self.__aexit__(None, None, None) ``` ### shutdown ```python shutdown() ``` Shutdown a running worker and wait for it to stop This method will try to gracefully stop the worker and wait for it to stop. If the worker does not stop in the grace period, it will force shutdown by cancelling the task. Source code in `fluid/utils/worker.py` ```python async def shutdown(self) -> None: """Shutdown a running worker and wait for it to stop This method will try to gracefully stop the worker and wait for it to stop. If the worker does not stop in the grace period, it will force shutdown by cancelling the task. """ if self._worker_task_runner is not None: await self._worker_task_runner.shutdown() ``` ### wait_for_shutdown ```python wait_for_shutdown() ``` Wait for the worker to stop This method will wait for the worker to stop running, but doesn't try to gracefully stop it nor force shutdown. Source code in `fluid/utils/worker.py` ```python async def wait_for_shutdown(self) -> None: """Wait for the worker to stop This method will wait for the worker to stop running, but doesn't try to gracefully stop it nor force shutdown. """ if self._worker_task_runner is not None: await self._worker_task_runner.wait_for_shutdown() ``` ### workers ```python workers() ``` Source code in `fluid/utils/worker.py` ```python def workers(self) -> Iterator[Worker]: return iter(self._workers) ``` ### run ```python run() ``` Source code in `fluid/utils/worker.py` ```python async def run(self) -> None: while self.is_running(): for worker in self._workers: if not worker.has_started(): await worker.startup() if not worker.is_running(): self.gracefully_stop() break await asyncio.sleep(self._heartbeat) await self._wait_for_workers() ``` ### add_workers ```python add_workers(*workers) ``` add workers to the workers They can be added while the worker is running. Source code in `fluid/utils/worker.py` ```python def add_workers(self, *workers: Worker) -> None: """add workers to the workers They can be added while the worker is running. """ for worker in workers: if worker not in self._workers: self._workers.append(worker) ``` ### add_async_context_manager ```python add_async_context_manager(cm) ``` Add an async context manager to the task manager These context managers are entered when the task manager starts Source code in `fluid/scheduler/consumer.py` ```python def add_async_context_manager(self, cm: Any) -> None: """Add an async context manager to the task manager These context managers are entered when the task manager starts """ self._async_contexts.append(cm) ``` ### register_task ```python register_task(task, tags=None) ``` Register a task with the task manager | PARAMETER | DESCRIPTION | | --------- | ----------------------------------------------------------------------------- | | `task` | Task to register **TYPE:** `Task` | | `tags` | Extra tags to add to the task before registering it **TYPE:** \`Sequence[str] | Source code in `fluid/scheduler/consumer.py` ```python def register_task( self, task: Annotated[Task, Doc("Task to register")], tags: Annotated[ Sequence[str] | None, Doc("Extra tags to add to the task before registering it"), ] = None, ) -> None: """Register a task with the task manager""" if tags: task = task._replace(tags=task.tags | frozenset(tags)) self.broker.register_task(task) ``` ### execute ```python execute(task, *, run_id='', priority=None, **params) ``` Execute a task and wait for it to finish This method is an async method that should be used in an asynchronous context when one need to wait for the task to finish execution. | PARAMETER | DESCRIPTION | | ---------- | ----------------------------------------------------------------------------------------------------------------- | | `task` | The task or task name, if a task name it must be registered with the task manager. **TYPE:** \`str | | `run_id` | Unique ID for the task run. If not provided a new UUID is generated. **TYPE:** `str` **DEFAULT:** `''` | | `priority` | Override the default task priority if provided **TYPE:** \`TaskPriority | | `**params` | The optional parameters for the task run. They must match the task params model **TYPE:** `Any` **DEFAULT:** `{}` | Source code in `fluid/scheduler/consumer.py` ```python async def execute( self, task: Annotated[ str | Task, Doc( "The task or task name," " if a task name it must be registered with the task manager." ), ], *, run_id: Annotated[ str, Doc("Unique ID for the task run. If not provided a new UUID is generated."), ] = "", priority: Annotated[ TaskPriority | None, Doc("Override the default task priority if provided") ] = None, **params: Annotated[ Any, Doc( "The optional parameters for the task run. " "They must match the task params model" ), ], ) -> TaskRun: """Execute a task and wait for it to finish This method is an async method that should be used in an asynchronous context when one need to wait for the task to finish execution. """ task_run = self.create_task_run( task, run_id=run_id, priority=priority, **params, ) try: await task_run._execute() except TaskAbortedError as exc: await self.broker.set_task_aborted(task_run.id, str(exc)) return task_run ``` ### execute_sync ```python execute_sync(task, *, run_id='', priority=None, **params) ``` Execute a task synchronously This method is a blocking method that should be used in a synchronous context. | PARAMETER | DESCRIPTION | | ---------- | ----------------------------------------------------------------------------------------------------------------- | | `task` | The task or task name, if a task name it must be registered with the task manager. **TYPE:** \`str | | `run_id` | Unique ID for the task run. If not provided a new UUID is generated. **TYPE:** `str` **DEFAULT:** `''` | | `priority` | Override the default task priority if provided **TYPE:** \`TaskPriority | | `**params` | The optional parameters for the task run. They must match the task params model **TYPE:** `Any` **DEFAULT:** `{}` | Source code in `fluid/scheduler/consumer.py` ```python def execute_sync( self, task: Annotated[ str | Task, Doc( "The task or task name," " if a task name it must be registered with the task manager." ), ], *, run_id: Annotated[ str, Doc("Unique ID for the task run. If not provided a new UUID is generated."), ] = "", priority: Annotated[ TaskPriority | None, Doc("Override the default task priority if provided") ] = None, **params: Annotated[ Any, Doc( "The optional parameters for the task run. " "They must match the task params model" ), ], ) -> TaskRun: """Execute a task synchronously This method is a blocking method that should be used in a synchronous context. """ return asyncio.run( self._execute_and_exit( task, run_id=run_id, priority=priority, **params, ) ) ``` ### queue ```python queue( task, *, run_id="", priority=None, from_task_run=None, **params ) ``` Queue a task for execution This methods fires two events: - `init`: when the task run is created - `queued`: after the task is queued It returns the TaskRun object | PARAMETER | DESCRIPTION | | --------------- | ----------------------------------------------------------------------------------------------------------------- | | `task` | The task or task name, if a task name it must be registered with the task manager. **TYPE:** \`str | | `run_id` | Unique ID for the task run. If not provided a new UUID is generated. **TYPE:** `str` **DEFAULT:** `''` | | `priority` | Override the default task priority if provided **TYPE:** \`TaskPriority | | `from_task_run` | The task run queueing this one, if any. Prefer TaskRun.queue, which passes it for you. **TYPE:** \`TaskRun | | `**params` | The optional parameters for the task run. They must match the task params model **TYPE:** `Any` **DEFAULT:** `{}` | Source code in `fluid/scheduler/consumer.py` ```python async def queue( self, task: Annotated[ str | Task, Doc( "The task or task name," " if a task name it must be registered with the task manager." ), ], *, run_id: Annotated[ str, Doc("Unique ID for the task run. If not provided a new UUID is generated."), ] = "", priority: Annotated[ TaskPriority | None, Doc("Override the default task priority if provided") ] = None, from_task_run: Annotated[ TaskRun | None, Doc( "The task run queueing this one, if any. " "Prefer [TaskRun.queue][fluid.scheduler.TaskRun.queue], " "which passes it for you." ), ] = None, **params: Annotated[ Any, Doc( "The optional parameters for the task run. " "They must match the task params model" ), ], ) -> TaskRun: """Queue a task for execution This methods fires two events: - `init`: when the task run is created - `queued`: after the task is queued It returns the [TaskRun][fluid.scheduler.TaskRun] object """ task_run = self.create_task_run( task, run_id=run_id, priority=priority, from_task_run=from_task_run, **params, ) return await self._queue_task_run(task_run) ``` ### create_task_run ```python create_task_run( task, *, run_id="", priority=None, from_task_run=None, **params ) ``` Create a TaskRun in `init` state | PARAMETER | DESCRIPTION | | --------------- | ----------------------------------------------------------------------------------------------------------------- | | `task` | The task or task name, if a task name it must be registered with the task manager. **TYPE:** \`str | | `run_id` | Unique ID for the task run. If not provided a new UUID is generated. **TYPE:** `str` **DEFAULT:** `''` | | `priority` | Override the default task priority if provided **TYPE:** \`TaskPriority | | `from_task_run` | The task run creating this one, if any. It records the chain. **TYPE:** \`TaskRun | | `**params` | The optional parameters for the task run. They must match the task params model **TYPE:** `Any` **DEFAULT:** `{}` | Source code in `fluid/scheduler/consumer.py` ```python def create_task_run( self, task: Annotated[ str | Task, Doc( "The task or task name," " if a task name it must be registered with the task manager." ), ], *, run_id: Annotated[ str, Doc("Unique ID for the task run. If not provided a new UUID is generated."), ] = "", priority: Annotated[ TaskPriority | None, Doc("Override the default task priority if provided") ] = None, from_task_run: Annotated[ TaskRun | None, Doc("The task run creating this one, if any. It records the chain."), ] = None, **params: Annotated[ Any, Doc( "The optional parameters for the task run. " "They must match the task params model" ), ], ) -> TaskRun: """Create a [TaskRun][fluid.scheduler.TaskRun] in `init` state""" task = self.broker.task_from_registry(task) run_id = run_id or self.broker.new_uuid() return TaskRun( id=run_id, task=task, priority=priority or task.priority, params=task.params_model(**params), task_manager=self, from_run_id=from_task_run.id if from_task_run else "", # the first run in a chain has no root of its own, it is the root root_run_id=( (from_task_run.root_run_id or from_task_run.id) if from_task_run else "" ), ) ``` ### register_from_module ```python register_from_module(module, tags=None) ``` Register tasks from a python module | PARAMETER | DESCRIPTION | | --------- | ------------------------------------------------------------------------------------------------------------------------------- | | `module` | Python module with tasks implementations - can contain any object, only instances of Task are registered **TYPE:** `ModuleType` | | `tags` | Extra tags to add to every registered task **TYPE:** \`Sequence[str] | Source code in `fluid/scheduler/consumer.py` ```python def register_from_module( self, module: Annotated[ ModuleType, Doc( "Python module with tasks implementations " "- can contain any object, only instances of Task are registered" ), ], tags: Annotated[ Sequence[str] | None, Doc("Extra tags to add to every registered task"), ] = None, ) -> None: """Register tasks from a python module""" for name in dir(module): if name.startswith("_"): continue if isinstance(obj := getattr(module, name), Task): self.register_task(obj, tags=tags) ``` ### register_from_dict ```python register_from_dict(data, tags=None) ``` Register tasks from a python dictionary | PARAMETER | DESCRIPTION | | --------- | --------------------------------------------------------------------------------------------------------------------------------------- | | `data` | Python dictionary with tasks implementations - can contain any object, only instances of Task are registered **TYPE:** `dict[str, Any]` | | `tags` | Extra tags to add to every registered task **TYPE:** \`Sequence[str] | Source code in `fluid/scheduler/consumer.py` ```python def register_from_dict( self, data: Annotated[ dict[str, Any], Doc( "Python dictionary with tasks implementations " "- can contain any object, only instances of Task are registered" ), ], tags: Annotated[ Sequence[str] | None, Doc("Extra tags to add to every registered task"), ] = None, ) -> None: """Register tasks from a python dictionary""" for name, obj in data.items(): if name.startswith("_"): continue if isinstance(obj, Task): self.register_task(obj, tags=tags) ``` ### with_plugin ```python with_plugin(plugin) ``` Register a plugin with the task manager | PARAMETER | DESCRIPTION | | --------- | ---------------------------------------------------- | | `plugin` | The plugin to register **TYPE:** `TaskManagerPlugin` | Source code in `fluid/scheduler/consumer.py` ```python def with_plugin( self, plugin: Annotated[TaskManagerPlugin, Doc("The plugin to register")], ) -> Self: """Register a plugin with the task manager""" self._plugins.append(plugin) plugin.register(self) return self ``` ### startup ```python startup() ``` Start the task consumer workers. A cpu bound process executes a single task and exits, it never consumes the queue. Reaching this point means the entry point ignored the `exec` command, so it cannot run cpu bound tasks. Source code in `fluid/scheduler/consumer.py` ```python async def startup(self) -> None: """Start the task consumer workers. A cpu bound process executes a single task and exits, it never consumes the queue. Reaching this point means the entry point ignored the `exec` command, so it cannot run cpu bound tasks. """ if is_in_cpu_process(): raise CpuBoundEntryPointError( "a task consumer cannot start in a cpu bound process: " "running cpu bound tasks requires the application entry point " "to be a TaskManagerCLI" ) await super().startup() ``` ### sync_queue ```python sync_queue(task, delay=0) ``` Queue a task synchronously Source code in `fluid/scheduler/consumer.py` ```python def sync_queue(self, task: str | Task | TaskRun, delay: float = 0) -> None: """Queue a task synchronously""" self._in_process_queue.queue(task, delay=delay) ``` ### queue_and_wait ```python queue_and_wait(task, *, timeout=None, **params) ``` Queue a task and wait for it to finish | PARAMETER | DESCRIPTION | | ---------- | ----------------------------------------------------------------------------------------------------------------- | | `task` | The task or task name, if a task name it must be registered with the task manager. **TYPE:** \`str | | `timeout` | Timeout for waiting the task to finish **TYPE:** \`int | | `**params` | The optional parameters for the task run. They must match the task params model **TYPE:** `Any` **DEFAULT:** `{}` | Source code in `fluid/scheduler/consumer.py` ```python async def queue_and_wait( self, task: Annotated[ str | Task, Doc( "The task or task name," " if a task name it must be registered with the task manager." ), ], *, timeout: Annotated[ int | None, Doc("Timeout for waiting the task to finish") ] = None, **params: Annotated[ Any, Doc( "The optional parameters for the task run. " "They must match the task params model" ), ], ) -> TaskRun: """Queue a task and wait for it to finish""" with TaskRunWaiter(self) as waiter: task_run = await self.queue(task, **params) return await waiter.wait(task_run, timeout=timeout) ``` ### register_async_handler ```python register_async_handler(event, handler) ``` Source code in `fluid/scheduler/consumer.py` ```python def register_async_handler(self, event: Event | str, handler: AsyncHandler) -> None: event = Event.from_string_or_event(event) self.dispatcher.register_handler( f"{event.type}.async_dispatch", self._async_dispatcher_worker.send, ) self._async_dispatcher_worker.dispatcher.register_handler(event, handler) ``` ### unregister_async_handler ```python unregister_async_handler(event) ``` Source code in `fluid/scheduler/consumer.py` ```python def unregister_async_handler(self, event: Event | str) -> AsyncHandler | None: return self._async_dispatcher_worker.dispatcher.unregister_handler(event) ``` # Task Manager The Task Manager is a component that manages the execution of tasks. It is the simplest way to run tasks and it is the base class for the TaskConsumer and the TaskScheduler. It can be imported from `fluid.scheduler`: ```python from fluid.scheduler import TaskManager ``` The Task Manager is useful if you want to execute tasks in a synchronous or asynchronous way. ## fluid.scheduler.TaskManager ```python TaskManager(*, deps=None, config=None, **kwargs) ``` The task manager is the main class for managing tasks | PARAMETER | DESCRIPTION | | ---------- | ----------------------------------------------------------------------------------------------------------------------------- | | `deps` | Application dependencies available to every task run. See the Task Dependencies tutorial. **TYPE:** `Any` **DEFAULT:** `None` | | `config` | Task manager configuration. Built from the extra keyword arguments when not provided. **TYPE:** \`TaskManagerConfig | | `**kwargs` | Configuration fields, used when config is not provided. **TYPE:** `Any` **DEFAULT:** `{}` | Source code in `fluid/scheduler/consumer.py` ```python def __init__( self, *, deps: Annotated[ Any, Doc(""" Application dependencies available to every task run. See the [Task Dependencies](../tutorials/task_deps.md) tutorial. """), ] = None, config: Annotated[ TaskManagerConfig | None, Doc(""" Task manager configuration. Built from the extra keyword arguments when not provided. """), ] = None, **kwargs: Annotated[ Any, Doc("Configuration fields, used when `config` is not provided."), ], ) -> None: self.deps: Annotated[ Any, Doc(""" Dependencies for the task manager. Production applications requires global dependencies to be available to all tasks. This can be achieved by setting the `deps` attribute of the task manager to an object with the required dependencies. Each task can cast the dependencies to the required type. """), ] = ( deps if deps is not None else State() ) self.state: Annotated[ State, Doc(""" State for the task manager. This can be used by plugins to store state in the task manager. """), ] = State() self.config: Annotated[ TaskManagerConfig, Doc("""Task manager configuration""") ] = config or TaskManagerConfig(**kwargs) self.dispatcher: Annotated[ TaskDispatcher, Doc(""" A dispatcher of [TaskRun][fluid.scheduler.TaskRun] events. Application can register handlers to listen for events happening during the lifecycle of a task run. """), ] = TaskDispatcher() self.broker = TaskBroker.from_url(self.config.broker_url) self.manager_id: str = self.broker.new_uuid() self._plugins: list[TaskManagerPlugin] = [] self._async_contexts: list[Any] = [] self._stack = AsyncExitStack() ``` ### deps ```python deps = deps if deps is not None else State() ``` Dependencies for the task manager. Production applications requires global dependencies to be available to all tasks. This can be achieved by setting the `deps` attribute of the task manager to an object with the required dependencies. Each task can cast the dependencies to the required type. ### state ```python state = State() ``` State for the task manager. This can be used by plugins to store state in the task manager. ### config ```python config = config or TaskManagerConfig(**kwargs) ``` Task manager configuration ### dispatcher ```python dispatcher = TaskDispatcher() ``` A dispatcher of TaskRun events. Application can register handlers to listen for events happening during the lifecycle of a task run. ### broker ```python broker = TaskBroker.from_url(self.config.broker_url) ``` ### manager_id ```python manager_id = self.broker.new_uuid() ``` ### registry ```python registry ``` The task registry ### type ```python type ``` The type of the task manager ### on_startup ```python on_startup() ``` Source code in `fluid/scheduler/consumer.py` ```python async def on_startup(self) -> None: await self.__aenter__() ``` ### on_shutdown ```python on_shutdown() ``` Source code in `fluid/scheduler/consumer.py` ```python async def on_shutdown(self) -> None: await self.__aexit__(None, None, None) ``` ### add_async_context_manager ```python add_async_context_manager(cm) ``` Add an async context manager to the task manager These context managers are entered when the task manager starts Source code in `fluid/scheduler/consumer.py` ```python def add_async_context_manager(self, cm: Any) -> None: """Add an async context manager to the task manager These context managers are entered when the task manager starts """ self._async_contexts.append(cm) ``` ### register_task ```python register_task(task, tags=None) ``` Register a task with the task manager | PARAMETER | DESCRIPTION | | --------- | ----------------------------------------------------------------------------- | | `task` | Task to register **TYPE:** `Task` | | `tags` | Extra tags to add to the task before registering it **TYPE:** \`Sequence[str] | Source code in `fluid/scheduler/consumer.py` ```python def register_task( self, task: Annotated[Task, Doc("Task to register")], tags: Annotated[ Sequence[str] | None, Doc("Extra tags to add to the task before registering it"), ] = None, ) -> None: """Register a task with the task manager""" if tags: task = task._replace(tags=task.tags | frozenset(tags)) self.broker.register_task(task) ``` ### execute ```python execute(task, *, run_id='', priority=None, **params) ``` Execute a task and wait for it to finish This method is an async method that should be used in an asynchronous context when one need to wait for the task to finish execution. | PARAMETER | DESCRIPTION | | ---------- | ----------------------------------------------------------------------------------------------------------------- | | `task` | The task or task name, if a task name it must be registered with the task manager. **TYPE:** \`str | | `run_id` | Unique ID for the task run. If not provided a new UUID is generated. **TYPE:** `str` **DEFAULT:** `''` | | `priority` | Override the default task priority if provided **TYPE:** \`TaskPriority | | `**params` | The optional parameters for the task run. They must match the task params model **TYPE:** `Any` **DEFAULT:** `{}` | Source code in `fluid/scheduler/consumer.py` ```python async def execute( self, task: Annotated[ str | Task, Doc( "The task or task name," " if a task name it must be registered with the task manager." ), ], *, run_id: Annotated[ str, Doc("Unique ID for the task run. If not provided a new UUID is generated."), ] = "", priority: Annotated[ TaskPriority | None, Doc("Override the default task priority if provided") ] = None, **params: Annotated[ Any, Doc( "The optional parameters for the task run. " "They must match the task params model" ), ], ) -> TaskRun: """Execute a task and wait for it to finish This method is an async method that should be used in an asynchronous context when one need to wait for the task to finish execution. """ task_run = self.create_task_run( task, run_id=run_id, priority=priority, **params, ) try: await task_run._execute() except TaskAbortedError as exc: await self.broker.set_task_aborted(task_run.id, str(exc)) return task_run ``` ### execute_sync ```python execute_sync(task, *, run_id='', priority=None, **params) ``` Execute a task synchronously This method is a blocking method that should be used in a synchronous context. | PARAMETER | DESCRIPTION | | ---------- | ----------------------------------------------------------------------------------------------------------------- | | `task` | The task or task name, if a task name it must be registered with the task manager. **TYPE:** \`str | | `run_id` | Unique ID for the task run. If not provided a new UUID is generated. **TYPE:** `str` **DEFAULT:** `''` | | `priority` | Override the default task priority if provided **TYPE:** \`TaskPriority | | `**params` | The optional parameters for the task run. They must match the task params model **TYPE:** `Any` **DEFAULT:** `{}` | Source code in `fluid/scheduler/consumer.py` ```python def execute_sync( self, task: Annotated[ str | Task, Doc( "The task or task name," " if a task name it must be registered with the task manager." ), ], *, run_id: Annotated[ str, Doc("Unique ID for the task run. If not provided a new UUID is generated."), ] = "", priority: Annotated[ TaskPriority | None, Doc("Override the default task priority if provided") ] = None, **params: Annotated[ Any, Doc( "The optional parameters for the task run. " "They must match the task params model" ), ], ) -> TaskRun: """Execute a task synchronously This method is a blocking method that should be used in a synchronous context. """ return asyncio.run( self._execute_and_exit( task, run_id=run_id, priority=priority, **params, ) ) ``` ### queue ```python queue( task, *, run_id="", priority=None, from_task_run=None, **params ) ``` Queue a task for execution This methods fires two events: - `init`: when the task run is created - `queued`: after the task is queued It returns the TaskRun object | PARAMETER | DESCRIPTION | | --------------- | ----------------------------------------------------------------------------------------------------------------- | | `task` | The task or task name, if a task name it must be registered with the task manager. **TYPE:** \`str | | `run_id` | Unique ID for the task run. If not provided a new UUID is generated. **TYPE:** `str` **DEFAULT:** `''` | | `priority` | Override the default task priority if provided **TYPE:** \`TaskPriority | | `from_task_run` | The task run queueing this one, if any. Prefer TaskRun.queue, which passes it for you. **TYPE:** \`TaskRun | | `**params` | The optional parameters for the task run. They must match the task params model **TYPE:** `Any` **DEFAULT:** `{}` | Source code in `fluid/scheduler/consumer.py` ```python async def queue( self, task: Annotated[ str | Task, Doc( "The task or task name," " if a task name it must be registered with the task manager." ), ], *, run_id: Annotated[ str, Doc("Unique ID for the task run. If not provided a new UUID is generated."), ] = "", priority: Annotated[ TaskPriority | None, Doc("Override the default task priority if provided") ] = None, from_task_run: Annotated[ TaskRun | None, Doc( "The task run queueing this one, if any. " "Prefer [TaskRun.queue][fluid.scheduler.TaskRun.queue], " "which passes it for you." ), ] = None, **params: Annotated[ Any, Doc( "The optional parameters for the task run. " "They must match the task params model" ), ], ) -> TaskRun: """Queue a task for execution This methods fires two events: - `init`: when the task run is created - `queued`: after the task is queued It returns the [TaskRun][fluid.scheduler.TaskRun] object """ task_run = self.create_task_run( task, run_id=run_id, priority=priority, from_task_run=from_task_run, **params, ) return await self._queue_task_run(task_run) ``` ### create_task_run ```python create_task_run( task, *, run_id="", priority=None, from_task_run=None, **params ) ``` Create a TaskRun in `init` state | PARAMETER | DESCRIPTION | | --------------- | ----------------------------------------------------------------------------------------------------------------- | | `task` | The task or task name, if a task name it must be registered with the task manager. **TYPE:** \`str | | `run_id` | Unique ID for the task run. If not provided a new UUID is generated. **TYPE:** `str` **DEFAULT:** `''` | | `priority` | Override the default task priority if provided **TYPE:** \`TaskPriority | | `from_task_run` | The task run creating this one, if any. It records the chain. **TYPE:** \`TaskRun | | `**params` | The optional parameters for the task run. They must match the task params model **TYPE:** `Any` **DEFAULT:** `{}` | Source code in `fluid/scheduler/consumer.py` ```python def create_task_run( self, task: Annotated[ str | Task, Doc( "The task or task name," " if a task name it must be registered with the task manager." ), ], *, run_id: Annotated[ str, Doc("Unique ID for the task run. If not provided a new UUID is generated."), ] = "", priority: Annotated[ TaskPriority | None, Doc("Override the default task priority if provided") ] = None, from_task_run: Annotated[ TaskRun | None, Doc("The task run creating this one, if any. It records the chain."), ] = None, **params: Annotated[ Any, Doc( "The optional parameters for the task run. " "They must match the task params model" ), ], ) -> TaskRun: """Create a [TaskRun][fluid.scheduler.TaskRun] in `init` state""" task = self.broker.task_from_registry(task) run_id = run_id or self.broker.new_uuid() return TaskRun( id=run_id, task=task, priority=priority or task.priority, params=task.params_model(**params), task_manager=self, from_run_id=from_task_run.id if from_task_run else "", # the first run in a chain has no root of its own, it is the root root_run_id=( (from_task_run.root_run_id or from_task_run.id) if from_task_run else "" ), ) ``` ### register_from_module ```python register_from_module(module, tags=None) ``` Register tasks from a python module | PARAMETER | DESCRIPTION | | --------- | ------------------------------------------------------------------------------------------------------------------------------- | | `module` | Python module with tasks implementations - can contain any object, only instances of Task are registered **TYPE:** `ModuleType` | | `tags` | Extra tags to add to every registered task **TYPE:** \`Sequence[str] | Source code in `fluid/scheduler/consumer.py` ```python def register_from_module( self, module: Annotated[ ModuleType, Doc( "Python module with tasks implementations " "- can contain any object, only instances of Task are registered" ), ], tags: Annotated[ Sequence[str] | None, Doc("Extra tags to add to every registered task"), ] = None, ) -> None: """Register tasks from a python module""" for name in dir(module): if name.startswith("_"): continue if isinstance(obj := getattr(module, name), Task): self.register_task(obj, tags=tags) ``` ### register_from_dict ```python register_from_dict(data, tags=None) ``` Register tasks from a python dictionary | PARAMETER | DESCRIPTION | | --------- | --------------------------------------------------------------------------------------------------------------------------------------- | | `data` | Python dictionary with tasks implementations - can contain any object, only instances of Task are registered **TYPE:** `dict[str, Any]` | | `tags` | Extra tags to add to every registered task **TYPE:** \`Sequence[str] | Source code in `fluid/scheduler/consumer.py` ```python def register_from_dict( self, data: Annotated[ dict[str, Any], Doc( "Python dictionary with tasks implementations " "- can contain any object, only instances of Task are registered" ), ], tags: Annotated[ Sequence[str] | None, Doc("Extra tags to add to every registered task"), ] = None, ) -> None: """Register tasks from a python dictionary""" for name, obj in data.items(): if name.startswith("_"): continue if isinstance(obj, Task): self.register_task(obj, tags=tags) ``` ### register_async_handler ```python register_async_handler(event, handler) ``` Register an async handler for a given event This method is a no op for a TaskManager that is not a worker | PARAMETER | DESCRIPTION | | --------- | ------------------------------------------------------- | | `event` | The event to register the handler for **TYPE:** \`Event | Source code in `fluid/scheduler/consumer.py` ```python def register_async_handler( self, event: Annotated[Event | str, Doc("The event to register the handler for")], handler: AsyncHandler, ) -> None: """Register an async handler for a given event This method is a no op for a TaskManager that is not a worker """ ``` ### unregister_async_handler ```python unregister_async_handler(event) ``` Unregister an async handler for a given event This method is a no op for a TaskManager that is not a worker | PARAMETER | DESCRIPTION | | --------- | --------------------------------------------------------- | | `event` | The event to unregister the handler for **TYPE:** \`Event | Source code in `fluid/scheduler/consumer.py` ```python def unregister_async_handler( self, event: Annotated[Event | str, Doc("The event to unregister the handler for")], ) -> AsyncHandler | None: """Unregister an async handler for a given event This method is a no op for a TaskManager that is not a worker """ ``` ### with_plugin ```python with_plugin(plugin) ``` Register a plugin with the task manager | PARAMETER | DESCRIPTION | | --------- | ---------------------------------------------------- | | `plugin` | The plugin to register **TYPE:** `TaskManagerPlugin` | Source code in `fluid/scheduler/consumer.py` ```python def with_plugin( self, plugin: Annotated[TaskManagerPlugin, Doc("The plugin to register")], ) -> Self: """Register a plugin with the task manager""" self._plugins.append(plugin) plugin.register(self) return self ``` ## fluid.scheduler.TaskManagerConfig Bases: `BaseModel` Task manager configuration Fields: - `schedule_tasks` (`bool`) - `consume_tasks` (`bool`) - `max_concurrent_tasks` (`int`) - `sleep_millis` (`int`) - `broker_url` (`str`) ### schedule_tasks ```python schedule_tasks ``` Schedule tasks or sleep ### consume_tasks ```python consume_tasks = True ``` Consume tasks or sleep ### max_concurrent_tasks ```python max_concurrent_tasks ``` The number of coroutine workers consuming tasks. Each worker consumes one task at a time, therefore, this number is the maximum number of tasks that can run concurrently.It can be configured via the `FLUID_MAX_CONCURRENT_TASKS` environment variable, and by default is set to 5. ### sleep_millis ```python sleep_millis ``` Milliseconds to async sleep when no tasks available to consume.This value can be configured via the `FLUID_SLEEP_MILLIS` environment variable, and by default is set to 1000 milliseconds (1 second). ### broker_url ```python broker_url = '' ``` ### sleep ```python sleep ``` Sleep time in seconds ## fluid.scheduler.consumer.TaskDispatcher ```python TaskDispatcher() ``` Bases: `Dispatcher[TaskRun]` The task dispatcher is responsible for dispatching task run messages Source code in `fluid/utils/dispatcher.py` ```python def __init__(self) -> None: self._msg_handlers: defaultdict[str, dict[str, MessageHandlerType]] = ( defaultdict( dict, ) ) ``` ### event_type ```python event_type(message) ``` The event type is determined by the state of the task run Source code in `fluid/scheduler/consumer.py` ```python def event_type(self, message: TaskRun) -> str: """The event type is determined by the state of the task run""" return message.state ``` ### register_handler ```python register_handler(event, handler) ``` Register a handler for the given event It is possible to register multiple handlers for the same event type by providing a different tag for each handler. For example, to register two handlers for the event type `foo`: ```python dispatcher.register_handler("foo.first", handler1) dispatcher.register_handler("foo.second", handler2) ``` | PARAMETER | DESCRIPTION | | --------- | ------------------------------------------------------- | | `event` | The event to register the handler for **TYPE:** \`Event | | `handler` | The handler to register **TYPE:** `MessageHandlerType` | Source code in `fluid/utils/dispatcher.py` ````python def register_handler( self, event: Annotated[Event | str, Doc("The event to register the handler for")], handler: Annotated[MessageHandlerType, Doc("The handler to register")], ) -> MessageHandlerType | None: """Register a handler for the given event It is possible to register multiple handlers for the same event type by providing a different tag for each handler. For example, to register two handlers for the event type `foo`: ```python dispatcher.register_handler("foo.first", handler1) dispatcher.register_handler("foo.second", handler2) ``` """ event = Event.from_string_or_event(event) previous = self._msg_handlers[event.type].get(event.tag) self._msg_handlers[event.type][event.tag] = handler return previous ```` ### unregister_handler ```python unregister_handler(event) ``` Unregister a handler for the given event It returns the handler that was unregistered or `None` if no handler was registered for the given event. | PARAMETER | DESCRIPTION | | --------- | ----------------------------------------------------- | | `event` | The event to unregister the handler **TYPE:** \`Event | Source code in `fluid/utils/dispatcher.py` ```python def unregister_handler( self, event: Annotated[Event | str, Doc("The event to unregister the handler")] ) -> MessageHandlerType | None: """Unregister a handler for the given event It returns the handler that was unregistered or `None` if no handler was registered for the given event. """ event = Event.from_string_or_event(event) return self._msg_handlers[event.type].pop(event.tag, None) ``` ### get_handlers ```python get_handlers(message) ``` Get all event handlers for the given message This method returns a dictionary of all handlers registered for the given message type. If no handlers are registered for the message type, it returns `None`. | PARAMETER | DESCRIPTION | | --------- | ----------------------------------------------------------- | | `message` | The message to get the handlers for **TYPE:** `MessageType` | Source code in `fluid/utils/dispatcher.py` ```python def get_handlers( self, message: Annotated[MessageType, Doc("The message to get the handlers for")], ) -> dict[str, MessageHandlerType] | None: """Get all event handlers for the given message This method returns a dictionary of all handlers registered for the given message type. If no handlers are registered for the message type, it returns `None`. """ event_type = self.event_type(message) return self._msg_handlers.get(event_type) ``` ### dispatch ```python dispatch(message) ``` dispatch the message to all handlers It returns the number of handlers that were called Source code in `fluid/utils/dispatcher.py` ```python def dispatch(self, message: MessageType) -> int: """dispatch the message to all handlers It returns the number of handlers that were called """ handlers = self.get_handlers(message) if handlers: for handler in handlers.values(): handler(message) return len(handlers or ()) ``` ## fluid.scheduler.task_manager_fastapi ```python task_manager_fastapi( task_manager, *, app=None, include_router=True, prefix="/tasks", tags=None, **kwargs ) ``` Setup the FastAPI app and add the task manager to the state If the task manager is a Worker, it is also added to the app workers to be started with the app. | PARAMETER | DESCRIPTION | | ---------------- | --------------------------------------------------------------------------------------------------- | | `task_manager` | A TaskManager, TaskConsumer or TaskScheduler instance **TYPE:** `TaskManager` | | `app` | FastAPI app instance. If not provided, a new instance is created. **TYPE:** \`FastAPI | | `include_router` | Whether to include the task manager router in the FastAPI app. **TYPE:** `bool` **DEFAULT:** `True` | | `prefix` | Prefix for the task manager routes. **TYPE:** `str` **DEFAULT:** `'/tasks'` | | `tags` | Tags for the task manager routes. **TYPE:** \`Sequence\[str | | `**kwargs` | Additional keyword arguments for the FastAPI app if not provided **TYPE:** `Any` **DEFAULT:** `{}` | Source code in `fluid/scheduler/endpoints.py` ```python def task_manager_fastapi( task_manager: Annotated[ TaskManager, Doc( ( "A [TaskManager][fluid.scheduler.TaskManager], " "[TaskConsumer][fluid.scheduler.TaskConsumer] or " "[TaskScheduler][fluid.scheduler.TaskScheduler] instance" ) ), ], *, app: Annotated[ FastAPI | None, Doc("FastAPI app instance. If not provided, a new instance is created."), ] = None, include_router: Annotated[ bool, Doc("Whether to include the task manager router in the FastAPI app."), ] = True, prefix: Annotated[ str, Doc("Prefix for the task manager routes."), ] = "/tasks", tags: Annotated[ Sequence[str | Enum] | None, Doc("Tags for the task manager routes."), ] = None, **kwargs: Annotated[ Any, Doc("Additional keyword arguments for the FastAPI app if not provided"), ], ) -> FastAPI: """Setup the FastAPI app and add the task manager to the state If the task manager is a [Worker][fluid.utils.worker.Worker], it is also added to the app workers to be started with the app. """ app = app or FastAPI(**kwargs) if include_router: tags_ = tags if tags is not None else ["Tasks"] app.include_router(get_router(task_manager), prefix=prefix, tags=list(tags_)) for plugin in task_manager._plugins: plugin.register_routes(app, prefix=prefix, tags=list(tags_)) app.state.task_manager = task_manager if isinstance(task_manager, Worker): app_workers(app).add_workers(task_manager) else: app.router.on_startup.append(task_manager.on_startup) app.router.on_shutdown.append(task_manager.on_shutdown) return app ``` ## fluid.scheduler.endpoints.get_task_manager ```python get_task_manager(app) ``` Get the task manager added to the app state by task_manager_fastapi. Use this outside a request, where there is an app but no request to depend on, and TaskManagerDep inside a route. Source code in `fluid/scheduler/endpoints.py` ```python def get_task_manager(app: FastAPI) -> TaskManager: """Get the task manager added to the app state by [task_manager_fastapi][fluid.scheduler.task_manager_fastapi]. Use this outside a request, where there is an app but no request to depend on, and [TaskManagerDep][fluid.scheduler.endpoints.TaskManagerDep] inside a route. """ return cast(TaskManager, app.state.task_manager) ``` ## fluid.scheduler.endpoints.get_task_manager_from_request ```python get_task_manager_from_request(request) ``` Get the task manager of the app serving the request. This is the callable behind TaskManagerDep. Source code in `fluid/scheduler/endpoints.py` ```python def get_task_manager_from_request(request: Request) -> TaskManager: """Get the task manager of the app serving the request. This is the callable behind [TaskManagerDep][fluid.scheduler.endpoints.TaskManagerDep]. """ return get_task_manager(request.app) ``` ## fluid.scheduler.endpoints.TaskManagerDep ```python TaskManagerDep = TaskManager ``` FastAPI dependency injecting the TaskManager into a route. Application routes use it to reach the task manager, and through it the dependencies and the resources shared with every task run. # Task Manager Plugins Plugins extend the TaskManager with additional behaviour by hooking into task lifecycle events. A plugin implements the TaskManagerPlugin interface and is registered via TaskManager.with_plugin. ```python from fluid.scheduler import TaskScheduler, task_manager_fastapi from fluid.scheduler.db import TaskDbPlugin task_manager = TaskScheduler(...) task_manager.with_plugin(TaskDbPlugin(db)) app = task_manager_fastapi(task_manager) ``` ## fluid.scheduler.TaskManagerPlugin Bases: `ABC` Plugin for a task Manager ### register ```python register(task_manager) ``` Register the plugin with the task manager Source code in `fluid/scheduler/plugin.py` ```python @abc.abstractmethod def register(self, task_manager: TaskManager) -> None: """Register the plugin with the task manager""" ``` ### register_routes ```python register_routes(app, prefix='/tasks', tags=None) ``` Register routes with the FastAPI app | PARAMETER | DESCRIPTION | | --------- | ---------------------------------------------------------------------- | | `app` | FastAPI app instance. **TYPE:** `FastAPI` | | `prefix` | The URL prefix for the routes. **TYPE:** `str` **DEFAULT:** `'/tasks'` | | `tags` | The tags for the routes. **TYPE:** \`list\[str | Source code in `fluid/scheduler/plugin.py` ```python def register_routes( # noqa: B027 self, app: Annotated[ FastAPI, Doc("FastAPI app instance."), ], prefix: Annotated[ str, Doc("The URL prefix for the routes."), ] = "/tasks", tags: Annotated[ list[str | Enum] | None, Doc("The tags for the routes."), ] = None, ) -> None: """Register routes with the FastAPI app""" ``` ## fluid.scheduler.db.TaskDbPlugin ```python TaskDbPlugin( db, *, table_name="fluid_tasks", tag="db", skip_db_tag="skip_db", route_prefix=None ) ``` Bases: `TaskManagerPlugin` A plugin to store TaskRun in a postgresql database. This plugin listens to task state changes and updates the database accordingly. It requires a CrudDB instance to perform database operations and allows customization of the table name and event tags. You can use the `skip_db` tag to prevent database operations for specific tasks. It can be used if the `db` extra is installed, and requires a compatible database backend supported by CrudDB. | PARAMETER | DESCRIPTION | | -------------- | ------------------------------------------------------------------------------------------------------------------------------------------ | | `table_name` | The name of the table to store task runs **TYPE:** `str` **DEFAULT:** `'fluid_tasks'` | | `tag` | The tag for the plugin event registration **TYPE:** `str` **DEFAULT:** `'db'` | | `skip_db_tag` | The tag to skip database operations **TYPE:** `str` **DEFAULT:** `'skip_db'` | | `route_prefix` | Fix the URL prefix for the history routes. If None, routes are registered using the prefix parameter from register_routes. **TYPE:** \`str | Source code in `fluid/scheduler/db.py` ```python def __init__( self, db: CrudDB, *, table_name: Annotated[ str, Doc("The name of the table to store task runs"), ] = "fluid_tasks", tag: Annotated[ str, Doc("The tag for the plugin event registration"), ] = "db", skip_db_tag: Annotated[ str, Doc("The tag to skip database operations"), ] = "skip_db", route_prefix: Annotated[ str | None, Doc( "Fix the URL prefix for the history routes. If None, " "routes are registered using the prefix parameter from register_routes." ), ] = None, ) -> None: if table_name not in db.tables: task_meta(db.metadata, table_name=table_name) self.table_name = table_name self.db = db self.tag = tag self.skip_db_tag = skip_db_tag self.route_prefix = route_prefix ``` ### table_name ```python table_name = table_name ``` ### db ```python db = db ``` ### tag ```python tag = tag ``` ### skip_db_tag ```python skip_db_tag = skip_db_tag ``` ### route_prefix ```python route_prefix = route_prefix ``` ### register ```python register(task_manager) ``` Source code in `fluid/scheduler/db.py` ```python def register(self, task_manager: TaskManager) -> None: task_manager.state.task_db_plugin = self self.task_manager = task_manager if is_in_cpu_process(): return task_manager.register_async_handler( Event(TaskState.queued, self.tag), self._handle_update, ) task_manager.register_async_handler( Event(TaskState.running, self.tag), self._handle_update, ) task_manager.register_async_handler( Event(TaskState.success, self.tag), self._handle_update, ) task_manager.register_async_handler( Event(TaskState.failure, self.tag), self._handle_update, ) task_manager.register_async_handler( Event(TaskState.aborted, self.tag), self._handle_update, ) task_manager.register_async_handler( Event(TaskState.rate_limited, self.tag), self._handle_update, ) task_manager.register_async_handler( Event(TaskState.interrupted, self.tag), self._handle_update, ) ``` ### register_routes ```python register_routes(app, prefix='/tasks', tags=None) ``` Register routes with the FastAPI app | PARAMETER | DESCRIPTION | | --------- | ---------------------------------------------------------------------- | | `app` | FastAPI app instance. **TYPE:** `FastAPI` | | `prefix` | The URL prefix for the routes. **TYPE:** `str` **DEFAULT:** `'/tasks'` | | `tags` | The tags for the routes. **TYPE:** \`list\[str | Source code in `fluid/scheduler/db.py` ```python def register_routes( self, app: Annotated[ FastAPI, Doc("FastAPI app instance."), ], prefix: Annotated[ str, Doc("The URL prefix for the routes."), ] = "/tasks", tags: Annotated[ list[str | Enum] | None, Doc("The tags for the routes."), ] = None, ) -> None: """Register routes with the FastAPI app""" prefix = self.route_prefix or f"{prefix}-history" app.include_router(router, prefix=prefix, tags=tags) ``` ### get_history ```python get_history(q) ``` Get task run history based on the provided query parameters. | PARAMETER | DESCRIPTION | | --------- | --------------------------------------------------------------------------- | | `q` | Query parameters for fetching task run history **TYPE:** `TaskHistoryQuery` | Source code in `fluid/scheduler/db.py` ```python async def get_history( self, q: Annotated[ TaskHistoryQuery, Doc("Query parameters for fetching task run history") ], ) -> TaskRunHistoryPage: """Get task run history based on the provided query parameters.""" table = self.db.tables[self.table_name] filters = q.filters() if q.tags: wanted = set(q.tags) names = { name for name, task in self.task_manager.registry.items() if wanted & task.tags } # AND with an explicit task filter; empty set → IN () → no rows if "name" in filters: names &= {filters["name"]} filters["name"] = list(names) pagination = Pagination.create( "queued", filters=filters, limit=q.limit, cursor=q.cursor, desc=True, ) rows, cursor = await pagination.execute(self.db, table) return TaskRunHistoryPage( data=[_row_to_task_run(row) for row in rows], cursor=cursor, ) ``` ### get_run ```python get_run(run_id) ``` Get a specific task run by its ID. Source code in `fluid/scheduler/db.py` ```python async def get_run(self, run_id: str) -> TaskRunHistory: """Get a specific task run by its ID.""" table = self.db.tables[self.table_name] result = await self.db.db_select(table, {"id": run_id}) rows = result.fetchall() if not rows: raise NoResultFound(f"Task run with id {run_id} not found") return _row_to_task_run(rows[0]) ``` ## Accessing the plugin from a task get_db_plugin retrieves the registered TaskDbPlugin from the task manager state. It is designed as a FastAPI dependency for route handlers, but can also be called directly from within a task by passing `context.task_manager`: ```python from fluid.scheduler import TaskRun, task from fluid.scheduler.db import get_db_plugin, TaskHistoryQuery @task() async def report(context: TaskRun) -> None: db_plugin = get_db_plugin(context.task_manager) page = await db_plugin.get_history(TaskHistoryQuery(task="my-task", limit=10)) for run in page.data: print(run.id, run.state) ``` ## fluid.scheduler.db.get_db_plugin ```python get_db_plugin(task_manager) ``` Retrieve the registered TaskDbPlugin. Can be used as a FastAPI dependency in route handlers, or called directly from within a task by passing `context.task_manager`. Source code in `fluid/scheduler/db.py` ```python def get_db_plugin(task_manager: TaskManagerDep) -> TaskDbPlugin: """Retrieve the registered [TaskDbPlugin][fluid.scheduler.db.TaskDbPlugin]. Can be used as a FastAPI dependency in route handlers, or called directly from within a task by passing `context.task_manager`. """ return task_manager.state.task_db_plugin ``` ## History Models The following models are used when querying task run history via TaskDbPlugin.get_history or the HTTP endpoints. They can be imported from `fluid.scheduler.db`: ```python from fluid.scheduler.db import TaskHistoryQuery, TaskRunHistory, TaskRunHistoryPage ``` ### Filtering by tags The `tags` field of TaskHistoryQuery filters runs by the tags of their Task. A run matches when its task carries **at least one** of the supplied tags (OR semantics, the same as the `tags` query parameter on the task list endpoint). Tags are resolved against the live task registry at query time, so they always reflect each task's *current* tags rather than the tags it had when the run executed. When combined with the `task` filter, the two are applied together (AND): the run's task must match the name *and* carry one of the tags. Tags that match no registered task return an empty result. ## fluid.scheduler.db.TaskHistoryQuery Bases: `BaseModel` Query parameters for fetching task run history. Fields: - `task` (`str | None`) - `start` (`datetime | None`) - `end` (`datetime | None`) - `state` (`TaskState | None`) - `params` (`dict[str, Any] | str | None`) - `tags` (`list[str] | None`) - `limit` (`int | None`) - `cursor` (`str`) Validators: - `_parse_params_str` ### task ```python task = None ``` Filter by task name when provided ### start ```python start = None ``` Filter runs queued at or after this time when provided ### end ```python end = None ``` Filter runs queued at or before this time when provided ### state ```python state = None ``` Filter by task state when provided ### params ```python params = None ``` Filter by params using JSON containment when provided ### tags ```python tags = None ``` Filter runs whose task has at least one of these tags when provided. Tags are resolved against the live task registry, so they reflect each task's current tags. ### limit ```python limit = None ``` Maximum number of results to return when provided ### cursor ```python cursor = '' ``` Pagination cursor from a previous response when provided ### filters ```python filters() ``` Source code in `fluid/scheduler/db.py` ```python def filters(self) -> dict: return { self._filter_map.get(k, k): v for k, v in self.model_dump( exclude_none=True, exclude={"limit", "cursor", "tags"} ).items() } ``` ## fluid.scheduler.db.TaskRunHistory Bases: `BaseModel` A model representing the history of a task run, including its parameters and timing information. Fields: - `id` (`str`) - `task` (`str`) - `priority` (`TaskPriority`) - `state` (`TaskState`) - `queued` (`datetime`) - `start` (`datetime | None`) - `end` (`datetime | None`) - `params` (`dict[str, Any]`) ### id ```python id ``` The unique ID of the task run ### task ```python task ``` The name of the task ### priority ```python priority ``` The priority of the task ### state ```python state ``` The state of the task ### queued ```python queued ``` The time the task was queued ### start ```python start = None ``` The start time of the task ### end ```python end = None ``` The end time of the task ### params ```python params ``` The parameters of the task run ## fluid.scheduler.db.TaskRunHistoryPage Bases: `BaseModel` A paginated response containing a list of task run history records. Returned by TaskDbPlugin.get_history and the `GET /task-history` endpoint. Fields: - `data` (`list[TaskRunHistory]`) - `cursor` (`str`) ### data ```python data ``` The task run history records ### cursor ```python cursor ``` Pagination cursor to fetch the next page # Task Registry ## fluid.scheduler.broker.TaskRegistry Bases: `dict[str, Task[TP]]` A registry of tasks ### periodic ```python periodic() ``` Iterate over periodic tasks Source code in `fluid/scheduler/broker.py` ```python def periodic(self) -> Iterable[Task]: """Iterate over periodic tasks""" for task in self.values(): yield task ``` # Task Retry `aio-fluid` supports automatic retries for two distinct failure modes: execution failures and rate limiting. Both are configured per-task via RetryPolicy objects passed to the @task decorator. ```python from fluid.scheduler import RetryPolicy, task, TaskRun ``` ## RetryPolicy ## fluid.scheduler.models.RetryPolicy ```python RetryPolicy( max_attempts=None, wait=1.0, backoff=1.0, max_wait=60.0, exceptions=(), ) ``` Retry policy for task execution failures. ```python from fluid.scheduler import RetryPolicy, task @task(retry=RetryPolicy(max_attempts=3, wait=2.0, backoff=2.0)) async def my_task(ctx: TaskRun) -> None: ... ``` ### max_attempts ```python max_attempts = None ``` Maximum number of retry attempts, not counting the initial attempt. If None, there is no limit on the number of attempts. ### wait ```python wait = 1.0 ``` Base wait time in seconds before the first retry. ### backoff ```python backoff = 1.0 ``` Multiplier applied to `wait` on each successive attempt. Use `1.0` for fixed delay, `2.0` for exponential backoff. ### max_wait ```python max_wait = 60.0 ``` Upper bound on wait time in seconds regardless of backoff. ### exceptions ```python exceptions = () ``` Exception types that trigger a retry. Empty tuple matches all exceptions. ### delay ```python delay(attempt) ``` Compute wait time before the given attempt number (1-based). Source code in `fluid/scheduler/models.py` ```python def delay(self, attempt: int) -> float: """Compute wait time before the given attempt number (1-based).""" return min(self.wait * (self.backoff ** (attempt - 1)), self.max_wait) ``` ### matches ```python matches(exc) ``` Return True if this exception should trigger a retry. Source code in `fluid/scheduler/models.py` ```python def matches(self, exc: Exception) -> bool: """Return True if this exception should trigger a retry.""" if not self.exceptions: return True return isinstance(exc, self.exceptions) ``` ## Configuring retries on a task ### Failure retry Set `retry` on @task to re-queue the task when its executor raises an exception. The TaskRun is re-queued with an `execute_after` delay computed from the RetryPolicy; the worker that dequeued it is freed immediately to process other tasks. ```python from fluid.scheduler import RetryPolicy, task, TaskRun @task(retry=RetryPolicy(max_attempts=3, wait=2.0, backoff=2.0)) async def fetch(ctx: TaskRun) -> None: ... ``` With `backoff=2.0` the delays between attempts are `2s → 4s → 8s`. Use `backoff=1.0` (the default) for a fixed delay. To retry only on specific exception types, pass `exceptions`: ```python @task(retry=RetryPolicy(max_attempts=5, wait=1.0, exceptions=(IOError, TimeoutError))) async def fetch(ctx: TaskRun) -> None: ... ``` ### Rate-limit retry Set `rate_limit_retry` on @task to re-queue the task when it cannot start because max_concurrency is already reached. Without this policy, the TaskRun ends immediately in the rate_limited state. ```python @task( max_concurrency=1, rate_limit_retry=RetryPolicy(max_attempts=5, wait=10.0, backoff=1.5, max_wait=120.0), ) async def exclusive(ctx: TaskRun) -> None: ... ``` ### How re-queuing works Both retry modes share the same mechanism: 1. The TaskConsumer detects the failure (execution error or concurrency limit). 1. It creates a copy of the TaskRun with a fresh state and an `execute_after` timestamp set to `now + delay`. 1. The copy is pushed back onto the Redis queue via the TaskBroker. 1. The TaskConsumer is freed immediately — no sleeping. 1. When the copy is next dequeued, if `execute_after` is still in the future it is re-scheduled via `call_later` and the worker moves on; otherwise execution proceeds normally. Note The minimum effective re-queue delay is **5 seconds**, regardless of the `wait` value in the policy. A `call_later` is used to avoid busy-looping, and the floor ensures the worker is not called back too aggressively. # Task Run It can be imported from `fluid.scheduler`: ```python from fluid.scheduler import TaskRun ``` ## fluid.scheduler.TaskRun Bases: `BaseModel`, `Generic[TP, TD]` A TaskRun contains all the data generated by a Task run This model is never initialized directly, it is created by the TaskManager Fields: - `id` (`str`) - `task` (`Task`) - `priority` (`TaskPriority`) - `params` (`TP`) - `state` (`TaskState`) - `task_manager` (`TaskManager`) - `queued` (`datetime | None`) - `start` (`datetime | None`) - `end` (`datetime | None`) - `execute_after` (`datetime | None`) - `rate_limit_attempt` (`int`) - `retry_attempt` (`int`) - `from_run_id` (`str`) - `root_run_id` (`str`) ### id ```python id ``` Unique task run id ### task ```python task ``` Task to be executed ### priority ```python priority ``` Task priority ### params ```python params ``` Task parameters ### state ```python state = TaskState.init ``` Task state ### task_manager ```python task_manager ``` ### queued ```python queued = None ``` ### start ```python start = None ``` ### end ```python end = None ``` ### execute_after ```python execute_after = None ``` Do not execute before this UTC timestamp. Set by retry logic. ### rate_limit_attempt ```python rate_limit_attempt = 0 ``` Number of rate-limit retries already consumed. ### retry_attempt ```python retry_attempt = 0 ``` Number of failure retries already consumed. ### from_run_id ```python from_run_id = '' ``` ID of the task run that queued this one, empty when the run was not queued from another task run. ### root_run_id ```python root_run_id = '' ``` ID of the first task run in the chain, shared by every run queued from it, directly or indirectly. Empty for a run that starts a chain. ### logger ```python logger ``` ### in_queue ```python in_queue ``` ### duration ```python duration ``` ### duration_ms ```python duration_ms ``` ### total ```python total ``` ### name ```python name ``` ### name_id ```python name_id ``` ### is_done ```python is_done ``` ### is_failure ```python is_failure ``` ### deps ```python deps ``` Dependencies of the TaskManager running this task. Annotate the second type parameter to type them, for example `TaskRun[MyParams, MyDeps]`. It defaults to `Any`, therefore `TaskRun` and `TaskRun[MyParams]` keep working unchanged. ### abort ```python abort(reason='') ``` Abort the task run by raising TaskAbortedError. Source code in `fluid/scheduler/models.py` ```python def abort(self, reason: str = "") -> None: """Abort the task run by raising [TaskAbortedError][fluid.scheduler.errors.TaskAbortedError]. """ raise TaskAbortedError(reason) from None ``` ### set_state ```python set_state(state, state_time=None) ``` Set the state of the task run, with proper handling of timestamps and state transitions. This method is called by the task consumer and should not be called directly by the task executor. Source code in `fluid/scheduler/models.py` ```python def set_state( self, state: TaskState, state_time: datetime | None = None, ) -> None: """Set the state of the task run, with proper handling of timestamps and state transitions. This method is called by the task consumer and should not be called directly by the task executor. """ if self.state == state: return state_time = as_utc(state_time) match (self.state, state): case (TaskState.init, TaskState.queued): self.queued = state_time self.state = state self._dispatch() case (TaskState.init, _): self.set_state(TaskState.queued, state_time) self.set_state(state, state_time) case (TaskState.queued, TaskState.running): self.start = state_time self.state = state self._dispatch() case ( TaskState.queued, TaskState.success | TaskState.aborted | TaskState.rate_limited | TaskState.failure, ): self.set_state(TaskState.running, state_time) self.set_state(state, state_time) case ( TaskState.running, TaskState.success | TaskState.aborted | TaskState.rate_limited | TaskState.failure | TaskState.interrupted, ): self.end = state_time self.state = state self._dispatch() case _: raise TaskRunError(f"invalid state transition {self.state} -> {state}") ``` ### queue ```python queue(task, *, run_id='', priority=None, **params) ``` Queue another task from within this task run. The new run records this run in from_run_id and inherits its root_run_id, so a chain of tasks queued this way can be traced back to the run that started it. This returns as soon as the run is on the queue, it does not wait for it to execute. | PARAMETER | DESCRIPTION | | ---------- | ----------------------------------------------------------------------------------------------------------------- | | `task` | The task or task name, if a task name it must be registered with the task manager. **TYPE:** \`str | | `run_id` | Unique ID for the task run. If not provided a new UUID is generated. **TYPE:** `str` **DEFAULT:** `''` | | `priority` | Override the default task priority if provided **TYPE:** \`TaskPriority | | `**params` | The optional parameters for the task run. They must match the task params model **TYPE:** `Any` **DEFAULT:** `{}` | Source code in `fluid/scheduler/models.py` ```python async def queue( self, task: Annotated[ str | Task, Doc( "The task or task name," " if a task name it must be registered with the task manager." ), ], *, run_id: Annotated[ str, Doc("Unique ID for the task run. If not provided a new UUID is generated."), ] = "", priority: Annotated[ TaskPriority | None, Doc("Override the default task priority if provided") ] = None, **params: Annotated[ Any, Doc( "The optional parameters for the task run. " "They must match the task params model" ), ], ) -> TaskRun: """Queue another task from within this task run. The new run records this run in [from_run_id][fluid.scheduler.TaskRun.from_run_id] and inherits its [root_run_id][fluid.scheduler.TaskRun.root_run_id], so a chain of tasks queued this way can be traced back to the run that started it. This returns as soon as the run is on the queue, it does not wait for it to execute. """ return await self.task_manager.queue( task, run_id=run_id, priority=priority, from_task_run=self, **params, ) ``` ### lock ```python lock(timeout=None, name=None) ``` Get a lock for this task run Source code in `fluid/scheduler/models.py` ```python def lock(self, timeout: float | None = None, name: str | None = None) -> Lock: """Get a lock for this task run""" lock_name = f"tasks:{self.name}" if name: lock_name = f"{lock_name}:{name}" return self.task_manager.broker.lock(lock_name, timeout=timeout) ``` ### queue_dump_json ```python queue_dump_json() ``` Serialize the task run for the task queue Params are dumped with secret values revealed so they survive the round-trip through the queue - all other dumps keep secrets masked. Source code in `fluid/scheduler/models.py` ```python def queue_dump_json(self) -> bytes: """Serialize the task run for the task queue Params are dumped with secret values revealed so they survive the round-trip through the queue - all other dumps keep secrets masked. """ data = self.model_dump(mode="json", exclude={"params"}) data["params"] = params_dump(self.params) return to_json(data) ``` # Task Scheduler The task scheduler TaskScheduler inherits from the TaskConsumer to add scheduling of periodic tasks. It can be imported from `fluid.scheduler`: ```python from fluid.scheduler import TaskScheduler ``` ## fluid.scheduler.TaskScheduler ```python TaskScheduler( *, deps=None, config=None, name="", stopping_grace_period=None, **kwargs ) ``` Bases: `TaskConsumer` A task manager for scheduling tasks | PARAMETER | DESCRIPTION | | ----------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `deps` | Application dependencies available to every task run. See the Task Dependencies tutorial. **TYPE:** `Any` **DEFAULT:** `None` | | `config` | Task manager configuration. Built from the extra keyword arguments when not provided. **TYPE:** \`TaskManagerConfig | | `name` | Worker's name, if not provided it is evaluated from the class name **TYPE:** `str` **DEFAULT:** `''` | | `stopping_grace_period` | Grace period in seconds to wait for workers to stop running when this worker is shutdown. It defaults to the FLUID_STOPPING_GRACE_PERIOD environment variable or 10 seconds. **TYPE:** \`float | | `**kwargs` | Configuration fields, used when config is not provided. **TYPE:** `Any` **DEFAULT:** `{}` | Source code in `fluid/scheduler/scheduler.py` ```python def __init__( self, *, deps: Annotated[ Any, Doc(""" Application dependencies available to every task run. See the [Task Dependencies](../tutorials/task_deps.md) tutorial. """), ] = None, config: Annotated[ TaskManagerConfig | None, Doc(""" Task manager configuration. Built from the extra keyword arguments when not provided. """), ] = None, name: Annotated[ str, Doc("Worker's name, if not provided it is evaluated from the class name"), ] = "", stopping_grace_period: Annotated[ float | None, Doc( "Grace period in seconds to wait for workers to stop running " "when this worker is shutdown. " "It defaults to the `FLUID_STOPPING_GRACE_PERIOD` " "environment variable or 10 seconds." ), ] = None, **kwargs: Annotated[ Any, Doc("Configuration fields, used when `config` is not provided."), ], ) -> None: super().__init__( deps=deps, config=config, name=name, stopping_grace_period=stopping_grace_period, **kwargs, ) self.add_workers(ScheduleTasks(self)) ``` ### worker_state ```python worker_state ``` The running state of the worker ### worker_name ```python worker_name ``` The name of the worker ### num_workers ```python num_workers ``` ### deps ```python deps = deps if deps is not None else State() ``` Dependencies for the task manager. Production applications requires global dependencies to be available to all tasks. This can be achieved by setting the `deps` attribute of the task manager to an object with the required dependencies. Each task can cast the dependencies to the required type. ### state ```python state = State() ``` State for the task manager. This can be used by plugins to store state in the task manager. ### config ```python config = config or TaskManagerConfig(**kwargs) ``` Task manager configuration ### dispatcher ```python dispatcher = TaskDispatcher() ``` A dispatcher of TaskRun events. Application can register handlers to listen for events happening during the lifecycle of a task run. ### broker ```python broker = TaskBroker.from_url(self.config.broker_url) ``` ### manager_id ```python manager_id = self.broker.new_uuid() ``` ### registry ```python registry ``` The task registry ### type ```python type ``` The type of the task manager ### has_started ```python has_started() ``` Source code in `fluid/utils/worker.py` ```python def has_started(self) -> bool: return self._worker_state != WorkerState.INIT ``` ### is_running ```python is_running() ``` Source code in `fluid/utils/worker.py` ```python def is_running(self) -> bool: return self._worker_state == WorkerState.RUNNING ``` ### is_stopping ```python is_stopping() ``` Source code in `fluid/utils/worker.py` ```python def is_stopping(self) -> bool: return self._worker_state == WorkerState.STOPPING ``` ### is_stopped ```python is_stopped() ``` Source code in `fluid/utils/worker.py` ```python def is_stopped(self) -> bool: return self._worker_state in (WorkerState.STOPPED, WorkerState.FORCE_STOPPED) ``` ### gracefully_stop ```python gracefully_stop() ``` Try to gracefully stop the workers and this worker Source code in `fluid/utils/worker.py` ```python def gracefully_stop(self) -> None: """Try to gracefully stop the workers and this worker""" super().gracefully_stop() for worker in self._workers: worker.gracefully_stop() ``` ### after_shutdown ```python after_shutdown(reason, code) ``` Called after shutdown of worker By default it does nothing, but can be overriden to do something such as exit the process. Source code in `fluid/utils/worker.py` ```python def after_shutdown(self, reason: str, code: int) -> None: # noqa: B027 """Called after shutdown of worker By default it does nothing, but can be overriden to do something such as exit the process. """ ``` ### status ```python status() ``` Source code in `fluid/utils/worker.py` ```python async def status(self) -> dict: status_workers = await asyncio.gather( *[worker.status() for worker in self._workers], ) return { worker.worker_name: status for worker, status in zip(self._workers, status_workers, strict=False) } ``` ### on_startup ```python on_startup() ``` Source code in `fluid/scheduler/consumer.py` ```python async def on_startup(self) -> None: await self.__aenter__() ``` ### on_shutdown ```python on_shutdown() ``` Source code in `fluid/scheduler/consumer.py` ```python async def on_shutdown(self) -> None: await self.__aexit__(None, None, None) ``` ### startup ```python startup() ``` Start the task consumer workers. A cpu bound process executes a single task and exits, it never consumes the queue. Reaching this point means the entry point ignored the `exec` command, so it cannot run cpu bound tasks. Source code in `fluid/scheduler/consumer.py` ```python async def startup(self) -> None: """Start the task consumer workers. A cpu bound process executes a single task and exits, it never consumes the queue. Reaching this point means the entry point ignored the `exec` command, so it cannot run cpu bound tasks. """ if is_in_cpu_process(): raise CpuBoundEntryPointError( "a task consumer cannot start in a cpu bound process: " "running cpu bound tasks requires the application entry point " "to be a TaskManagerCLI" ) await super().startup() ``` ### shutdown ```python shutdown() ``` Shutdown a running worker and wait for it to stop This method will try to gracefully stop the worker and wait for it to stop. If the worker does not stop in the grace period, it will force shutdown by cancelling the task. Source code in `fluid/utils/worker.py` ```python async def shutdown(self) -> None: """Shutdown a running worker and wait for it to stop This method will try to gracefully stop the worker and wait for it to stop. If the worker does not stop in the grace period, it will force shutdown by cancelling the task. """ if self._worker_task_runner is not None: await self._worker_task_runner.shutdown() ``` ### wait_for_shutdown ```python wait_for_shutdown() ``` Wait for the worker to stop This method will wait for the worker to stop running, but doesn't try to gracefully stop it nor force shutdown. Source code in `fluid/utils/worker.py` ```python async def wait_for_shutdown(self) -> None: """Wait for the worker to stop This method will wait for the worker to stop running, but doesn't try to gracefully stop it nor force shutdown. """ if self._worker_task_runner is not None: await self._worker_task_runner.wait_for_shutdown() ``` ### workers ```python workers() ``` Source code in `fluid/utils/worker.py` ```python def workers(self) -> Iterator[Worker]: return iter(self._workers) ``` ### run ```python run() ``` Source code in `fluid/utils/worker.py` ```python async def run(self) -> None: while self.is_running(): for worker in self._workers: if not worker.has_started(): await worker.startup() if not worker.is_running(): self.gracefully_stop() break await asyncio.sleep(self._heartbeat) await self._wait_for_workers() ``` ### add_workers ```python add_workers(*workers) ``` add workers to the workers They can be added while the worker is running. Source code in `fluid/utils/worker.py` ```python def add_workers(self, *workers: Worker) -> None: """add workers to the workers They can be added while the worker is running. """ for worker in workers: if worker not in self._workers: self._workers.append(worker) ``` ### add_async_context_manager ```python add_async_context_manager(cm) ``` Add an async context manager to the task manager These context managers are entered when the task manager starts Source code in `fluid/scheduler/consumer.py` ```python def add_async_context_manager(self, cm: Any) -> None: """Add an async context manager to the task manager These context managers are entered when the task manager starts """ self._async_contexts.append(cm) ``` ### register_task ```python register_task(task, tags=None) ``` Register a task with the task manager | PARAMETER | DESCRIPTION | | --------- | ----------------------------------------------------------------------------- | | `task` | Task to register **TYPE:** `Task` | | `tags` | Extra tags to add to the task before registering it **TYPE:** \`Sequence[str] | Source code in `fluid/scheduler/consumer.py` ```python def register_task( self, task: Annotated[Task, Doc("Task to register")], tags: Annotated[ Sequence[str] | None, Doc("Extra tags to add to the task before registering it"), ] = None, ) -> None: """Register a task with the task manager""" if tags: task = task._replace(tags=task.tags | frozenset(tags)) self.broker.register_task(task) ``` ### execute ```python execute(task, *, run_id='', priority=None, **params) ``` Execute a task and wait for it to finish This method is an async method that should be used in an asynchronous context when one need to wait for the task to finish execution. | PARAMETER | DESCRIPTION | | ---------- | ----------------------------------------------------------------------------------------------------------------- | | `task` | The task or task name, if a task name it must be registered with the task manager. **TYPE:** \`str | | `run_id` | Unique ID for the task run. If not provided a new UUID is generated. **TYPE:** `str` **DEFAULT:** `''` | | `priority` | Override the default task priority if provided **TYPE:** \`TaskPriority | | `**params` | The optional parameters for the task run. They must match the task params model **TYPE:** `Any` **DEFAULT:** `{}` | Source code in `fluid/scheduler/consumer.py` ```python async def execute( self, task: Annotated[ str | Task, Doc( "The task or task name," " if a task name it must be registered with the task manager." ), ], *, run_id: Annotated[ str, Doc("Unique ID for the task run. If not provided a new UUID is generated."), ] = "", priority: Annotated[ TaskPriority | None, Doc("Override the default task priority if provided") ] = None, **params: Annotated[ Any, Doc( "The optional parameters for the task run. " "They must match the task params model" ), ], ) -> TaskRun: """Execute a task and wait for it to finish This method is an async method that should be used in an asynchronous context when one need to wait for the task to finish execution. """ task_run = self.create_task_run( task, run_id=run_id, priority=priority, **params, ) try: await task_run._execute() except TaskAbortedError as exc: await self.broker.set_task_aborted(task_run.id, str(exc)) return task_run ``` ### execute_sync ```python execute_sync(task, *, run_id='', priority=None, **params) ``` Execute a task synchronously This method is a blocking method that should be used in a synchronous context. | PARAMETER | DESCRIPTION | | ---------- | ----------------------------------------------------------------------------------------------------------------- | | `task` | The task or task name, if a task name it must be registered with the task manager. **TYPE:** \`str | | `run_id` | Unique ID for the task run. If not provided a new UUID is generated. **TYPE:** `str` **DEFAULT:** `''` | | `priority` | Override the default task priority if provided **TYPE:** \`TaskPriority | | `**params` | The optional parameters for the task run. They must match the task params model **TYPE:** `Any` **DEFAULT:** `{}` | Source code in `fluid/scheduler/consumer.py` ```python def execute_sync( self, task: Annotated[ str | Task, Doc( "The task or task name," " if a task name it must be registered with the task manager." ), ], *, run_id: Annotated[ str, Doc("Unique ID for the task run. If not provided a new UUID is generated."), ] = "", priority: Annotated[ TaskPriority | None, Doc("Override the default task priority if provided") ] = None, **params: Annotated[ Any, Doc( "The optional parameters for the task run. " "They must match the task params model" ), ], ) -> TaskRun: """Execute a task synchronously This method is a blocking method that should be used in a synchronous context. """ return asyncio.run( self._execute_and_exit( task, run_id=run_id, priority=priority, **params, ) ) ``` ### queue ```python queue( task, *, run_id="", priority=None, from_task_run=None, **params ) ``` Queue a task for execution This methods fires two events: - `init`: when the task run is created - `queued`: after the task is queued It returns the TaskRun object | PARAMETER | DESCRIPTION | | --------------- | ----------------------------------------------------------------------------------------------------------------- | | `task` | The task or task name, if a task name it must be registered with the task manager. **TYPE:** \`str | | `run_id` | Unique ID for the task run. If not provided a new UUID is generated. **TYPE:** `str` **DEFAULT:** `''` | | `priority` | Override the default task priority if provided **TYPE:** \`TaskPriority | | `from_task_run` | The task run queueing this one, if any. Prefer TaskRun.queue, which passes it for you. **TYPE:** \`TaskRun | | `**params` | The optional parameters for the task run. They must match the task params model **TYPE:** `Any` **DEFAULT:** `{}` | Source code in `fluid/scheduler/consumer.py` ```python async def queue( self, task: Annotated[ str | Task, Doc( "The task or task name," " if a task name it must be registered with the task manager." ), ], *, run_id: Annotated[ str, Doc("Unique ID for the task run. If not provided a new UUID is generated."), ] = "", priority: Annotated[ TaskPriority | None, Doc("Override the default task priority if provided") ] = None, from_task_run: Annotated[ TaskRun | None, Doc( "The task run queueing this one, if any. " "Prefer [TaskRun.queue][fluid.scheduler.TaskRun.queue], " "which passes it for you." ), ] = None, **params: Annotated[ Any, Doc( "The optional parameters for the task run. " "They must match the task params model" ), ], ) -> TaskRun: """Queue a task for execution This methods fires two events: - `init`: when the task run is created - `queued`: after the task is queued It returns the [TaskRun][fluid.scheduler.TaskRun] object """ task_run = self.create_task_run( task, run_id=run_id, priority=priority, from_task_run=from_task_run, **params, ) return await self._queue_task_run(task_run) ``` ### create_task_run ```python create_task_run( task, *, run_id="", priority=None, from_task_run=None, **params ) ``` Create a TaskRun in `init` state | PARAMETER | DESCRIPTION | | --------------- | ----------------------------------------------------------------------------------------------------------------- | | `task` | The task or task name, if a task name it must be registered with the task manager. **TYPE:** \`str | | `run_id` | Unique ID for the task run. If not provided a new UUID is generated. **TYPE:** `str` **DEFAULT:** `''` | | `priority` | Override the default task priority if provided **TYPE:** \`TaskPriority | | `from_task_run` | The task run creating this one, if any. It records the chain. **TYPE:** \`TaskRun | | `**params` | The optional parameters for the task run. They must match the task params model **TYPE:** `Any` **DEFAULT:** `{}` | Source code in `fluid/scheduler/consumer.py` ```python def create_task_run( self, task: Annotated[ str | Task, Doc( "The task or task name," " if a task name it must be registered with the task manager." ), ], *, run_id: Annotated[ str, Doc("Unique ID for the task run. If not provided a new UUID is generated."), ] = "", priority: Annotated[ TaskPriority | None, Doc("Override the default task priority if provided") ] = None, from_task_run: Annotated[ TaskRun | None, Doc("The task run creating this one, if any. It records the chain."), ] = None, **params: Annotated[ Any, Doc( "The optional parameters for the task run. " "They must match the task params model" ), ], ) -> TaskRun: """Create a [TaskRun][fluid.scheduler.TaskRun] in `init` state""" task = self.broker.task_from_registry(task) run_id = run_id or self.broker.new_uuid() return TaskRun( id=run_id, task=task, priority=priority or task.priority, params=task.params_model(**params), task_manager=self, from_run_id=from_task_run.id if from_task_run else "", # the first run in a chain has no root of its own, it is the root root_run_id=( (from_task_run.root_run_id or from_task_run.id) if from_task_run else "" ), ) ``` ### register_from_module ```python register_from_module(module, tags=None) ``` Register tasks from a python module | PARAMETER | DESCRIPTION | | --------- | ------------------------------------------------------------------------------------------------------------------------------- | | `module` | Python module with tasks implementations - can contain any object, only instances of Task are registered **TYPE:** `ModuleType` | | `tags` | Extra tags to add to every registered task **TYPE:** \`Sequence[str] | Source code in `fluid/scheduler/consumer.py` ```python def register_from_module( self, module: Annotated[ ModuleType, Doc( "Python module with tasks implementations " "- can contain any object, only instances of Task are registered" ), ], tags: Annotated[ Sequence[str] | None, Doc("Extra tags to add to every registered task"), ] = None, ) -> None: """Register tasks from a python module""" for name in dir(module): if name.startswith("_"): continue if isinstance(obj := getattr(module, name), Task): self.register_task(obj, tags=tags) ``` ### register_from_dict ```python register_from_dict(data, tags=None) ``` Register tasks from a python dictionary | PARAMETER | DESCRIPTION | | --------- | --------------------------------------------------------------------------------------------------------------------------------------- | | `data` | Python dictionary with tasks implementations - can contain any object, only instances of Task are registered **TYPE:** `dict[str, Any]` | | `tags` | Extra tags to add to every registered task **TYPE:** \`Sequence[str] | Source code in `fluid/scheduler/consumer.py` ```python def register_from_dict( self, data: Annotated[ dict[str, Any], Doc( "Python dictionary with tasks implementations " "- can contain any object, only instances of Task are registered" ), ], tags: Annotated[ Sequence[str] | None, Doc("Extra tags to add to every registered task"), ] = None, ) -> None: """Register tasks from a python dictionary""" for name, obj in data.items(): if name.startswith("_"): continue if isinstance(obj, Task): self.register_task(obj, tags=tags) ``` ### register_async_handler ```python register_async_handler(event, handler) ``` Source code in `fluid/scheduler/consumer.py` ```python def register_async_handler(self, event: Event | str, handler: AsyncHandler) -> None: event = Event.from_string_or_event(event) self.dispatcher.register_handler( f"{event.type}.async_dispatch", self._async_dispatcher_worker.send, ) self._async_dispatcher_worker.dispatcher.register_handler(event, handler) ``` ### unregister_async_handler ```python unregister_async_handler(event) ``` Source code in `fluid/scheduler/consumer.py` ```python def unregister_async_handler(self, event: Event | str) -> AsyncHandler | None: return self._async_dispatcher_worker.dispatcher.unregister_handler(event) ``` ### with_plugin ```python with_plugin(plugin) ``` Register a plugin with the task manager | PARAMETER | DESCRIPTION | | --------- | ---------------------------------------------------- | | `plugin` | The plugin to register **TYPE:** `TaskManagerPlugin` | Source code in `fluid/scheduler/consumer.py` ```python def with_plugin( self, plugin: Annotated[TaskManagerPlugin, Doc("The plugin to register")], ) -> Self: """Register a plugin with the task manager""" self._plugins.append(plugin) plugin.register(self) return self ``` ### sync_queue ```python sync_queue(task, delay=0) ``` Queue a task synchronously Source code in `fluid/scheduler/consumer.py` ```python def sync_queue(self, task: str | Task | TaskRun, delay: float = 0) -> None: """Queue a task synchronously""" self._in_process_queue.queue(task, delay=delay) ``` ### queue_and_wait ```python queue_and_wait(task, *, timeout=None, **params) ``` Queue a task and wait for it to finish | PARAMETER | DESCRIPTION | | ---------- | ----------------------------------------------------------------------------------------------------------------- | | `task` | The task or task name, if a task name it must be registered with the task manager. **TYPE:** \`str | | `timeout` | Timeout for waiting the task to finish **TYPE:** \`int | | `**params` | The optional parameters for the task run. They must match the task params model **TYPE:** `Any` **DEFAULT:** `{}` | Source code in `fluid/scheduler/consumer.py` ```python async def queue_and_wait( self, task: Annotated[ str | Task, Doc( "The task or task name," " if a task name it must be registered with the task manager." ), ], *, timeout: Annotated[ int | None, Doc("Timeout for waiting the task to finish") ] = None, **params: Annotated[ Any, Doc( "The optional parameters for the task run. " "They must match the task params model" ), ], ) -> TaskRun: """Queue a task and wait for it to finish""" with TaskRunWaiter(self) as waiter: task_run = await self.queue(task, **params) return await waiter.wait(task_run, timeout=timeout) ``` # Task Scheduling Scheduling functions for tasks. They can be imported from `fluid.scheduler`: ```python from fluid.scheduler import every, crontab ``` ## fluid.scheduler.Scheduler Bases: `ABC` Base class for all schedulers. ### info ```python info() ``` Return a string representation of the schedule. Source code in `fluid/scheduler/scheduler_crontab.py` ```python @abstractmethod def info(self) -> str: """Return a string representation of the schedule.""" ``` ## fluid.scheduler.every ```python every(delta, delay=timedelta(), jitter=timedelta()) ``` Bases: `Scheduler` Run a task every delta time, with optional delay and jitter | PARAMETER | DESCRIPTION | | --------- | --------------------------------------------------------------------------------------------- | | `delta` | The time delta between runs **TYPE:** `timedelta` | | `delay` | The initial delay before the first run **TYPE:** `timedelta` **DEFAULT:** `timedelta()` | | `jitter` | The maximum random jitter added to the delta **TYPE:** `timedelta` **DEFAULT:** `timedelta()` | Source code in `fluid/scheduler/scheduler_every.py` ```python def __init__( self, delta: Annotated[ timedelta, Doc("The time delta between runs"), ], delay: Annotated[ timedelta, Doc("The initial delay before the first run"), ] = timedelta(), jitter: Annotated[ timedelta, Doc("The maximum random jitter added to the delta"), ] = timedelta(), ) -> None: self.delta: timedelta = delta self.delay: timedelta = delay self.jitter: timedelta = jitter self._delta: timedelta = self.next_delta() self._started: datetime | None = None ``` ### delta ```python delta = delta ``` ### delay ```python delay = delay ``` ### jitter ```python jitter = jitter ``` ### info ```python info() ``` Source code in `fluid/scheduler/scheduler_every.py` ```python def info(self) -> str: return str(self.delta) ``` ### next_delta ```python next_delta() ``` Source code in `fluid/scheduler/scheduler_every.py` ```python def next_delta(self) -> timedelta: return self.delta + random.uniform(0, 1) * self.jitter ``` ## fluid.scheduler.crontab ```python crontab( minute="*", hour="*", day="*", month="*", day_of_week="*", tz=utc, ) ``` Bases: `Scheduler` Convert a "crontab"-style set of parameters into a test function that will return True when the given datetime matches the parameters set forth in the crontab. For day-of-week, 0=Sunday and 6=Saturday. Acceptable inputs: * = every distinct value */n = run every "n" times, i.e. hours='*/4' == 0, 4, 8, 12, 16, 20 m-n = run every time m..n m,n = run on m and n Source code in `fluid/scheduler/scheduler_crontab.py` ```python def __init__( self, minute: CI = "*", hour: CI = "*", day: CI = "*", month: CI = "*", day_of_week: CI = "*", tz: tzinfo = timezone.utc, ) -> None: self.tz: tzinfo = tz self._info = ( f"minute {minute}; hour {hour}; day {day}; month {month}; " f"day_of_week {day_of_week}" ) validation = ( ("m", month, range(1, 13)), ("d", day, range(1, 32)), ("w", day_of_week, range(8)), # 0-6, but also 7 for Sunday. ("H", hour, range(24)), ("M", minute, range(60)), ) cron_settings = [] for date_str, value, acceptable in validation: settings: Set[int] = set() if isinstance(value, int): value = str(value) for piece in value.split(","): if piece == "*": settings.update(acceptable) continue if piece.isdigit(): digit = int(piece) if digit not in acceptable: raise ValueError("%d is not a valid input" % digit) elif date_str == "w": digit %= 7 settings.add(digit) else: dash_match = dash_re.match(piece) if dash_match: lhs, rhs = map(int, dash_match.groups()) if lhs not in acceptable or rhs not in acceptable: raise ValueError("%s is not a valid input" % piece) elif date_str == "w": lhs %= 7 rhs %= 7 settings.update(range(lhs, rhs + 1)) continue # Handle stuff like */3, */6. every_match = every_re.match(piece) if every_match: if date_str == "w": raise ValueError( "Cannot perform this kind of matching" " on day-of-week." ) interval = int(every_match.groups()[0]) settings.update(acceptable[::interval]) cron_settings.append(sorted(list(settings))) self.cron_settings = tuple(cron_settings) ``` ### tz ```python tz = tz ``` ### cron_settings ```python cron_settings = tuple(cron_settings) ``` ### info ```python info() ``` Source code in `fluid/scheduler/scheduler_crontab.py` ```python def info(self) -> str: return self._info ``` # Utils ## fluid.utils.lazy.LazyGroup ```python LazyGroup(*, lazy_subcommands=None, **kwargs) ``` Bases: `Group` A click Group that can lazily load subcommands This class extends the click.Group class to allow for subcommands to be lazily loaded from a module path. It is useful when you have a large number of subcommands that you don't want to load until they are actually needed. Available with the `cli` extra dependencies. | PARAMETER | DESCRIPTION | | ------------------ | ----------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `lazy_subcommands` | A dictionary mapping command names to their import paths. This allows subcommands to be lazily loaded from the specified module paths. **TYPE:** \`dict[str, str] | | `**kwargs` | Additional keyword arguments passed to the click.Group initializer. **TYPE:** `Any` **DEFAULT:** `{}` | Source code in `fluid/utils/lazy.py` ```python def __init__( self, *, lazy_subcommands: Annotated[ dict[str, str] | None, Doc(""" A dictionary mapping command names to their import paths. This allows subcommands to be lazily loaded from the specified module paths. """), ] = None, **kwargs: Annotated[ Any, Doc(""" Additional keyword arguments passed to the click.Group initializer. """), ], ): super().__init__(**kwargs) self.lazy_subcommands = lazy_subcommands or {} ``` ### lazy_subcommands ```python lazy_subcommands = lazy_subcommands or {} ``` ### list_commands ```python list_commands(ctx) ``` Source code in `fluid/utils/lazy.py` ```python def list_commands(self, ctx: click.Context) -> list[str]: commands = super().list_commands(ctx) commands.extend(self.lazy_subcommands) return sorted(commands) ``` ### get_command ```python get_command(ctx, cmd_name) ``` Source code in `fluid/utils/lazy.py` ```python def get_command(self, ctx: click.Context, cmd_name: str) -> click.Command | None: if cmd_name in self.lazy_subcommands: return self._lazy_load(cmd_name) return super().get_command(ctx, cmd_name) ``` ## fluid.utils.log.config ```python config( level=None, other_level=WARNING, app_names=None, log_handler=None, log_format=None, formatters=None, ) ``` Configure logging for the application | PARAMETER | DESCRIPTION | | ------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------- | | `level` | Log levels for application loggers defined by the app_names parameter. By default this value is taken from the LOG_LEVEL env variable **TYPE:** \`str | | `other_level` | log levels for loggers not prefixed by app_names **TYPE:** \`str | | `app_names` | Application names for which the log level is set, these are the prefixes which will be set at log_level **TYPE:** \`Sequence[str] | | `log_handler` | Log handler to use, by default it is taken from the LOG_HANDLER env variable and if missing plain is used **TYPE:** \`str | | `log_format` | log format to use, by default it is taken from the PYTHON_LOG_FORMAT env variable **TYPE:** \`str | | `formatters` | Additional formatters to add to the logging configuration **TYPE:** \`dict\[str, dict[str, str]\] | Source code in `fluid/utils/log.py` ```python def config( level: Annotated[ str | int | None, Doc( "Log levels for application loggers defined by the `app_names` parameter. " "By default this value is taken from the `LOG_LEVEL` env variable" ), ] = None, other_level: Annotated[ str | int, Doc("log levels for loggers not prefixed by `app_names`"), ] = logging.WARNING, app_names: Annotated[ Sequence[str] | None, Doc( "Application names for which the log level is set, " "these are the prefixes which will be set at `log_level`" ), ] = None, log_handler: Annotated[ str | None, Doc( "Log handler to use, by default it is taken from the " "`LOG_HANDLER` env variable and if missing `plain` is used" ), ] = None, log_format: Annotated[ str | None, Doc( "log format to use, by default it is taken from the " "`PYTHON_LOG_FORMAT` env variable" ), ] = None, formatters: Annotated[ dict[str, dict[str, str]] | None, Doc("Additional formatters to add to the logging configuration"), ] = None, ) -> dict: """Configure logging for the application""" cfg = _log_config( level=level, other_level=other_level, app_names=app_names, log_handler=log_handler, log_format=log_format, formatters=formatters, ) dictConfig(cfg) return cfg ``` # Workers Workers are the main building block for asynchronous programming with `aio-fluid`. They are responsible for running asynchronous tasks and managing their lifecycle. There are several worker classes which can be imported from `fluid.utils.worker`, and they aall derive from the abstract `fluid.utils.worker.Worker` class. ```python from fluid.utils.worker import Worker ``` ## fluid.utils.worker.WorkerState Bases: `StrEnum` The lifecycle state of a Worker. ### INIT ```python INIT = enum.auto() ``` Worker has been created but not yet started. ### RUNNING ```python RUNNING = enum.auto() ``` Worker is actively executing its run loop. ### STOPPING ```python STOPPING = enum.auto() ``` Graceful stop requested; run should exit at its next safe point. ### STOPPED ```python STOPPED = enum.auto() ``` Worker exited cleanly after a graceful stop. ### FORCE_STOPPED ```python FORCE_STOPPED = enum.auto() ``` Worker was cancelled because it did not exit within the grace period. ## fluid.utils.worker.Worker ```python Worker(*, name='', stopping_grace_period=None) ``` Bases: `ABC` Abstract base class for all workers. A worker encapsulates a long-running async task with a managed lifecycle. Subclasses implement run, which is called once the worker is started and should loop until is_running returns `False`. Use startup to start the worker as an asyncio task, and shutdown (or gracefully_stop + wait_for_shutdown) to stop it. Override on_startup and on_shutdown to initialise and clean up async resources that the worker owns. | PARAMETER | DESCRIPTION | | ----------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `name` | Worker's name, if not provided it is evaluated from the class name **TYPE:** `str` **DEFAULT:** `''` | | `stopping_grace_period` | Grace period in seconds to wait for workers to stop running when this worker is shutdown. It defaults to the FLUID_STOPPING_GRACE_PERIOD environment variable or 10 seconds. **TYPE:** \`float | Source code in `fluid/utils/worker.py` ```python def __init__( self, *, name: Annotated[ str, Doc("Worker's name, if not provided it is evaluated from the class name"), ] = "", stopping_grace_period: Annotated[ float | None, Doc( "Grace period in seconds to wait for workers to stop running " "when this worker is shutdown. " "It defaults to the `FLUID_STOPPING_GRACE_PERIOD` " "environment variable or 10 seconds." ), ] = None, ) -> None: if stopping_grace_period is None: stopping_grace_period = settings.STOPPING_GRACE_PERIOD self._worker_name: str = name or snake_case(type(self).__name__) self._worker_state: WorkerState = WorkerState.INIT self._stopping_grace_period = stopping_grace_period self._worker_task_runner: WorkerTaskRunner | None = None ``` ### worker_state ```python worker_state ``` The running state of the worker ### worker_name ```python worker_name ``` The name of the worker ### num_workers ```python num_workers ``` The number of workers in this worker ### has_started ```python has_started() ``` Source code in `fluid/utils/worker.py` ```python def has_started(self) -> bool: return self._worker_state != WorkerState.INIT ``` ### is_running ```python is_running() ``` Source code in `fluid/utils/worker.py` ```python def is_running(self) -> bool: return self._worker_state == WorkerState.RUNNING ``` ### is_stopping ```python is_stopping() ``` Source code in `fluid/utils/worker.py` ```python def is_stopping(self) -> bool: return self._worker_state == WorkerState.STOPPING ``` ### is_stopped ```python is_stopped() ``` Source code in `fluid/utils/worker.py` ```python def is_stopped(self) -> bool: return self._worker_state in (WorkerState.STOPPED, WorkerState.FORCE_STOPPED) ``` ### gracefully_stop ```python gracefully_stop() ``` Try to gracefully stop the worker Source code in `fluid/utils/worker.py` ```python def gracefully_stop(self) -> None: """Try to gracefully stop the worker""" if self.is_running(): self._worker_state = WorkerState.STOPPING ``` ### after_shutdown ```python after_shutdown(reason, code) ``` Called after shutdown of worker By default it does nothing, but can be overriden to do something such as exit the process. Source code in `fluid/utils/worker.py` ```python def after_shutdown(self, reason: str, code: int) -> None: # noqa: B027 """Called after shutdown of worker By default it does nothing, but can be overriden to do something such as exit the process. """ ``` ### status ```python status() ``` Source code in `fluid/utils/worker.py` ```python async def status(self) -> dict: return {"stopping": self.is_stopping(), "running": self.is_running()} ``` ### on_startup ```python on_startup() ``` Called when the worker starts running Use this function to initialize other async resources connected with the worker Source code in `fluid/utils/worker.py` ```python async def on_startup(self) -> None: # noqa: B027 """Called when the worker starts running Use this function to initialize other async resources connected with the worker """ ``` ### on_shutdown ```python on_shutdown() ``` called after the worker stopped running Use this function to cleanup resources connected with the worker Source code in `fluid/utils/worker.py` ```python async def on_shutdown(self) -> None: # noqa: B027 """called after the worker stopped running Use this function to cleanup resources connected with the worker """ ``` ### startup ```python startup() ``` start the worker This method creates a task to run the worker. Source code in `fluid/utils/worker.py` ```python async def startup(self) -> None: """start the worker This method creates a task to run the worker. """ if self.has_started(): raise WorkerStartError( "worker %s already started: %s", self.worker_name, self._worker_state ) else: self._worker_task_runner = await WorkerTaskRunner.start(self) ``` ### shutdown ```python shutdown() ``` Shutdown a running worker and wait for it to stop This method will try to gracefully stop the worker and wait for it to stop. If the worker does not stop in the grace period, it will force shutdown by cancelling the task. Source code in `fluid/utils/worker.py` ```python async def shutdown(self) -> None: """Shutdown a running worker and wait for it to stop This method will try to gracefully stop the worker and wait for it to stop. If the worker does not stop in the grace period, it will force shutdown by cancelling the task. """ if self._worker_task_runner is not None: await self._worker_task_runner.shutdown() ``` ### wait_for_shutdown ```python wait_for_shutdown() ``` Wait for the worker to stop This method will wait for the worker to stop running, but doesn't try to gracefully stop it nor force shutdown. Source code in `fluid/utils/worker.py` ```python async def wait_for_shutdown(self) -> None: """Wait for the worker to stop This method will wait for the worker to stop running, but doesn't try to gracefully stop it nor force shutdown. """ if self._worker_task_runner is not None: await self._worker_task_runner.wait_for_shutdown() ``` ### workers ```python workers() ``` An iterator of workers in this worker Source code in `fluid/utils/worker.py` ```python def workers(self) -> Iterator[Worker]: """An iterator of workers in this worker""" yield self ``` ### run ```python run() ``` run the worker This is the only abstract method and that needs implementing. It is the coroutine that mantains the worker running. Source code in `fluid/utils/worker.py` ```python @abstractmethod async def run(self) -> None: """run the worker This is the only abstract method and that needs implementing. It is the coroutine that mantains the worker running. """ ``` ## fluid.utils.worker.WorkerFunction ```python WorkerFunction( run_function, *, heartbeat=0, name="", stopping_grace_period=None ) ``` Bases: `Worker` A Worker that calls a coroutine function in a loop. On each iteration the supplied `run_function` is awaited, then the worker sleeps for `heartbeat` seconds before repeating. The loop exits when is_running returns `False`. | PARAMETER | DESCRIPTION | | ----------------------- | ----------------------------------------------------------------------------------------------------------------------- | | `run_function` | The coroutine function tuo run and await at each iteration of the worker loop **TYPE:** `Callable[[], Awaitable[None]]` | | `heartbeat` | The time to wait between each coroutine function run **TYPE:** \`float | | `name` | Worker's name, if not provided it is evaluated from the class name **TYPE:** `str` **DEFAULT:** `''` | | `stopping_grace_period` | Grace period in seconds before force-cancelling this worker **TYPE:** \`float | Source code in `fluid/utils/worker.py` ```python def __init__( self, run_function: Annotated[ Callable[[], Awaitable[None]], Doc( "The coroutine function tuo run and await at each iteration " "of the worker loop" ), ], *, heartbeat: Annotated[ float | int, Doc("The time to wait between each coroutine function run") ] = 0, name: Annotated[ str, Doc("Worker's name, if not provided it is evaluated from the class name"), ] = "", stopping_grace_period: Annotated[ float | None, Doc("Grace period in seconds before force-cancelling this worker"), ] = None, ) -> None: super().__init__(name=name, stopping_grace_period=stopping_grace_period) self._run_function = run_function self._heartbeat = heartbeat ``` ### worker_state ```python worker_state ``` The running state of the worker ### worker_name ```python worker_name ``` The name of the worker ### num_workers ```python num_workers ``` The number of workers in this worker ### run ```python run() ``` Source code in `fluid/utils/worker.py` ```python async def run(self) -> None: while self.is_running(): await self._run_function() await asyncio.sleep(self._heartbeat) ``` ### has_started ```python has_started() ``` Source code in `fluid/utils/worker.py` ```python def has_started(self) -> bool: return self._worker_state != WorkerState.INIT ``` ### is_running ```python is_running() ``` Source code in `fluid/utils/worker.py` ```python def is_running(self) -> bool: return self._worker_state == WorkerState.RUNNING ``` ### is_stopping ```python is_stopping() ``` Source code in `fluid/utils/worker.py` ```python def is_stopping(self) -> bool: return self._worker_state == WorkerState.STOPPING ``` ### is_stopped ```python is_stopped() ``` Source code in `fluid/utils/worker.py` ```python def is_stopped(self) -> bool: return self._worker_state in (WorkerState.STOPPED, WorkerState.FORCE_STOPPED) ``` ### gracefully_stop ```python gracefully_stop() ``` Try to gracefully stop the worker Source code in `fluid/utils/worker.py` ```python def gracefully_stop(self) -> None: """Try to gracefully stop the worker""" if self.is_running(): self._worker_state = WorkerState.STOPPING ``` ### after_shutdown ```python after_shutdown(reason, code) ``` Called after shutdown of worker By default it does nothing, but can be overriden to do something such as exit the process. Source code in `fluid/utils/worker.py` ```python def after_shutdown(self, reason: str, code: int) -> None: # noqa: B027 """Called after shutdown of worker By default it does nothing, but can be overriden to do something such as exit the process. """ ``` ### status ```python status() ``` Source code in `fluid/utils/worker.py` ```python async def status(self) -> dict: return {"stopping": self.is_stopping(), "running": self.is_running()} ``` ### on_startup ```python on_startup() ``` Called when the worker starts running Use this function to initialize other async resources connected with the worker Source code in `fluid/utils/worker.py` ```python async def on_startup(self) -> None: # noqa: B027 """Called when the worker starts running Use this function to initialize other async resources connected with the worker """ ``` ### on_shutdown ```python on_shutdown() ``` called after the worker stopped running Use this function to cleanup resources connected with the worker Source code in `fluid/utils/worker.py` ```python async def on_shutdown(self) -> None: # noqa: B027 """called after the worker stopped running Use this function to cleanup resources connected with the worker """ ``` ### startup ```python startup() ``` start the worker This method creates a task to run the worker. Source code in `fluid/utils/worker.py` ```python async def startup(self) -> None: """start the worker This method creates a task to run the worker. """ if self.has_started(): raise WorkerStartError( "worker %s already started: %s", self.worker_name, self._worker_state ) else: self._worker_task_runner = await WorkerTaskRunner.start(self) ``` ### shutdown ```python shutdown() ``` Shutdown a running worker and wait for it to stop This method will try to gracefully stop the worker and wait for it to stop. If the worker does not stop in the grace period, it will force shutdown by cancelling the task. Source code in `fluid/utils/worker.py` ```python async def shutdown(self) -> None: """Shutdown a running worker and wait for it to stop This method will try to gracefully stop the worker and wait for it to stop. If the worker does not stop in the grace period, it will force shutdown by cancelling the task. """ if self._worker_task_runner is not None: await self._worker_task_runner.shutdown() ``` ### wait_for_shutdown ```python wait_for_shutdown() ``` Wait for the worker to stop This method will wait for the worker to stop running, but doesn't try to gracefully stop it nor force shutdown. Source code in `fluid/utils/worker.py` ```python async def wait_for_shutdown(self) -> None: """Wait for the worker to stop This method will wait for the worker to stop running, but doesn't try to gracefully stop it nor force shutdown. """ if self._worker_task_runner is not None: await self._worker_task_runner.wait_for_shutdown() ``` ### workers ```python workers() ``` An iterator of workers in this worker Source code in `fluid/utils/worker.py` ```python def workers(self) -> Iterator[Worker]: """An iterator of workers in this worker""" yield self ``` ## fluid.utils.worker.QueueConsumer ```python QueueConsumer(*, name='', stopping_grace_period=None) ``` Bases: `Worker`, `MessageProducer[MessageType]` Abstract Worker backed by an asyncio queue. Provides send for thread-safe message delivery and get_message for retrieving the next message with a timeout. Subclasses implement run to consume messages from the queue. | PARAMETER | DESCRIPTION | | ----------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `name` | Worker's name, if not provided it is evaluated from the class name **TYPE:** `str` **DEFAULT:** `''` | | `stopping_grace_period` | Grace period in seconds to wait for workers to stop running when this worker is shutdown. It defaults to the FLUID_STOPPING_GRACE_PERIOD environment variable or 10 seconds. **TYPE:** \`float | Source code in `fluid/utils/worker.py` ```python def __init__( self, *, name: Annotated[ str, Doc("Worker's name, if not provided it is evaluated from the class name"), ] = "", stopping_grace_period: Annotated[ float | None, Doc( "Grace period in seconds to wait for workers to stop running " "when this worker is shutdown. " "It defaults to the `FLUID_STOPPING_GRACE_PERIOD` " "environment variable or 10 seconds." ), ] = None, ) -> None: super().__init__(name=name, stopping_grace_period=stopping_grace_period) self._queue: asyncio.Queue[MessageType | None] = asyncio.Queue() ``` ### worker_state ```python worker_state ``` The running state of the worker ### worker_name ```python worker_name ``` The name of the worker ### num_workers ```python num_workers ``` The number of workers in this worker ### get_message ```python get_message(timeout=0.5) ``` Get the next message from the queue Source code in `fluid/utils/worker.py` ```python async def get_message(self, timeout: float = 0.5) -> MessageType | None: """Get the next message from the queue""" try: async with asyncio.timeout(timeout): return await self._queue.get() except asyncio.TimeoutError: return None except (asyncio.CancelledError, RuntimeError): if not self.is_stopping(): raise return None ``` ### queue_size ```python queue_size() ``` Get the size of the queue Source code in `fluid/utils/worker.py` ```python def queue_size(self) -> int: """Get the size of the queue""" return self._queue.qsize() ``` ### status ```python status() ``` Source code in `fluid/utils/worker.py` ```python async def status(self) -> dict: status = await super().status() status.update(queue_size=self.queue_size()) return status ``` ### send ```python send(message) ``` Send a message into the worker Source code in `fluid/utils/worker.py` ```python def send(self, message: MessageType | None) -> None: """Send a message into the worker""" self._queue.put_nowait(message) ``` ### has_started ```python has_started() ``` Source code in `fluid/utils/worker.py` ```python def has_started(self) -> bool: return self._worker_state != WorkerState.INIT ``` ### is_running ```python is_running() ``` Source code in `fluid/utils/worker.py` ```python def is_running(self) -> bool: return self._worker_state == WorkerState.RUNNING ``` ### is_stopping ```python is_stopping() ``` Source code in `fluid/utils/worker.py` ```python def is_stopping(self) -> bool: return self._worker_state == WorkerState.STOPPING ``` ### is_stopped ```python is_stopped() ``` Source code in `fluid/utils/worker.py` ```python def is_stopped(self) -> bool: return self._worker_state in (WorkerState.STOPPED, WorkerState.FORCE_STOPPED) ``` ### gracefully_stop ```python gracefully_stop() ``` Try to gracefully stop the worker Source code in `fluid/utils/worker.py` ```python def gracefully_stop(self) -> None: """Try to gracefully stop the worker""" if self.is_running(): self._worker_state = WorkerState.STOPPING ``` ### after_shutdown ```python after_shutdown(reason, code) ``` Called after shutdown of worker By default it does nothing, but can be overriden to do something such as exit the process. Source code in `fluid/utils/worker.py` ```python def after_shutdown(self, reason: str, code: int) -> None: # noqa: B027 """Called after shutdown of worker By default it does nothing, but can be overriden to do something such as exit the process. """ ``` ### on_startup ```python on_startup() ``` Called when the worker starts running Use this function to initialize other async resources connected with the worker Source code in `fluid/utils/worker.py` ```python async def on_startup(self) -> None: # noqa: B027 """Called when the worker starts running Use this function to initialize other async resources connected with the worker """ ``` ### on_shutdown ```python on_shutdown() ``` called after the worker stopped running Use this function to cleanup resources connected with the worker Source code in `fluid/utils/worker.py` ```python async def on_shutdown(self) -> None: # noqa: B027 """called after the worker stopped running Use this function to cleanup resources connected with the worker """ ``` ### startup ```python startup() ``` start the worker This method creates a task to run the worker. Source code in `fluid/utils/worker.py` ```python async def startup(self) -> None: """start the worker This method creates a task to run the worker. """ if self.has_started(): raise WorkerStartError( "worker %s already started: %s", self.worker_name, self._worker_state ) else: self._worker_task_runner = await WorkerTaskRunner.start(self) ``` ### shutdown ```python shutdown() ``` Shutdown a running worker and wait for it to stop This method will try to gracefully stop the worker and wait for it to stop. If the worker does not stop in the grace period, it will force shutdown by cancelling the task. Source code in `fluid/utils/worker.py` ```python async def shutdown(self) -> None: """Shutdown a running worker and wait for it to stop This method will try to gracefully stop the worker and wait for it to stop. If the worker does not stop in the grace period, it will force shutdown by cancelling the task. """ if self._worker_task_runner is not None: await self._worker_task_runner.shutdown() ``` ### wait_for_shutdown ```python wait_for_shutdown() ``` Wait for the worker to stop This method will wait for the worker to stop running, but doesn't try to gracefully stop it nor force shutdown. Source code in `fluid/utils/worker.py` ```python async def wait_for_shutdown(self) -> None: """Wait for the worker to stop This method will wait for the worker to stop running, but doesn't try to gracefully stop it nor force shutdown. """ if self._worker_task_runner is not None: await self._worker_task_runner.wait_for_shutdown() ``` ### workers ```python workers() ``` An iterator of workers in this worker Source code in `fluid/utils/worker.py` ```python def workers(self) -> Iterator[Worker]: """An iterator of workers in this worker""" yield self ``` ### run ```python run() ``` run the worker This is the only abstract method and that needs implementing. It is the coroutine that mantains the worker running. Source code in `fluid/utils/worker.py` ```python @abstractmethod async def run(self) -> None: """run the worker This is the only abstract method and that needs implementing. It is the coroutine that mantains the worker running. """ ``` ## fluid.utils.worker.QueueConsumerWorker ```python QueueConsumerWorker( on_message, *, name="", stopping_grace_period=None ) ``` Bases: `QueueConsumer[MessageType]` A QueueConsumer that dispatches each message to a single async callback. | PARAMETER | DESCRIPTION | | ----------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `on_message` | The async callback to call when a message is received **TYPE:** `Callable[[MessageType], Awaitable[None]]` | | `name` | Worker's name, if not provided it is evaluated from the class name **TYPE:** `str` **DEFAULT:** `''` | | `stopping_grace_period` | Grace period in seconds to wait for workers to stop running when this worker is shutdown. It defaults to the FLUID_STOPPING_GRACE_PERIOD environment variable or 10 seconds. **TYPE:** \`float | Source code in `fluid/utils/worker.py` ```python def __init__( self, on_message: Annotated[ Callable[[MessageType], Awaitable[None]], Doc("The async callback to call when a message is received"), ], *, name: Annotated[ str, Doc("Worker's name, if not provided it is evaluated from the class name"), ] = "", stopping_grace_period: Annotated[ float | None, Doc( "Grace period in seconds to wait for workers to stop running " "when this worker is shutdown. " "It defaults to the `FLUID_STOPPING_GRACE_PERIOD` " "environment variable or 10 seconds." ), ] = None, ) -> None: super().__init__(name=name, stopping_grace_period=stopping_grace_period) self.on_message = on_message ``` ### on_message ```python on_message = on_message ``` ### worker_state ```python worker_state ``` The running state of the worker ### worker_name ```python worker_name ``` The name of the worker ### num_workers ```python num_workers ``` The number of workers in this worker ### run ```python run() ``` Source code in `fluid/utils/worker.py` ```python async def run(self) -> None: while not self.is_stopping(): message = await self.get_message() if message is not None: await self.on_message(message) ``` ### send ```python send(message) ``` Send a message into the worker Source code in `fluid/utils/worker.py` ```python def send(self, message: MessageType | None) -> None: """Send a message into the worker""" self._queue.put_nowait(message) ``` ### has_started ```python has_started() ``` Source code in `fluid/utils/worker.py` ```python def has_started(self) -> bool: return self._worker_state != WorkerState.INIT ``` ### is_running ```python is_running() ``` Source code in `fluid/utils/worker.py` ```python def is_running(self) -> bool: return self._worker_state == WorkerState.RUNNING ``` ### is_stopping ```python is_stopping() ``` Source code in `fluid/utils/worker.py` ```python def is_stopping(self) -> bool: return self._worker_state == WorkerState.STOPPING ``` ### is_stopped ```python is_stopped() ``` Source code in `fluid/utils/worker.py` ```python def is_stopped(self) -> bool: return self._worker_state in (WorkerState.STOPPED, WorkerState.FORCE_STOPPED) ``` ### gracefully_stop ```python gracefully_stop() ``` Try to gracefully stop the worker Source code in `fluid/utils/worker.py` ```python def gracefully_stop(self) -> None: """Try to gracefully stop the worker""" if self.is_running(): self._worker_state = WorkerState.STOPPING ``` ### after_shutdown ```python after_shutdown(reason, code) ``` Called after shutdown of worker By default it does nothing, but can be overriden to do something such as exit the process. Source code in `fluid/utils/worker.py` ```python def after_shutdown(self, reason: str, code: int) -> None: # noqa: B027 """Called after shutdown of worker By default it does nothing, but can be overriden to do something such as exit the process. """ ``` ### status ```python status() ``` Source code in `fluid/utils/worker.py` ```python async def status(self) -> dict: status = await super().status() status.update(queue_size=self.queue_size()) return status ``` ### on_startup ```python on_startup() ``` Called when the worker starts running Use this function to initialize other async resources connected with the worker Source code in `fluid/utils/worker.py` ```python async def on_startup(self) -> None: # noqa: B027 """Called when the worker starts running Use this function to initialize other async resources connected with the worker """ ``` ### on_shutdown ```python on_shutdown() ``` called after the worker stopped running Use this function to cleanup resources connected with the worker Source code in `fluid/utils/worker.py` ```python async def on_shutdown(self) -> None: # noqa: B027 """called after the worker stopped running Use this function to cleanup resources connected with the worker """ ``` ### startup ```python startup() ``` start the worker This method creates a task to run the worker. Source code in `fluid/utils/worker.py` ```python async def startup(self) -> None: """start the worker This method creates a task to run the worker. """ if self.has_started(): raise WorkerStartError( "worker %s already started: %s", self.worker_name, self._worker_state ) else: self._worker_task_runner = await WorkerTaskRunner.start(self) ``` ### shutdown ```python shutdown() ``` Shutdown a running worker and wait for it to stop This method will try to gracefully stop the worker and wait for it to stop. If the worker does not stop in the grace period, it will force shutdown by cancelling the task. Source code in `fluid/utils/worker.py` ```python async def shutdown(self) -> None: """Shutdown a running worker and wait for it to stop This method will try to gracefully stop the worker and wait for it to stop. If the worker does not stop in the grace period, it will force shutdown by cancelling the task. """ if self._worker_task_runner is not None: await self._worker_task_runner.shutdown() ``` ### wait_for_shutdown ```python wait_for_shutdown() ``` Wait for the worker to stop This method will wait for the worker to stop running, but doesn't try to gracefully stop it nor force shutdown. Source code in `fluid/utils/worker.py` ```python async def wait_for_shutdown(self) -> None: """Wait for the worker to stop This method will wait for the worker to stop running, but doesn't try to gracefully stop it nor force shutdown. """ if self._worker_task_runner is not None: await self._worker_task_runner.wait_for_shutdown() ``` ### workers ```python workers() ``` An iterator of workers in this worker Source code in `fluid/utils/worker.py` ```python def workers(self) -> Iterator[Worker]: """An iterator of workers in this worker""" yield self ``` ### get_message ```python get_message(timeout=0.5) ``` Get the next message from the queue Source code in `fluid/utils/worker.py` ```python async def get_message(self, timeout: float = 0.5) -> MessageType | None: """Get the next message from the queue""" try: async with asyncio.timeout(timeout): return await self._queue.get() except asyncio.TimeoutError: return None except (asyncio.CancelledError, RuntimeError): if not self.is_stopping(): raise return None ``` ### queue_size ```python queue_size() ``` Get the size of the queue Source code in `fluid/utils/worker.py` ```python def queue_size(self) -> int: """Get the size of the queue""" return self._queue.qsize() ``` ## fluid.utils.worker.AsyncConsumer ```python AsyncConsumer( dispatcher, *, name="", stopping_grace_period=None ) ``` Bases: `QueueConsumer[MessageType]` A QueueConsumer that fans out each message to all registered async handlers via an AsyncDispatcher. The run loop processes messages until is_stopping returns `True`. Any messages remaining in the queue when the worker stops are discarded; callers that need guaranteed delivery should drain the queue before requesting a stop. | PARAMETER | DESCRIPTION | | ----------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `dispatcher` | Async message dispatcher to dispatch messages **TYPE:** `AsyncDispatcher[MessageType]` | | `name` | Worker's name, if not provided it is evaluated from the class name **TYPE:** `str` **DEFAULT:** `''` | | `stopping_grace_period` | Grace period in seconds to wait for workers to stop running when this worker is shutdown. It defaults to the FLUID_STOPPING_GRACE_PERIOD environment variable or 10 seconds. **TYPE:** \`float | Source code in `fluid/utils/worker.py` ```python def __init__( self, dispatcher: Annotated[ AsyncDispatcher[MessageType], Doc("Async message dispatcher to dispatch messages"), ], *, name: Annotated[ str, Doc("Worker's name, if not provided it is evaluated from the class name"), ] = "", stopping_grace_period: Annotated[ float | None, Doc( "Grace period in seconds to wait for workers to stop running " "when this worker is shutdown. " "It defaults to the `FLUID_STOPPING_GRACE_PERIOD` " "environment variable or 10 seconds." ), ] = None, ) -> None: super().__init__(name=name, stopping_grace_period=stopping_grace_period) self.dispatcher: AsyncDispatcher[MessageType] = dispatcher ``` ### dispatcher ```python dispatcher = dispatcher ``` ### worker_state ```python worker_state ``` The running state of the worker ### worker_name ```python worker_name ``` The name of the worker ### num_workers ```python num_workers ``` The number of workers in this worker ### run ```python run() ``` Source code in `fluid/utils/worker.py` ```python async def run(self) -> None: while not self.is_stopping(): message = await self.get_message() if message is not None: await self.dispatcher.dispatch(message) ``` ### send ```python send(message) ``` Send a message into the worker Source code in `fluid/utils/worker.py` ```python def send(self, message: MessageType | None) -> None: """Send a message into the worker""" self._queue.put_nowait(message) ``` ### has_started ```python has_started() ``` Source code in `fluid/utils/worker.py` ```python def has_started(self) -> bool: return self._worker_state != WorkerState.INIT ``` ### is_running ```python is_running() ``` Source code in `fluid/utils/worker.py` ```python def is_running(self) -> bool: return self._worker_state == WorkerState.RUNNING ``` ### is_stopping ```python is_stopping() ``` Source code in `fluid/utils/worker.py` ```python def is_stopping(self) -> bool: return self._worker_state == WorkerState.STOPPING ``` ### is_stopped ```python is_stopped() ``` Source code in `fluid/utils/worker.py` ```python def is_stopped(self) -> bool: return self._worker_state in (WorkerState.STOPPED, WorkerState.FORCE_STOPPED) ``` ### gracefully_stop ```python gracefully_stop() ``` Try to gracefully stop the worker Source code in `fluid/utils/worker.py` ```python def gracefully_stop(self) -> None: """Try to gracefully stop the worker""" if self.is_running(): self._worker_state = WorkerState.STOPPING ``` ### after_shutdown ```python after_shutdown(reason, code) ``` Called after shutdown of worker By default it does nothing, but can be overriden to do something such as exit the process. Source code in `fluid/utils/worker.py` ```python def after_shutdown(self, reason: str, code: int) -> None: # noqa: B027 """Called after shutdown of worker By default it does nothing, but can be overriden to do something such as exit the process. """ ``` ### status ```python status() ``` Source code in `fluid/utils/worker.py` ```python async def status(self) -> dict: status = await super().status() status.update(queue_size=self.queue_size()) return status ``` ### on_startup ```python on_startup() ``` Called when the worker starts running Use this function to initialize other async resources connected with the worker Source code in `fluid/utils/worker.py` ```python async def on_startup(self) -> None: # noqa: B027 """Called when the worker starts running Use this function to initialize other async resources connected with the worker """ ``` ### on_shutdown ```python on_shutdown() ``` called after the worker stopped running Use this function to cleanup resources connected with the worker Source code in `fluid/utils/worker.py` ```python async def on_shutdown(self) -> None: # noqa: B027 """called after the worker stopped running Use this function to cleanup resources connected with the worker """ ``` ### startup ```python startup() ``` start the worker This method creates a task to run the worker. Source code in `fluid/utils/worker.py` ```python async def startup(self) -> None: """start the worker This method creates a task to run the worker. """ if self.has_started(): raise WorkerStartError( "worker %s already started: %s", self.worker_name, self._worker_state ) else: self._worker_task_runner = await WorkerTaskRunner.start(self) ``` ### shutdown ```python shutdown() ``` Shutdown a running worker and wait for it to stop This method will try to gracefully stop the worker and wait for it to stop. If the worker does not stop in the grace period, it will force shutdown by cancelling the task. Source code in `fluid/utils/worker.py` ```python async def shutdown(self) -> None: """Shutdown a running worker and wait for it to stop This method will try to gracefully stop the worker and wait for it to stop. If the worker does not stop in the grace period, it will force shutdown by cancelling the task. """ if self._worker_task_runner is not None: await self._worker_task_runner.shutdown() ``` ### wait_for_shutdown ```python wait_for_shutdown() ``` Wait for the worker to stop This method will wait for the worker to stop running, but doesn't try to gracefully stop it nor force shutdown. Source code in `fluid/utils/worker.py` ```python async def wait_for_shutdown(self) -> None: """Wait for the worker to stop This method will wait for the worker to stop running, but doesn't try to gracefully stop it nor force shutdown. """ if self._worker_task_runner is not None: await self._worker_task_runner.wait_for_shutdown() ``` ### workers ```python workers() ``` An iterator of workers in this worker Source code in `fluid/utils/worker.py` ```python def workers(self) -> Iterator[Worker]: """An iterator of workers in this worker""" yield self ``` ### get_message ```python get_message(timeout=0.5) ``` Get the next message from the queue Source code in `fluid/utils/worker.py` ```python async def get_message(self, timeout: float = 0.5) -> MessageType | None: """Get the next message from the queue""" try: async with asyncio.timeout(timeout): return await self._queue.get() except asyncio.TimeoutError: return None except (asyncio.CancelledError, RuntimeError): if not self.is_stopping(): raise return None ``` ### queue_size ```python queue_size() ``` Get the size of the queue Source code in `fluid/utils/worker.py` ```python def queue_size(self) -> int: """Get the size of the queue""" return self._queue.qsize() ``` ## fluid.utils.worker.Workers ```python Workers( *workers, name="", heartbeat=0.1, stopping_grace_period=None ) ``` Bases: `Worker` A Worker that owns and manages a collection of child workers. Child workers are registered with add_workers. When the `Workers` instance starts, its run loop starts each child worker and monitors their health — if any child stops unexpectedly the whole group is gracefully stopped. On shutdown all child workers are stopped concurrently. Workers that do not exit within `stopping_grace_period` seconds are force-cancelled. | PARAMETER | DESCRIPTION | | ----------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `*workers` | Workers to manage, they can also be added later via add_workers method **TYPE:** `Worker` **DEFAULT:** `()` | | `name` | Worker's name, if not provided it is evaluated from the class name **TYPE:** `str` **DEFAULT:** `''` | | `heartbeat` | The time to wait between each workers status check **TYPE:** \`float | | `stopping_grace_period` | Grace period in seconds to wait for workers to stop running when this worker is shutdown. It defaults to the FLUID_STOPPING_GRACE_PERIOD environment variable or 10 seconds. **TYPE:** \`float | Source code in `fluid/utils/worker.py` ```python def __init__( self, *workers: Annotated[ Worker, Doc( "Workers to manage, they can also be added later " "via `add_workers` method" ), ], name: Annotated[ str, Doc("Worker's name, if not provided it is evaluated from the class name"), ] = "", heartbeat: Annotated[ float | int, Doc("The time to wait between each workers status check"), ] = 0.1, stopping_grace_period: Annotated[ float | None, Doc( "Grace period in seconds to wait for workers to stop running " "when this worker is shutdown. " "It defaults to the `FLUID_STOPPING_GRACE_PERIOD` " "environment variable or 10 seconds." ), ] = None, ) -> None: super().__init__(name=name, stopping_grace_period=stopping_grace_period) self._heartbeat = heartbeat self._workers: list[Worker] = [] self.add_workers(*workers) ``` ### num_workers ```python num_workers ``` ### worker_state ```python worker_state ``` The running state of the worker ### worker_name ```python worker_name ``` The name of the worker ### add_workers ```python add_workers(*workers) ``` add workers to the workers They can be added while the worker is running. Source code in `fluid/utils/worker.py` ```python def add_workers(self, *workers: Worker) -> None: """add workers to the workers They can be added while the worker is running. """ for worker in workers: if worker not in self._workers: self._workers.append(worker) ``` ### workers ```python workers() ``` Source code in `fluid/utils/worker.py` ```python def workers(self) -> Iterator[Worker]: return iter(self._workers) ``` ### gracefully_stop ```python gracefully_stop() ``` Try to gracefully stop the workers and this worker Source code in `fluid/utils/worker.py` ```python def gracefully_stop(self) -> None: """Try to gracefully stop the workers and this worker""" super().gracefully_stop() for worker in self._workers: worker.gracefully_stop() ``` ### status ```python status() ``` Source code in `fluid/utils/worker.py` ```python async def status(self) -> dict: status_workers = await asyncio.gather( *[worker.status() for worker in self._workers], ) return { worker.worker_name: status for worker, status in zip(self._workers, status_workers, strict=False) } ``` ### run ```python run() ``` Source code in `fluid/utils/worker.py` ```python async def run(self) -> None: while self.is_running(): for worker in self._workers: if not worker.has_started(): await worker.startup() if not worker.is_running(): self.gracefully_stop() break await asyncio.sleep(self._heartbeat) await self._wait_for_workers() ``` ### has_started ```python has_started() ``` Source code in `fluid/utils/worker.py` ```python def has_started(self) -> bool: return self._worker_state != WorkerState.INIT ``` ### is_running ```python is_running() ``` Source code in `fluid/utils/worker.py` ```python def is_running(self) -> bool: return self._worker_state == WorkerState.RUNNING ``` ### is_stopping ```python is_stopping() ``` Source code in `fluid/utils/worker.py` ```python def is_stopping(self) -> bool: return self._worker_state == WorkerState.STOPPING ``` ### is_stopped ```python is_stopped() ``` Source code in `fluid/utils/worker.py` ```python def is_stopped(self) -> bool: return self._worker_state in (WorkerState.STOPPED, WorkerState.FORCE_STOPPED) ``` ### after_shutdown ```python after_shutdown(reason, code) ``` Called after shutdown of worker By default it does nothing, but can be overriden to do something such as exit the process. Source code in `fluid/utils/worker.py` ```python def after_shutdown(self, reason: str, code: int) -> None: # noqa: B027 """Called after shutdown of worker By default it does nothing, but can be overriden to do something such as exit the process. """ ``` ### on_startup ```python on_startup() ``` Called when the worker starts running Use this function to initialize other async resources connected with the worker Source code in `fluid/utils/worker.py` ```python async def on_startup(self) -> None: # noqa: B027 """Called when the worker starts running Use this function to initialize other async resources connected with the worker """ ``` ### on_shutdown ```python on_shutdown() ``` called after the worker stopped running Use this function to cleanup resources connected with the worker Source code in `fluid/utils/worker.py` ```python async def on_shutdown(self) -> None: # noqa: B027 """called after the worker stopped running Use this function to cleanup resources connected with the worker """ ``` ### startup ```python startup() ``` start the worker This method creates a task to run the worker. Source code in `fluid/utils/worker.py` ```python async def startup(self) -> None: """start the worker This method creates a task to run the worker. """ if self.has_started(): raise WorkerStartError( "worker %s already started: %s", self.worker_name, self._worker_state ) else: self._worker_task_runner = await WorkerTaskRunner.start(self) ``` ### shutdown ```python shutdown() ``` Shutdown a running worker and wait for it to stop This method will try to gracefully stop the worker and wait for it to stop. If the worker does not stop in the grace period, it will force shutdown by cancelling the task. Source code in `fluid/utils/worker.py` ```python async def shutdown(self) -> None: """Shutdown a running worker and wait for it to stop This method will try to gracefully stop the worker and wait for it to stop. If the worker does not stop in the grace period, it will force shutdown by cancelling the task. """ if self._worker_task_runner is not None: await self._worker_task_runner.shutdown() ``` ### wait_for_shutdown ```python wait_for_shutdown() ``` Wait for the worker to stop This method will wait for the worker to stop running, but doesn't try to gracefully stop it nor force shutdown. Source code in `fluid/utils/worker.py` ```python async def wait_for_shutdown(self) -> None: """Wait for the worker to stop This method will wait for the worker to stop running, but doesn't try to gracefully stop it nor force shutdown. """ if self._worker_task_runner is not None: await self._worker_task_runner.wait_for_shutdown() ``` # Background # Python task queues compared Python has a crowded field of task queues, and picking one is mostly about matching a library to how your service is shaped. This page lays out the landscape honestly, with real download data, and explains where `aio-fluid` does, and does not, make sense. If you are choosing for the first time, the short version: - **You want the biggest ecosystem and battle-tested defaults**: [Celery](https://docs.celeryq.dev/). - **You want something simple and Redis-only**: [RQ](https://python-rq.org/). - **You run an async (`asyncio`) service and your tasks never block the loop**: an async-native queue like [arq](https://arq-docs.helpmanual.io/) or [taskiq](https://taskiq-python.github.io/). - **You run an async service *and* some tasks are CPU-heavy** (parsing, pandas, model scoring, rendering) that would otherwise freeze your event loop: this is exactly what `aio-fluid` is built for. See [CPU bound tasks](https://fluid.quantmind.com/tutorials/tasks/#cpu-bound-tasks) and [K8s Jobs](https://fluid.quantmind.com/tutorials/task_k8s/index.md). - **You need ordered batch pipelines, backfills and data-aware scheduling**: you want a workflow orchestrator (Airflow, Dagster, Prefect, Luigi), not a task queue. See [Workflow orchestrators](#workflow-orchestrators-a-different-layer). ## Popularity Is Celery still the one everyone uses? Yes, decisively. It out-downloads the nearest real queue (RQ) by roughly an order of magnitude, and the async-native niche that `aio-fluid` competes in is comparatively small. That is both the opportunity (few libraries own "async-native + CPU-bound") and an honest reality check on the size of that audience. *Downloads = last 30 days on PyPI, via [pypistats.org](https://pypistats.org). Seeded figures; run `make stats` to refresh and re-rank from live data. Counts are inflated by CI, mirrors and Docker builds, so read them as orders of magnitude, not user counts.* | Library | Downloads / mo | Async-native | CPU work off the loop | Notes | | -------------------------------------------------------- | -------------- | ------------ | --------------------- | ---------------------------------------------------------------- | | [Celery](https://pypi.org/project/celery/) | 48.9M | partial | separate worker fleet | The incumbent: biggest ecosystem, broker-agnostic. | | [APScheduler](https://pypi.org/project/apscheduler/) | 37.1M | ✅ | n/a | A *scheduler*, not a distributed queue, listed for scale. | | [RQ](https://pypi.org/project/rq/) | 2.4M | no | separate worker | Simple, Redis-only; the common 'lite Celery'. | | [Dramatiq](https://pypi.org/project/dramatiq/) | 270k | no | separate worker | Ergonomic Celery alternative. | | [Huey](https://pypi.org/project/huey/) | 117k | no | separate worker | Lightweight, very few dependencies. | | [taskiq](https://pypi.org/project/taskiq/) | 34k | ✅ | no | Async-native, pluggable brokers; typed parameters. | | [arq](https://pypi.org/project/arq/) | n/a | ✅ | no | Async-native, Redis; count pending first `make stats`. | | [SAQ](https://pypi.org/project/saq/) | n/a | ✅ | no | Async-native, Redis; count pending first `make stats`. | | [Procrastinate](https://pypi.org/project/procrastinate/) | n/a | ✅ | no | Async-native, Postgres-backed; count pending first `make stats`. | | [aio-fluid](https://pypi.org/project/aio-fluid/) | 4k | ✅ | subprocess / k8s Job | This library: CPU-bound work is a first-class task type. | ## The libraries **Celery** is the default answer for a reason: the largest ecosystem, broker flexibility (RabbitMQ, Redis, SQS…), and years of production hardening. It predates `asyncio`, though, and its standard answer to CPU-bound work is to run a separate worker fleet. If you need breadth and maturity, reach for Celery. **RQ** is the "simple Celery": Redis-only, small API, easy to reason about. Synchronous by design, so it is great for straightforward background jobs but not aimed at async services. **Dramatiq** is an ergonomic, well-designed Celery alternative with a cleaner API and retries/middleware built in. Still a worker-per-process model rather than async-native. **Huey** is deliberately tiny and dependency-light, a good fit for small projects that want a queue without operational weight. **arq**, **taskiq**, **SAQ** and **Procrastinate** are the async-native cohort. They run `async def` tasks on the event loop and are excellent for IO-bound work. By design they assume tasks do not block the loop, so CPU-bound work is out of scope. That is the same limitation that motivated `aio-fluid`. **APScheduler** is included only for scale: it is a *scheduler* (fire a job on an interval or cron), not a broker-backed distributed queue, so it solves a different problem. ### Workflow orchestrators, a different layer **Luigi**, **Airflow**, **Prefect** and **Dagster** come up in the same searches, but they are workflow orchestrators rather than task queues. You declare a graph of batch steps and the framework owns ordering, backfills and data-aware scheduling. They do execute your code, and several of them delegate that execution to a queue (Airflow's `CeleryExecutor`, Dagster's Celery executor), which is the tell: they sit a layer above the field on this page, and can run on top of it. `aio-fluid` covers the simple end of that layer. A task queues the next one with TaskRun.queue, and each run records the run it came from, so multi-step pipelines are ordinary Python with no extra machinery and can be traced back to whatever started them. See [Chaining tasks](https://fluid.quantmind.com/tutorials/tasks/#chaining-tasks). If your pipeline is a handful of steps triggered on a schedule, that is very likely all you need. Reach for a real orchestrator when you need backfills over historical partitions, resuming a run from the step that failed, or lineage across hundreds of datasets. `aio-fluid` models none of those, and pretending otherwise would waste your time. ## Where aio-fluid fits `aio-fluid` is an async-native queue like arq/taskiq, but it treats CPU-bound work as a first-class task type instead of assuming it away. Mark a task [`cpu_bound=True`](https://fluid.quantmind.com/tutorials/tasks/#cpu-bound-tasks) and it runs in a **fresh subprocess** so heavy work never blocks the event loop; when the consumer runs inside Kubernetes, the *same task* dispatches as a [Kubernetes Job](https://fluid.quantmind.com/tutorials/task_k8s/index.md) instead, with no code change and no parallel worker deployment to maintain. So the honest positioning is not "Celery killer." It is this: **if you run an async service and have ever watched one CPU-heavy task freeze the whole thing, `aio-fluid` is built for that specific pain.** If you have no CPU-bound work, a lighter async queue will serve you just as well; if you need Celery's ecosystem, use Celery. ## Caveats on the numbers PyPI download counts are a rough popularity proxy, not a user count. They are inflated by CI/CD pipelines, Docker image builds, and mirrors, and libraries that are transitive dependencies of popular packages (Celery, APScheduler) are inflated the most. Treat every figure above as an order of magnitude. A value of `n/a` means the count has not been fetched yet. ## Refreshing this table The table above is regenerated from live data by `scripts/task_queue_stats.py`: ```bash make stats ``` It fetches last-30-days downloads from the [pypistats.org](https://pypistats.org) API for the curated library list, re-ranks the table, and rewrites the region between the `STATS` marker comments in this page. Run it before a release, or whenever the numbers look stale. # Release Notes This page is the source of truth for aio-fluid release notes. Each section below maps to a tagged release on [GitHub](https://github.com/quantmind/aio-fluid/releases). When a new tag is pushed, the matching section is extracted by `.github/workflows/release.yml` and published as the GitHub Release body. ## v2.5.0 Task runs can queue other task runs and the chain is recorded on every run, CPU bound tasks behave the same way in a subprocess as on the event loop, and Kubernetes Jobs inherit the task timeout. The documentation gained a settings reference, a recipes cheat sheet and a page on pointing coding agents at the library. - A task run can queue another task run with [TaskRun.queue](https://fluid.quantmind.com/reference/task_run/), which records the queueing run in `from_run_id` and carries the `root_run_id` of the first run in the chain, so any run can be traced back to the one that started it. ([#110](https://github.com/quantmind/aio-fluid/pull/110)) - CPU bound tasks now start the async event dispatcher in the subprocess that runs them, so lifecycle events reach the handlers and plugins there as well, and a consumer with CPU bound tasks that is not started from [TaskManagerCLI](https://fluid.quantmind.com/reference/task_cli/) raises the new `CpuBoundEntryPointError` on startup rather than failing when the first such task runs. - A Kubernetes Job created for a CPU bound task sets `active_deadline_seconds` from the task `timeout_seconds`, so the cluster terminates a Job that overruns. ([#108](https://github.com/quantmind/aio-fluid/pull/108)) - New [Settings](https://fluid.quantmind.com/reference/settings/) reference page covering the environment variables that configure the task consumer, broker, database and HTTP client, including the prefix rules and the fields that keep an unprefixed name. - Fixed the [Task Broker](https://fluid.quantmind.com/tutorials/task_broker/) tutorial, which imported a name that does not exist, quoted the wrong default Redis port, and listed six outdated abstract methods instead of the sixteen a broker has to implement. - Admonition blocks are rendered as admonitions instead of literal text, which also fixes the note on retry delays in the [Task Retry](https://fluid.quantmind.com/reference/task_retry/) reference. - New [recipes](https://fluid.quantmind.com/recipes/) cheat sheet, a page on [using the docs with AI agents](https://fluid.quantmind.com/ai-agents/) and an `AGENTS.md` for contributors. ([#111](https://github.com/quantmind/aio-fluid/pull/111)) - New tutorials on [task dependencies](https://fluid.quantmind.com/tutorials/task_deps/), [choosing a task manager](https://fluid.quantmind.com/tutorials/task_managers/) and [extending the FastAPI app](https://fluid.quantmind.com/tutorials/task_fastapi/). ([#109](https://github.com/quantmind/aio-fluid/pull/109), [#111](https://github.com/quantmind/aio-fluid/pull/111)) - New [comparison](https://fluid.quantmind.com/comparison/) page placing the library next to Celery, RQ, arq and taskiq, with download numbers refreshed by a scheduled workflow, and a landing page rewritten around CPU bound work. ([#104](https://github.com/quantmind/aio-fluid/pull/104), [#105](https://github.com/quantmind/aio-fluid/pull/105), [#107](https://github.com/quantmind/aio-fluid/pull/107)) ## v2.4.3 Fixes two task queue issues: pydantic secret params were masked when a task run was serialized to the queue, and a params validation error on the consumer side crashed the worker. - Secret params now survive the round-trip through the task queue and the cpu-bound subprocess. Task runs are serialized for the queue with secret values revealed via the new `params_dump` helper; all other dumps (logs, endpoints) keep secrets masked. ([#103](https://github.com/quantmind/aio-fluid/pull/103)) - A task run consumed from the queue with invalid params no longer kills the consumer worker. The broker raises the new `TaskParamsError` carrying the task run, and the consumer logs the error and marks the run as failed. ([#103](https://github.com/quantmind/aio-fluid/pull/103)) ## v2.4.2 Tasks can now be tagged at registration time. - The task registration methods accept an optional `tags` argument. When provided, the extra tags are merged into each task's own tags as it is registered — applied across `register_task`, `register_from_module`, and `register_from_dict`. ([#102](https://github.com/quantmind/aio-fluid/pull/102)) - Bumped `python-json-logger` to `>= 4.1.0` and switched JSON logging to the new `pythonjsonlogger.json` formatter path (the old `pythonjsonlogger.jsonlogger` module is deprecated). ([#102](https://github.com/quantmind/aio-fluid/pull/102)) ## v2.4.1 Task history can now be filtered by task tags. - Added a `tags` field to task history queries. Runs match when their task carries at least one of the given tags, resolved against the live registry. ([#101](https://github.com/quantmind/aio-fluid/pull/101)) ## v2.4.0 Lazy settings via pydantic-settings, JSONB params filtering for task history, and customisable route prefixes. - Settings are now lazy — resolved on first access instead of at import time. Env vars use a `FLUID_` prefix by default; legacy unprefixed names are kept as aliases. ([#100](https://github.com/quantmind/aio-fluid/pull/100)) - The task database plugin accepts a `route_prefix` parameter for customising history route URLs and replaces `with_task_history_router()` with a `register_routes()` method. ([#100](https://github.com/quantmind/aio-fluid/pull/100)) - Task history queries support filtering by run params via a new `params` field (renamed from `HistoryQuery` to `TaskHistoryQuery`). ([#99](https://github.com/quantmind/aio-fluid/pull/99)) - **Database migration required:** the `params` column is now `JSONB` with a GIN index. See the example [migration](https://github.com/quantmind/aio-fluid/blob/main/examples/tasks/migrations/versions/d941c11ca25a_jsonb.py) for the schema changes. - Removed `get_logger` from `fluid.utils.log`. Task loggers are now obtained directly via `logging.getLogger(module)`. ## v2.3.1 **v2.3.0 is broken — do not use it.** Fixes a regression in v2.3.0 where the `httpx2` dependency was pinned to `>=2.2.0`, which fails on Python 3.14 builds missing the `_zstd` C extension. Pins `httpx2` to `>=2.0.0, <2.1.0` and switches all `httpx` imports to `httpx2` for correct namespace resolution. - `httpx2` is now pinned to `>=2.0.0, <2.1.0` — versions 2.1.0+ require the `compression.zstd` stdlib module which is not available in all Python 3.14 builds. ([#98](https://github.com/quantmind/aio-fluid/pull/98)) - All `import httpx` statements replaced with `import httpx2 as httpx` (or `from httpx2 import ...`) to ensure correct namespace resolution regardless of `httpx2` version. - Added test coverage for [HttpxClient](https://fluid.quantmind.com/reference/http_client/#fluid.utils.http_client.HttpxClient) and [HttpxResponse](https://fluid.quantmind.com/reference/http_client/#fluid.utils.http_client.HttpxResponse). ## v2.3.0 Moves development and documentation dependencies from optional-dependencies to [dependency groups](https://peps.python.org/pep-0735/), switches to [httpx2](https://pypi.org/project/httpx2/) for HTTP client support, and removes the `inflection` dependency. - `dev` and `docs` dependencies are now declared under `[dependency-groups]` instead of `[project.optional-dependencies]`. Installed via `uv sync --all-groups`. - The `http` extra now uses `httpx2` instead of `httpx`. `httpx2` provides the same `httpx` module so no code changes are required. ([#97](https://github.com/quantmind/aio-fluid/pull/97)) - The `inflection` dependency has been removed. ([#96](https://github.com/quantmind/aio-fluid/pull/96)) ## v2.2.6 Adds tag filtering for task listings and fixes a race in the task database plugin. - Task listings can be filtered by tag: the `GET /tasks` endpoint and the `ls` command of the [task CLI](https://fluid.quantmind.com/reference/task_cli/) accept a repeatable `tags` option that returns only tasks carrying at least one of the given tags, and `TaskInfo` now reports each task's tags. ([#94](https://github.com/quantmind/aio-fluid/pull/94)) - The [task database plugin](https://fluid.quantmind.com/reference/task_plugin/#fluid.scheduler.db.TaskDbPlugin) now serialises its per-run lifecycle writes with a dedicated task-run lock. `CrudDB.db_upsert` is not atomic — it issues an `UPDATE` and only `INSERT`s when nothing matched — so when the scheduler wrote the `queued` row and a consumer wrote the `running` row a few milliseconds later, the consumer's `UPDATE` could miss the not-yet-committed `INSERT`, fall through to its own `INSERT` and violate the task-runs primary key. Holding the lock around the upsert removes the race. ([#95](https://github.com/quantmind/aio-fluid/pull/95)) - [TaskRun.lock](https://fluid.quantmind.com/reference/task_run/#fluid.scheduler.TaskRun.lock) accepts an optional `name` to acquire a named sub-lock for the task run, and `timeout` now defaults to `None`. ([#95](https://github.com/quantmind/aio-fluid/pull/95)) ## v2.2.5 - The [task decorator](https://fluid.quantmind.com/reference/task/#fluid.scheduler.task) accepts an `env` mapping of extra environment variables, injected into the subprocess for CPU-bound tasks and forwarded to the container for tasks dispatched as Kubernetes Jobs. ([#90](https://github.com/quantmind/aio-fluid/pull/90)) ## v2.2.4 Bug-fix release for the task scheduler. - Fix task interruption handling. ([#89](https://github.com/quantmind/aio-fluid/pull/89)) - Fix stale concurrent tasks not being released. ([#88](https://github.com/quantmind/aio-fluid/pull/88)) - Fix task abort behaviour. ([#87](https://github.com/quantmind/aio-fluid/pull/87)) - Patch `pydanclick` to work with `StrEnum`. ([#86](https://github.com/quantmind/aio-fluid/pull/86))