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

Contexts — Контексты задач

Каждая задача QTasks во время выполнения получает доступ к контексту выполнения — объекту (A)syncContext. Контекст предоставляет безопасный и унифицированный способ взаимодействия задачи с окружением приложения, воркером, плагинами и метаданными выполнения.

Контекст доступен через self.ctx внутри объекта (A)syncTask.

Передача контекста в функцию задачи

Контекст может быть передан напрямую в функцию задачи с помощью параметра echo=True в декораторе @app.task.

@app.task(echo=True)
async def my_task(self):
    print(self.ctx.task_name)

В этом случае:

  • первым аргументом функции передаётся дублированный (A)syncContext;
  • имя параметра может быть любым;
  • для удобства и единообразия рекомендуется использовать имя self.

Контекст добавляется на этапе выполнения задачи и не влияет на её регистрацию.

Назначение контекста

(A)syncContext предназначен для:

  • доступа к данным текущей задачи;
  • взаимодействия с приложением (QueueTasks);
  • работы с логированием;
  • управления выполнением (sleep, cancel);
  • взаимодействия с плагинами и триггерами;
  • получения метаданных и конфигурации.

Контекст не содержит бизнес-логики и служит исключительно интерфейсом среды выполнения.

Основные свойства и методы

Идентификация задачи

  • self.ctx.task_name — имя задачи.
  • self.ctx.task_uuid — UUID выполняемой задачи.

Генерация задач (yield)

  • self.ctx.generate_handler — функция-обработчик генерации для yield-задач.

Используется совместно с параметром декоратора:

@app.task(generate_handler=gen_func)

Логирование

self.ctx.get_logger(name: str | None = None)

Возвращает экземпляр qtasks.logs.Logger, привязанный к задаче.

  • если name не указан — используется task_name;
  • если имя задачи отсутствует — используется (A)syncContext.

Конфигурация и метаданные

  • self.ctx.get_config() — возвращает app.config.

  • self.ctx.get_metadata(cache: bool = True) — возвращает метаданные задачи через app.get(self.ctx.task_uuid).

При cache=True результат кешируется внутри контекста и повторные вызовы не обращаются к хранилищу.

Работа с задачами

  • self.ctx.get_task(uuid: UUID | str) — возвращает задачу через app.get(uuid).

Управление выполнением

  • self.ctx.sleep(seconds: float) — вызывает time.sleep или asyncio.sleep в зависимости от типа задачи.

  • self.ctx.cancel(reason: str = "") — инициирует отмену задачи путём выброса:

raise TaskCancelError(reason or f"{self.ctx.task_name}.cancel")

Отмена обрабатывается воркером как корректное завершение задачи.

Взаимодействие с плагинами

  • self.ctx.plugin_error(**kwargs) — выбрасывает TaskPluginTriggerError(**kwargs).

Используется для уведомления плагинов, имеющих триггер:

task_executor_run_task_trigger_error

Доступ к компонентам приложения

  • self.ctx.get_component(name: str) — эквивалентен getattr(app, name, None).

Позволяет получить доступ к любому зарегистрированному компоненту QTasks.

Источник app

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

  • на этапе выполнения задачи через TaskExecutor, который дополняет (A)syncTask и его ctx;
  • через глобальную ссылку qtasks._state.app_main, если задача была вызвана вне явного контекста.

Это обеспечивает единый доступ к приложению независимо от сценария запуска.

Итог

(A)syncContext:

  • является интерфейсом среды выполнения задачи;
  • изолирует бизнес-логику от инфраструктуры;
  • предоставляет доступ к конфигурации, логам и метаданным;
  • используется для управления жизненным циклом задачи;
  • служит точкой интеграции с плагинами и компонентами.

Страница оформлена как холст и может использоваться как базовая документация для контекста, echo-задач и плагинных механизмов QTasks.