AsyncTask

Async Task.

AsyncTask

Bases: Generic[P, R]

AsyncTask - класс для замены функции декоратором @app.task и @shared_task.

Пример

import asyncio
from qtasks import QueueTasks

app = QueueTasks()

@app.task("test")
async def test():
    print("Это тест!")

asyncio.run(await test.add_task())

__init__(task_name=None, priority=None, echo=False, retry=None, retry_on_exc=None, decode=None, tags=None, generate_handler=None, executor=None, middlewares=None, extra=None, app=None)

Инициализация асинхронной задачи.

Parameters:

Name	Type	Description	Default
task_name	str	Имя задачи. По умолчанию: None.	None
priority	int	Приоритет задачи. По умолчанию: None.	None
echo	bool	Включить вывод в консоль. По умолчанию: False.	False
retry	int | None	Количество попыток повторного выполнения задачи. По умолчанию: None.	None
retry_on_exc	list[Type[Exception]] | None	Исключения, при которых задача будет повторно выполнена. По умолчанию: None.	None
decode	Callable	Декодер результата задачи. По умолчанию: None.	None
tags	list[str] | None	Теги задачи. По умолчанию: None.	None
generate_handler	Callable | None	Генератор обработчика. По умолчанию: None.	None
executor	Type[BaseTaskExecutor]	Класс BaseTaskExecutor. По умолчанию: None.	None
middlewares	List[TaskMiddleware]	Мидлвари. По умолчанию: None.	None
app	QueueTasks	QueueTasks экземпляр. По умолчанию: None.	None

add_task(priority=None, args=None, kwargs=None, timeout=None, task_name=None) async

Добавить задачу.

Parameters:

Name	Type	Description	Default
priority	int	Приоритет задачи. По умолчанию: Значение приоритета у задачи.	None
args	tuple	args задачи. По умолчанию: ().	None
kwargs	dict	kwargs задачи. По умолчанию: {}.	None
timeout	float	Таймаут задачи. Если указан, задача возвращается через qtasks.results.SyncTask.	None
task_name	str	Имя задачи. По умолчанию: Значение имени у задачи.	None

Returns:

Type	Description
Task | None	Task|None: Результат задачи или None.