Unfazed Taskiq

taskiq wrapper with unfazed.

Installation

pip install unfazed-taskiq

Quick Start

This guide will help you get started with Unfazed Taskiq in just a few minutes.

1. Create new db table

CREATE TABLE `unfazed_taskiq_periodic_task` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `created_at` datetime NOT NULL,
  `updated_at` datetime NOT NULL,
  `description` text NOT NULL,
  `schedule_alias` varchar(255) NOT NULL,
  `task_name` varchar(255) NOT NULL,
  `task_args` text NOT NULL,
  `task_kwargs` text NOT NULL,
  `labels` text NOT NULL,
  `cron` varchar(255) DEFAULT NULL,
  `time` datetime(6) DEFAULT NULL,
  `last_run_at` datetime NOT NULL,
  `total_run_count` int(11) NOT NULL,
  `enabled` tinyint(1) NOT NULL,
  `schedule_id` varchar(255) NOT NULL,
  `name` varchar(255) NOT NULL DEFAULT '',
  PRIMARY KEY (`id`),
  KEY `idx_schedule_id` (`schedule_id`)
) ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8mb4;

2. Configure Settings

Add Taskiq configuration to your Unfazed settings file:

AMQP_URL = os.getenv("AMQP_URL", DEFAULT_AMQP_URL)
# entry/settings.py
UNFAZED_TASKIQ_SETTINGS = {
    "DEFAULT_TASKIQ_NAME": "default",
    "TASKIQ_CONFIG": {
        "default": {
            "BROKER": {
                "BACKEND": "taskiq_aio_pika.AioPikaBroker",
                "OPTIONS": {
                    "url": AMQP_URL,
                    "exchange_name": "unfazed-taskiq",
                    "queue_name": "unfazed-taskiq",
                },
            },
        },
    },
}

# Add Taskiq lifespan to your Unfazed settings
UNFAZED_SETTINGS = {
    "LIFESPAN": ["unfazed_taskiq.lifespan.TaskiqLifeSpan"],
    # ... your other settings
}

3. Create Tasks

Define your tasks in your app's tasks.py file:

# app/tasks.py
from unfazed_taskiq.decorators import task

@task
async def add_numbers(a: int, b: int) -> int:
    """Simple addition task."""
    return a + b

4. Start Worker

# auto discover all async task
uv run taskiq unfazed-worker unfazed_taskiq.agent:broker -fsd

# auto discover all async task by task file
uv run taskiq unfazed-worker unfazed_taskiq.agent:broker -fsd -tp backend/spider/tasks.py

5. Execute Tasks

from xxx.task import add_numbers
async def your_service():
    # Execute task immediately
    result = await add_numbers.kiq(10, 20)
    print(f"Task result: {result}")

How to use Scheduler

1. Configure Settings

Add Taskiq configuration to your Unfazed settings file:

from unfazed_taskiq.contrib.scheduler.sources import TortoiseScheduleSource

AMQP_URL = os.getenv("AMQP_URL", DEFAULT_AMQP_URL)
unfazedtaskiq_source = TortoiseScheduleSource(schedule_alias="unfazedtaskiq")
unfazedtaskiq_v2_source = TortoiseScheduleSource(schedule_alias="unfazedtaskiq_v2")

# entry/settings.py
UNFAZED_TASKIQ_SETTINGS = {
    "DEFAULT_TASKIQ_NAME": "default",
    "TASKIQ_CONFIG": {
        "default": {
            "BROKER": {
                "BACKEND": "taskiq_aio_pika.AioPikaBroker",
                "OPTIONS": {
                    "url": AMQP_URL,
                    "exchange_name": "unfazedtaskiq",
                    "queue_name": "unfazedtaskiq",
                },
                "MIDDLEWARES": { # options: if you want use sentry collect error
                    "unfazed_taskiq.middleware.UnfazedTaskiqExceptionMiddleware"
                }
            },
            "SCHEDULER": {
                "SOURCES": [unfazedtaskiq_source],
                "BACKEND": "taskiq.TaskiqScheduler",
            },
        },
        "taskiq_task": {
            "BROKER": {
                "BACKEND": "taskiq_aio_pika.AioPikaBroker",
                "OPTIONS": {
                    "url": AMQP_URL,
                    "exchange_name": "unfazedtaskiq_v2",
                    "queue_name": "unfazedtaskiq_v2",
                },
                "MIDDLEWARES": { # options: if you want use sentry collect error
                    "unfazed_taskiq.middleware.UnfazedTaskiqExceptionMiddleware"
                }
            },
            "SCHEDULER": {
                "SOURCES": [unfazedtaskiq_v2_source],
                "BACKEND": "taskiq.TaskiqScheduler",
            },
        },
    },
}

# Add Taskiq lifespan to your Unfazed settings
UNFAZED_SETTINGS = {
    "LIFESPAN": ["unfazed_taskiq.lifespan.TaskiqLifeSpan"],
    # ... your other settings
}

2. Create Tasks

Define your tasks in your app's tasks.py file:

# app/tasks.py
from unfazed_taskiq.decorators import task

@task
async def send_email(email: str, content: str) -> None:
    """send some msg for some body."""
    ...

3. Schedule Tasks

You can schedule tasks using the database scheduler:

# Create scheduled tasks in your database
from unfazed_taskiq.contrib.scheduler.models import PeriodicTask
import json

# Create a periodic task that runs every minute
# Also, can use Unfazed-admin UI config
await PeriodicTask.create(
    task_name="app.tasks.send_email",
    task_args=json.dumps(["test@gmail.com", "test content"]),
    task_kwargs=json.dumps({}),
    cron="*/1 * * * *",  # Every minute
    description="Add two numbers every minute",
)

4. Start Scheduler

Execute tasks from your application code:

• start all scheduler

  uv run taskiq unfazed-scheduler unfazed_taskiq.agent:scheduler
  

• start scheduler with alias_name (eg: default / taskiq_task)

  uv run taskiq unfazed-scheduler unfazed_taskiq.agent:scheduler --alias-name taskiq_task
  

5. Start Workers

Start the Taskiq worker to process tasks:

uv run taskiq unfazed-worker unfazed_taskiq.agent:broker -fsd -tp app/tasks.py

Result backend

Result backend is where this library stores Taskiq task results (MySQL/TiDB via MySQLResultBackend): each task gets a row with status, times, args/kwargs, and return value; the full serialized payload is kept in the result column.

Unfazed Admin: add unfazed_taskiq.contrib.result to INSTALLED_APPS. It registers TaskiqResultAdmin with TaskiqResultSerializer, so you can browse and open task runs in the admin UI (list + detail, including a readable return_value field). The raw binary result field is not exposed as JSON in admin APIs.

1. How to enable

Add the app under UNFAZED_SETTINGS["INSTALLED_APPS"], and add the middleware + result backend under TASKIQ_CONFIG for your Taskiq instance name (same place as broker/scheduler):

UNFAZED_SETTINGS = {
    # ...
    "INSTALLED_APPS": [
        # ...your apps...
        "unfazed_taskiq.contrib.result",
    ],
}

UNFAZED_TASKIQ_SETTINGS = {
    "TASKIQ_CONFIG": {
        "your_taskiq_name": {  # e.g. DEFAULT_TASKIQ_NAME
            "BROKER": {
                # ...broker BACKEND / OPTIONS...
                "MIDDLEWARES": [
                    "unfazed_taskiq.contrib.result.middleware.TaskiqResultPreSendMiddleware",
                ],
            },
            "RESULT": {
                "BACKEND": "unfazed_taskiq.contrib.result.mysql.MySQLResultBackend",
                "OPTIONS": {},
            },
            # ...SCHEDULER, etc...
        },
    },
}

2. Create table

CREATE TABLE `taskiq_result` (
    `id` INT NOT NULL PRIMARY KEY AUTO_INCREMENT,
    `task_id` VARCHAR(255) NOT NULL UNIQUE
        COMMENT 'Task unique identifier',
    `status` SMALLINT NOT NULL
        COMMENT 'TaskStatus: 1=STARTED, 2=SUCCESS, 3=FAILURE',
    `result` BLOB NULL
        COMMENT 'Serialized TaskiqResult from serializer.dumpb',
    `return_value` JSON NULL
        COMMENT 'Snapshot via encode_for_json_field; full value in result blob',
    `date_done` BIGINT NULL
        COMMENT 'Timestamp when task completed',
    `date_created` BIGINT NULL
        COMMENT 'Timestamp when task was enqueued',
    `task_name` VARCHAR(255) NULL
        COMMENT 'Task definition name',
    `schedule_id` VARCHAR(255) NULL
        COMMENT 'Schedule id for periodic tasks (from labels)',
    `task_args` JSON NULL
        COMMENT 'JSON array or __taskiq_json_str_fallback__ str(list)',
    `task_kwargs` JSON NULL
        COMMENT 'JSON object or __taskiq_json_str_fallback__ str(dict)',
    `traceback` TEXT NULL
        COMMENT 'Traceback when task failed',
    INDEX `idx_date_done` (`date_done`),
    INDEX `idx_task_name_date_done` (`task_name`, `date_done`),
    INDEX `idx_schedule_id_date_done` (`schedule_id`, `date_done`),
    INDEX `idx_status` (`status`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

3. How to get the result

task = await add_numbers.kiq(10, 20)
value = await task.wait_result()  # waits until done, then returns the value

📖 更多文档

pls read taskiq document

📄 许可证

本项目基于 MIT 许可证开源。