TaskIQ FastAPI Integration

0.5.0 verified Fri May 01 auth: no python

TaskIQ-FastAPI provides integration between FastAPI and TaskIQ, allowing you to run background tasks with access to FastAPI's context (e.g., dependency injection, request state). Current version is 0.5.0, compatible with Python >=3.10. Release cadence is irregular, driven by TaskIQ ecosystem updates.

pip install taskiq-fastapi

Common errors

error RuntimeError: No request context available ↓

cause Trying to access FastAPI request or dependency injection in a task without initializing the context via `InitFastAPI().init()`.

fix

Add a startup event or use @taskiq_app.on_app_init to call InitFastAPI().init() before any task runs.

error ImportError: cannot import name 'InitFastAPI' from 'taskiq_fastapi' ↓

cause Using an older version where the import path was different (e.g., `from taskiq_fastapi.broker import InitFastAPI`).

fix

Upgrade to the latest version (0.5.0) and use from taskiq_fastapi import InitFastAPI.

error AttributeError: module 'taskiq_fastapi' has no attribute 'TaskiqFastAPI' ↓

cause Misspelled class name or using an outdated version that used a different class name (e.g., `FastApiApp`).

fix

Use from taskiq_fastapi import TaskiqFastAPI and ensure you have version 0.5.0 compatible code.

Warnings

breaking In version 0.5.0, the import path changed from `taskiq_fastapi.broker` to `taskiq_fastapi` directly. Old imports will break. ↓

fix Use `from taskiq_fastapi import InitFastAPI, TaskiqFastAPI` instead of `from taskiq_fastapi.broker import ...`.

gotcha The `InitFastAPI` must be called during the app lifecycle (usually in a startup event or via `on_app_init` decorator) to inject FastAPI dependencies into tasks. Not calling it results in `RuntimeError: No request context available` when accessing FastAPI dependencies. ↓

fix Ensure `InitFastAPI().init()` is called at startup (e.g., in a lifespan event or by decorating a function with `@taskiq_app.on_app_init`).

gotcha Tasks defined before the broker is attached to `TaskiqFastAPI` may not have access to context. Define tasks after calling `taskiq_app = TaskiqFastAPI(app, broker)`. ↓

fix Define task functions after the `TaskiqFastAPI` instance is created, or ensure the broker is correctly wired before task definitions.

deprecated The `TaskiqFastAPI` class previously accepted a `middleware` argument; this is deprecated in favor of using `ResultBackendMiddleware` separately. ↓

fix Remove `middleware` parameter from `TaskiqFastAPI` and add `ResultBackendMiddleware` to the app if needed.

Imports

InitFastAPI
```
from taskiq_fastapi import InitFastAPI
```
Entry point for integrating TaskIQ with FastAPI.

TaskiqFastAPI

wrong

from taskiq_fastapi import TaskiqFastAPI (if misspelled)

correct

from taskiq_fastapi import TaskiqFastAPI

Class for managing TaskIQ lifecycle in FastAPI.

ResultBackendMiddleware
```
from taskiq_fastapi import ResultBackendMiddleware
```
Middleware to expose result backend via request state.

Quickstart

Minimal setup integrating TaskIQ with FastAPI using InMemoryBroker. Note: In production, use a persistent broker like Redis or RabbitMQ.

from fastapi import FastAPI
from taskiq import TaskIQ
from taskiq_fastapi import InitFastAPI, TaskiqFastAPI

app = FastAPI()

# Initialize TaskIQ broker (e.g., InMemoryBroker)
from taskiq import InMemoryBroker
broker = InMemoryBroker()

# Attach FastAPI integration
taskiq_app = TaskiqFastAPI(app, broker)

# Register an initialization callback (optional, for context access)
@taskiq_app.on_app_init
def on_startup():
    InitFastAPI().init()

@app.get("/")
async def root():
    return {"message": "Hello World"}

# Task must be defined after broker is available
@broker.task
def my_task(param: str):
    print(f"Processing {param}")
    return param