Перейти к содержанию

Pydantic

Существует встроенный в Worker плагин (A)syncPydanticWrapperPlugin, который автоматически преобразует входные параметры задачи в Pydantic-модели. Он делает работу с типизированными структурами данных естественной и прозрачной: модель собирается прямо перед вызовом функции задачи, а разработчик получает уже готовый валидированный объект.

Плагин работает без дополнительной настройки. Любая задача, принимающая параметр-модель, будет поддерживать автоматическую сборку.

Пример использования

import pydantic
from qtasks.asyncio import QueueTasks

app = QueueTasks()

class Item(pydantic.BaseModel):
    name: str
    value: int

@app.task(
    description="Тестовая задача с использованием Pydantic.",
    tags=["example"],
)
async def example_pydantic(item: Item):
    return f"Hello, {item.name}!"

Все следующие варианты будут эквивалентны — QTasks корректно соберёт модель:

task = await example_pydantic.add_task("Test", value=42, timeout=50)
task = await example_pydantic.add_task("Test", 42, timeout=50)
task = await example_pydantic.add_task(name="Test", value=42, timeout=50)

Каждый из них приведёт к созданию одной и той же модели:

Item(name="Test", value=42)

Под капотом

Работа плагина происходит последовательно.

1. Сборка модели

Плагин:

  1. Анализирует сигнатуру задачи.
  2. Находит параметры, являющиеся наследниками pydantic.BaseModel.
  3. Извлекает подходящие kwargs.
  4. При необходимости использует остаточные позиционные аргументы args.
  5. Собирает модель:
model = Item(*args_from_task, **kwargs_from_task)

2. Очистка аргументов

Все параметры, использованные при создании модели, удаляются из фактических аргументов. Остальные аргументы передаются в задачу без изменений.

Это позволяет гибко комбинировать модель с другими параметрами:

@app.task
async def task_with_mix(item: Item, retries: int = 3):
    ...

Возврат моделей

Если задача возвращает объект BaseModel, QTasks автоматически преобразует его в сериализуемый словарь с помощью:

model.model_dump()

Пример:

@app.task
async def create_item() -> Item:
    return Item(name="Generated", value=10)

Фактический результат хранения в системе:

{"name": "Generated", "value": 10}

Особенности

  • Поддерживаются любые наследники BaseModel.
  • Сборка выполняется на стороне Worker перед вызовом пользовательской функции.
  • Плагин не меняет сигнатуру задачи.
  • Модели можно комбинировать с позиционными и именованными аргументами.
  • Модели в результате задачи автоматически сериализуются.

Итог

(A)syncPydanticWrapperPlugin обеспечивает:

  • прозрачную работу с типизированными структурами;
  • автоматическую валидацию входящих данных;
  • удобную интеграцию моделей в сигнатуру задач;
  • стандартный формат возврата данных.