Роутер

Router for task execution.

Router

Bases: SyncPluginMixin

Роутер, который хранит в себе задачи, которые подключает к себе основной QueueTasks.

Example

from qtasks import QueueTasks, Router

app = QueueTasks()

router = Router()

@router.task()
async def test():
    pass

app.include_router(router)

Source code in src/qtasks/routers/router.py

class Router(SyncPluginMixin):
    """
    Роутер, который хранит в себе задачи, которые подключает к себе основной `QueueTasks`.

    ## Example

    ```python
    from qtasks import QueueTasks, Router

    app = QueueTasks()

    router = Router()

    @router.task()
    async def test():
        pass

    app.include_router(router)
    ```
    """

    @overload
    def __init__(self, method: Literal["sync"] = None) -> None:
        """Инициализация роутера для синхронных задач."""
        ...

    @overload
    def __init__(self, method: Literal["async"] = None) -> None:
        """Инициализация роутера для асинхронных задач."""
        ...

    def __init__(self, method: Literal["sync", "async"] = None):
        """Инициализация роутера.

        Args:
            method (Literal["sync", "async"], optional): Метод выполнения задач. По умолчанию: `None`.
        """
        self._method = method
        self.tasks: Annotated[
            Dict[str, TaskExecSchema],
            Doc(
                """
                Задачи, тип `{task_name:qtasks.schemas.TaskExecSchema}`.

                По умолчанию: `Пустой словарь`.
                """
            ),
        ] = {}

        self.plugins: Dict[str, List["BasePlugin"]] = {}

    def task(
        self,
        name: Annotated[
            Optional[str],
            Doc(
                """
                    Имя задачи.

                    По умолчанию: `func.__name__`.
                    """
            ),
        ] = None,
        *,
        priority: Annotated[
            Optional[int],
            Doc(
                """
                    Приоритет у задачи по умолчанию.

                    По умолчанию: `config.default_task_priority`.
                    """
            ),
        ] = None,
        echo: Annotated[
            bool,
            Doc(
                """
                    Добавить (A)syncTask первым параметром.

                    По умолчанию: `False`.
                    """
            ),
        ] = False,
        max_time: Annotated[
            Union[float, None],
            Doc(
                """
                    Максимальное время выполнения задачи в секундах.

                    По умолчанию: `None`.
                    """
            )
        ] = None,
        retry: Annotated[
            Union[int, None],
            Doc(
                """
                    Количество попыток повторного выполнения задачи.

                    По умолчанию: `None`.
                    """
            ),
        ] = None,
        retry_on_exc: Annotated[
            Union[List[Type[Exception]], None],
            Doc(
                """
                    Исключения, при которых задача будет повторно выполнена.

                    По умолчанию: `None`.
                    """
            ),
        ] = None,
        decode: Annotated[
            Union[Callable, None],
            Doc(
                """
                    Декодер результата задачи.

                    По умолчанию: `None`.
                """
            )
        ] = None,
        tags: Annotated[
            Union[List[str], None],
            Doc(
                """
                    Теги задачи.

                    По умолчанию: `None`.
                """
            )
        ] = None,
        description: Annotated[
            Union[str, None],
            Doc(
                """
                    Описание задачи.

                    По умолчанию: `None`.
                """
            )
        ] = None,
        generate_handler: Annotated[
            Union[Callable, None],
            Doc(
                """
                    Генератор обработчика.

                    По умолчанию: `None`.
                    """
            ),
        ] = None,
        executor: Annotated[
            Type["BaseTaskExecutor"],
            Doc(
                """
                    Класс `BaseTaskExecutor`.

                    По умолчанию: `SyncTaskExecutor`.
                    """
            ),
        ] = None,
        middlewares_before: Annotated[
            List["TaskMiddleware"],
            Doc(
                """
                    Мидлвари, которые будут выполнены перед задачей.

                    По умолчанию: `Пустой массив`.
                    """
            ),
        ] = None,
        middlewares_after: Annotated[
            List["TaskMiddleware"],
            Doc(
                """
                    Мидлвари, которые будут выполнены после задачи.

                    По умолчанию: `Пустой массив`.
                    """
            ),
        ] = None,
        **kwargs
    ) -> Callable[[Callable[P, R]], Union[SyncTask[P, R], AsyncTask[P, R]]]:
        """Декоратор для регистрации задач.

        Args:
            name (str, optional): Имя задачи. По умолчанию: `func.__name__`.
            priority (int, optional): Приоритет у задачи по умолчанию. По умолчанию: `config.default_task_priority`.
            echo (bool, optional): Добавить (A)syncTask первым параметром. По умолчанию: `False`.
            max_time (float, optional): Максимальное время выполнения задачи в секундах. По умолчанию: `None`.
            retry (int, optional): Количество попыток повторного выполнения задачи. По умолчанию: `None`.
            retry_on_exc (List[Type[Exception]], optional): Исключения, при которых задача будет повторно выполнена. По умолчанию: `None`.
            decode (Callable, optional): Декодер результата задачи. По умолчанию: `None`.
            tags (List[str], optional): Теги задачи. По умолчанию: `None`.
            description (str, optional): Описание задачи. По умолчанию: `None`.
            generate_handler (Callable, optional): Генератор обработчика. По умолчанию: `None`.
            executor (Type["BaseTaskExecutor"], optional): Класс `BaseTaskExecutor`. По умолчанию: `SyncTaskExecutor`.
            middlewares_before (List["TaskMiddleware"], optional): Мидлвари, которые будут выполнены перед задачей. По умолчанию: `Пустой массив`.
            middlewares_after (List["TaskMiddleware"], optional): Мидлвари, которые будут выполнены после задачи. По умолчанию: `Пустой массив`.

        Raises:
            ValueError: Если задача с таким именем уже зарегистрирована.
            ValueError: Неизвестный метод {self._method}.

        Returns:
            SyncTask | AsyncTask: Декоратор для регистрации задачи.
        """

        def wrapper(func):
            nonlocal priority, middlewares_before, middlewares_after

            task_name = name or func.__name__
            if task_name in self.tasks:
                raise ValueError(f"Задача с именем {task_name} уже зарегистрирована!")

            if priority is None:
                priority = 0

            generating = False
            if inspect.isgeneratorfunction(func):
                generating = "sync"
            if inspect.isasyncgenfunction(func):
                generating = "async"

            middlewares_before = middlewares_before or []
            middlewares_after = middlewares_after or []

            model = TaskExecSchema(
                name=task_name,
                priority=priority,
                func=func,
                awaiting=inspect.iscoroutinefunction(func),
                generating=generating,
                echo=echo,
                max_time=max_time,
                retry=retry,
                retry_on_exc=retry_on_exc,
                decode=decode,
                tags=tags,
                description=description,
                generate_handler=generate_handler,
                executor=executor,
                middlewares_before=middlewares_before,
                middlewares_after=middlewares_after,
                extra=kwargs
            )

            self.tasks[task_name] = model
            if self._method not in ["async", "sync"]:
                raise ValueError(f"Неизвестный метод {self._method}")

            method = AsyncTask if self._method == "async" else SyncTask
            return method(
                app=self,
                task_name=model.name,
                priority=model.priority,
                echo=model.echo,
                max_time=model.max_time,
                retry=model.retry,
                retry_on_exc=model.retry_on_exc,
                decode=model.decode,
                tags=model.tags,
                description=model.description,
                generate_handler=model.generate_handler,
                executor=model.executor,
                middlewares_before=model.middlewares_before,
                middlewares_after=model.middlewares_after,
            )

        return wrapper

__init__(method=None)

__init__(method: Literal['sync'] = None) -> None

__init__(method: Literal['async'] = None) -> None

Инициализация роутера.

Parameters:

Name	Type	Description	Default
method	Literal['sync', 'async']	Метод выполнения задач. По умолчанию: None.	None

Source code in src/qtasks/routers/router.py

def __init__(self, method: Literal["sync", "async"] = None):
    """Инициализация роутера.

    Args:
        method (Literal["sync", "async"], optional): Метод выполнения задач. По умолчанию: `None`.
    """
    self._method = method
    self.tasks: Annotated[
        Dict[str, TaskExecSchema],
        Doc(
            """
            Задачи, тип `{task_name:qtasks.schemas.TaskExecSchema}`.

            По умолчанию: `Пустой словарь`.
            """
        ),
    ] = {}

    self.plugins: Dict[str, List["BasePlugin"]] = {}

task(name=None, *, priority=None, echo=False, max_time=None, retry=None, retry_on_exc=None, decode=None, tags=None, description=None, generate_handler=None, executor=None, middlewares_before=None, middlewares_after=None, **kwargs)

Декоратор для регистрации задач.

Parameters:

Name	Type	Description	Default
name	str	Имя задачи. По умолчанию: func.__name__.	None
priority	int	Приоритет у задачи по умолчанию. По умолчанию: config.default_task_priority.	None
echo	bool	Добавить (A)syncTask первым параметром. По умолчанию: False.	False
max_time	float	Максимальное время выполнения задачи в секундах. По умолчанию: None.	None
retry	int	Количество попыток повторного выполнения задачи. По умолчанию: None.	None
retry_on_exc	List[Type[Exception]]	Исключения, при которых задача будет повторно выполнена. По умолчанию: None.	None
decode	Callable	Декодер результата задачи. По умолчанию: None.	None
tags	List[str]	Теги задачи. По умолчанию: None.	None
description	str	Описание задачи. По умолчанию: None.	None
generate_handler	Callable	Генератор обработчика. По умолчанию: None.	None
executor	Type[BaseTaskExecutor]	Класс BaseTaskExecutor. По умолчанию: SyncTaskExecutor.	None
middlewares_before	List[TaskMiddleware]	Мидлвари, которые будут выполнены перед задачей. По умолчанию: Пустой массив.	None
middlewares_after	List[TaskMiddleware]	Мидлвари, которые будут выполнены после задачи. По умолчанию: Пустой массив.	None

Raises:

Type	Description
ValueError	Если задача с таким именем уже зарегистрирована.
ValueError	Неизвестный метод {self._method}.

Returns:

Type	Description
Callable[[Callable[P, R]], Union[SyncTask[P, R], AsyncTask[P, R]]]	SyncTask | AsyncTask: Декоратор для регистрации задачи.

Source code in src/qtasks/routers/router.py

def task(
    self,
    name: Annotated[
        Optional[str],
        Doc(
            """
                Имя задачи.

                По умолчанию: `func.__name__`.
                """
        ),
    ] = None,
    *,
    priority: Annotated[
        Optional[int],
        Doc(
            """
                Приоритет у задачи по умолчанию.

                По умолчанию: `config.default_task_priority`.
                """
        ),
    ] = None,
    echo: Annotated[
        bool,
        Doc(
            """
                Добавить (A)syncTask первым параметром.

                По умолчанию: `False`.
                """
        ),
    ] = False,
    max_time: Annotated[
        Union[float, None],
        Doc(
            """
                Максимальное время выполнения задачи в секундах.

                По умолчанию: `None`.
                """
        )
    ] = None,
    retry: Annotated[
        Union[int, None],
        Doc(
            """
                Количество попыток повторного выполнения задачи.

                По умолчанию: `None`.
                """
        ),
    ] = None,
    retry_on_exc: Annotated[
        Union[List[Type[Exception]], None],
        Doc(
            """
                Исключения, при которых задача будет повторно выполнена.

                По умолчанию: `None`.
                """
        ),
    ] = None,
    decode: Annotated[
        Union[Callable, None],
        Doc(
            """
                Декодер результата задачи.

                По умолчанию: `None`.
            """
        )
    ] = None,
    tags: Annotated[
        Union[List[str], None],
        Doc(
            """
                Теги задачи.

                По умолчанию: `None`.
            """
        )
    ] = None,
    description: Annotated[
        Union[str, None],
        Doc(
            """
                Описание задачи.

                По умолчанию: `None`.
            """
        )
    ] = None,
    generate_handler: Annotated[
        Union[Callable, None],
        Doc(
            """
                Генератор обработчика.

                По умолчанию: `None`.
                """
        ),
    ] = None,
    executor: Annotated[
        Type["BaseTaskExecutor"],
        Doc(
            """
                Класс `BaseTaskExecutor`.

                По умолчанию: `SyncTaskExecutor`.
                """
        ),
    ] = None,
    middlewares_before: Annotated[
        List["TaskMiddleware"],
        Doc(
            """
                Мидлвари, которые будут выполнены перед задачей.

                По умолчанию: `Пустой массив`.
                """
        ),
    ] = None,
    middlewares_after: Annotated[
        List["TaskMiddleware"],
        Doc(
            """
                Мидлвари, которые будут выполнены после задачи.

                По умолчанию: `Пустой массив`.
                """
        ),
    ] = None,
    **kwargs
) -> Callable[[Callable[P, R]], Union[SyncTask[P, R], AsyncTask[P, R]]]:
    """Декоратор для регистрации задач.

    Args:
        name (str, optional): Имя задачи. По умолчанию: `func.__name__`.
        priority (int, optional): Приоритет у задачи по умолчанию. По умолчанию: `config.default_task_priority`.
        echo (bool, optional): Добавить (A)syncTask первым параметром. По умолчанию: `False`.
        max_time (float, optional): Максимальное время выполнения задачи в секундах. По умолчанию: `None`.
        retry (int, optional): Количество попыток повторного выполнения задачи. По умолчанию: `None`.
        retry_on_exc (List[Type[Exception]], optional): Исключения, при которых задача будет повторно выполнена. По умолчанию: `None`.
        decode (Callable, optional): Декодер результата задачи. По умолчанию: `None`.
        tags (List[str], optional): Теги задачи. По умолчанию: `None`.
        description (str, optional): Описание задачи. По умолчанию: `None`.
        generate_handler (Callable, optional): Генератор обработчика. По умолчанию: `None`.
        executor (Type["BaseTaskExecutor"], optional): Класс `BaseTaskExecutor`. По умолчанию: `SyncTaskExecutor`.
        middlewares_before (List["TaskMiddleware"], optional): Мидлвари, которые будут выполнены перед задачей. По умолчанию: `Пустой массив`.
        middlewares_after (List["TaskMiddleware"], optional): Мидлвари, которые будут выполнены после задачи. По умолчанию: `Пустой массив`.

    Raises:
        ValueError: Если задача с таким именем уже зарегистрирована.
        ValueError: Неизвестный метод {self._method}.

    Returns:
        SyncTask | AsyncTask: Декоратор для регистрации задачи.
    """

    def wrapper(func):
        nonlocal priority, middlewares_before, middlewares_after

        task_name = name or func.__name__
        if task_name in self.tasks:
            raise ValueError(f"Задача с именем {task_name} уже зарегистрирована!")

        if priority is None:
            priority = 0

        generating = False
        if inspect.isgeneratorfunction(func):
            generating = "sync"
        if inspect.isasyncgenfunction(func):
            generating = "async"

        middlewares_before = middlewares_before or []
        middlewares_after = middlewares_after or []

        model = TaskExecSchema(
            name=task_name,
            priority=priority,
            func=func,
            awaiting=inspect.iscoroutinefunction(func),
            generating=generating,
            echo=echo,
            max_time=max_time,
            retry=retry,
            retry_on_exc=retry_on_exc,
            decode=decode,
            tags=tags,
            description=description,
            generate_handler=generate_handler,
            executor=executor,
            middlewares_before=middlewares_before,
            middlewares_after=middlewares_after,
            extra=kwargs
        )

        self.tasks[task_name] = model
        if self._method not in ["async", "sync"]:
            raise ValueError(f"Неизвестный метод {self._method}")

        method = AsyncTask if self._method == "async" else SyncTask
        return method(
            app=self,
            task_name=model.name,
            priority=model.priority,
            echo=model.echo,
            max_time=model.max_time,
            retry=model.retry,
            retry_on_exc=model.retry_on_exc,
            decode=model.decode,
            tags=model.tags,
            description=model.description,
            generate_handler=model.generate_handler,
            executor=model.executor,
            middlewares_before=model.middlewares_before,
            middlewares_after=model.middlewares_after,
        )

    return wrapper