An async task queue that offloads CPU-bound work to subprocesses or Kubernetes Jobs
Documentation: fluid.quantmind.com
Source: 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.
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+ thek8sextra). Your task code is identical in both. See K8s Jobs. - Async-native and typed. Tasks are plain
async deffunctions; parameters are pydantic models, validated on the way in. - Dependency injection. A database manager, an HTTP client or any other resource is grouped
into a single typed
depsobject every task run can reach, with startup and shutdown handled by the task manager. See Task Dependencies. - The scheduling you expect.
every(timedelta(...))andcrontab(...), per-taskmax_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-historyAPI on top of it. See 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.
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.
- Async Postgres CRUD: a typed CRUD layer over
asyncpgand SQLAlchemy, with pagination and schema migrations. See Database. - Event dispatchers: sync and async
Dispatchertypes for decoupling event sources from handlers. See Dispatchers. - HTTP client helpers: a unified async client wrapping
httpxandaiohttp. See HTTP Client. - CLI tooling: ready-made
click/richcommand-line interfaces for task managers and databases.
Installation
This is a python package you can install via pip:
pip install aio-fluid
To install all the dependencies:
pip install aio-fluid[cli, db, http, log, k8s]
this includes the following extra dependencies:
clifor the command line interface using click and richdbfor database support with asyncpg and sqlalchemyhttpfor http client support with httpx and aiohttplogfor JSON logging support with python-json-loggerk8sfor Kubernetes support for CPU bound tasks
AI agents
The documentation is published in a form coding agents can consume directly, following the llms.txt convention:
- 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: the whole documentation in a single file, API reference included.
- 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. If you are pointing an agent at this repository to contribute, read AGENTS.md.
Development
You can run the examples via
uv run python -m examples
We use uv 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 file for details.
1 maintainer
Metadata
1 maintainer
Release files for aio-fluid 2.11.1
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | Provenance | |
|---|---|---|---|---|
| aio_fluid-2.11.1.tar.gz | 723.9 kB | Without provenance | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| aio_fluid-2.11.1-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 806.1 kB
Release files / aio_fluid-2.11.1.tar.gz
| Download URL | aio_fluid-2.11.1.tar.gz |
|---|---|
| Size | 723.9 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
19d7eaf8a49380ec0a0ad140847e367b46ec043dff7478c21966cf843a8181de
|
|
BLAKE2b-256 checksum How to use checksums |
cefd0d5c3e4b96b4dc55276879d0b7afe17542230e2dc903d4c56b57c3593336
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
uv/0.13.1 {"installer":{"name":"uv","version":"0.13.1","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"Ubuntu","version":"24.04","id":"noble","libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":true}
|
Release files / aio_fluid-2.11.1-py3-none-any.whl
| Download URL | aio_fluid-2.11.1-py3-none-any.whl |
|---|---|
| Size | 82.2 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
4a47c45f67e86e43659513e39f0cb28289cddcef0e09423dd6140f98c667c7ba
|
|
BLAKE2b-256 checksum How to use checksums |
7b9b16aebc1e930dd5faad3a93ff49c8fbe0cf28e120c9df8f4d813f1bc711d8
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
uv/0.13.1 {"installer":{"name":"uv","version":"0.13.1","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"Ubuntu","version":"24.04","id":"noble","libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":true}
|
Security for aio-fluid 2.11.1
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
No provenance found
None of the files in this release were uploaded with provenance metadata.
A note on package security
PyPI is a repository for distributing packages, not a service for auditing them. While third-party security researchers and our staff do their best to detect and remove malicious packages, PyPI cannot guarantee the quality, security or suitability of any given package.
To learn more about protecting yourself from malicious actors, visit our package security documentation.
