taskiq: Open Source Project & Work Management

What is taskiq?

What it is

Taskiq is an asynchronous distributed task queue for Python. It lives in the Python ecosystem and provides a way to declare functions as tasks, send those tasks to a broker, and run them in worker processes. The project takes inspiration from Celery and Dramatiq, but it is designed around asyncio and supports both sync and async functions.

The problem it solves is moving work out of web applications and scripts into separate workers. A web application can call a task, the message goes through a queue backend such as NATS, Redis, RabbitMQ, or Kafka, and a worker executes the function later. This lets developers use Python functions and type hints while running work outside the web application or script.

Key capabilities

Sends and runs both sync and async Python functions as tasks.
Uses broker objects to connect application code to workers through distributed queues.
Provides broker integrations for NATS, Redis, RabbitMQ, and Kafka.
Declares tasks by adding a decorator to a Python function.
Sends tasks with the kiq method, which places a message on the broker for a worker to execute.
Supports worker startup, automatic discovery of tasks.py files with --fs-discover, and hot reload with --reload.
Uses type hints and PEP-612 to provide autosuggestions for task arguments.

Who uses it and how

Python developers use it to run background jobs from asyncio applications without blocking the main event loop.
FastAPI and AioHTTP users use taskiq integrations to reuse dependencies from web applications inside tasks.
Teams use it when tasks need to run in separate worker processes while communicating through a queue backend.
Developers use it to organize work in modules called tasks.py and let workers discover those modules automatically.
Local developers use it with reload extra to restart workers when code changes during development.

Getting started

Install taskiq with pip or directly from git, then create a broker for a queue backend such as NATS, Redis, RabbitMQ, or Kafka. Run workers with taskiq worker path.to.the.module:broker and add --fs-discover or --reload when needed.

When to use it — and when not to

Taskiq fits Python projects that need async task execution and already operate a queue backend such as NATS, Redis, RabbitMQ, or Kafka. A self-hoster must run both the broker backend and worker processes, and the README does not mention a hosted option. The repository has open issues, so users should verify current behavior.

project readme (upstream, from github) — read inline

Documentation: https://taskiq-python.github.io/

What is taskiq?

Taskiq is an asynchronous distributed task queue for python. This project takes inspiration from big projects such as Celery and Dramatiq. But taskiq can send and run both the sync and async functions, has integration with popular async frameworks, such as FastAPI and AioHTTP.

Also, we use PEP-612 to provide the best autosuggestions possible. All code is type-hinted.

Installation

This project can be installed using pip:

pip install taskiq

Or it can be installed directly from git:

pip install git+https://github.com/taskiq-python/taskiq

Usage

At first you need to create a broker. Broker is an object that can communicate to workers using distributed queues.

We have different brokers for different queue backends. For example, we have a broker for NATS, Redis, RabbitMQ, Kafka and even more. Choose the one that fits you and create an instance.

from taskiq_nats import JetStreamBroker

broker = JetStreamBroker("nats://localhost:4222", queue="my_queue")

Declaring tasks is as easy as declaring a function. Just add a decorator to your function and you are ready to go.

import asyncio

from taskiq_nats import JetStreamBroker

broker = JetStreamBroker("nats://localhost:4222", queue="my_queue2")


@broker.task
async def my_task(a: int, b: int) -> None:
    print("AB", a + b)


async def main():
    await broker.startup()

    await my_task.kiq(1, 2)

    await broker.shutdown()


if __name__ == "__main__":
    asyncio.run(main())

The message is going to be sent to the broker and then to the worker. The worker will execute the function. To start worker processes, just run the following command:

taskiq worker path.to.the.module:broker

Where path.to.the.module is the path to the module where the broker is defined and broker is the name of the broker variable.

If you have tasks in different modules, you can ask taskiq to automatically import them by passing the --fs-discover flag:

taskiq worker path.to.the.module:broker --fs-discover

It will import all modules called tasks.py in the current directory and all subdirectories.

Also, we support hot reload for workers. To enable it, just pass the --reload flag. It will reload the worker when the code changes (To use it, install taskiq with reload extra. E.g pip install taskiq[reload]).

Also, we have cool integrations with popular async frameworks. For example, we have an integration with FastAPI or AioHTTP. You can use it to reuse dependencies from your web app in your tasks.

Read about all features in our documentation: https://taskiq-python.github.io/

Local development

Linting

We use pre-commit to do linting locally.

After cloning this project, please install pre-commit. It helps fix files before committing changes.

pre-commit install

Testing

Pytest can run without any additional actions or options.

pytest

Docs

To run docs locally, you need to install yarn.

First, you need to install dependencies.

yarn install

After that you can set up a docs server by running:

yarn docs:dev

Frequently asked questions

Is taskiq free to use?

taskiq is open source under the MIT licence. There is no licence fee and no seat count — you can self-host it or, where the project offers one, pay a vendor for a managed version instead.

What does taskiq do?

Distributed task queue with full async support

What is taskiq written in?

taskiq is primarily written in Python. Its source is publicly available at https://github.com/taskiq-python/taskiq, and it has 2,335 GitHub stars.