BaseStorage - Пишем свое Хранилище

Base storage class.

BaseStorage

Bases: ABC

BaseStorage - Абстрактный класс, который является фундаментом для Хранилищ.

Пример

from qtasks.storages.base import BaseStorage

class MyStorage(BaseStorage):
    def __init__(self, name: str = None):
        super().__init__(name=name)
        pass

Source code in src/qtasks/storages/base.py

class BaseStorage(ABC):
    """
    `BaseStorage` - Абстрактный класс, который является фундаментом для Хранилищ.

    ## Пример

    ```python
    from qtasks.storages.base import BaseStorage

    class MyStorage(BaseStorage):
        def __init__(self, name: str = None):
            super().__init__(name=name)
            pass
    ```
    """

    def __init__(
        self,
        name: Annotated[
            Optional[str],
            Doc(
                """
                    Имя проекта. Это имя можно использовать для тегов для Storage.

                    По умолчанию: `None`.
                    """
            ),
        ] = None,
        global_config: Annotated[
            Optional["BaseGlobalConfig"],
            Doc(
                """
                    Глобальный конфиг.

                    По умолчанию: `None`.
                    """
            ),
        ] = None,
        log: Annotated[
            Optional[Logger],
            Doc(
                """
                    Логгер.

                    По умолчанию: `qtasks.logs.Logger`.
                    """
            ),
        ] = None,
        config: Annotated[
            Optional[QueueConfig],
            Doc(
                """
                    Конфиг.

                    По умолчанию: `qtasks.configs.config.QueueConfig`.
                    """
            ),
        ] = None,
        events: Annotated[
            Optional["BaseEvents"],
            Doc(
                """
                    События.

                    По умолчанию: `None`.
                    """
            ),
        ] = None,
    ):
        """Инициализация базового хранилища.

        Args:
            name (str, optional): Имя проекта. По умолчанию: `None`.
            global_config (BaseGlobalConfig, optional): Глобальный конфиг. По умолчанию: `None`.
            log (Logger, optional): Логгер. По умолчанию: `qtasks.logs.Logger`.
            config (QueueConfig, optional): Конфиг. По умолчанию: `qtasks.configs.config.QueueConfig`.
            events (BaseEvents, optional): События. По умолчанию: `None`.
        """
        self.name = name
        self.global_config: Union["BaseGlobalConfig", None] = global_config

        self.config = config or QueueConfig()
        self.log = (
            log.with_subname("Storage")
            if log
            else Logger(
                name=self.name,
                subname="Storage",
                default_level=self.config.logs_default_level_server,
                format=self.config.logs_format,
            )
        )
        self.events = events

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

        self.init_plugins()

    @abstractmethod
    def add(
        self,
        uuid: Annotated[
            Union[UUID, str],
            Doc(
                """
                    UUID задачи.
                    """
            ),
        ],
        task_status: Annotated[
            TaskStatusNewSchema,
            Doc(
                """
                    Схема статуса новой задачи.
                    """
            ),
        ],
    ) -> None:
        """Добавление задачи в хранилище.

        Args:
            uuid (UUID | str): UUID задачи.
            task_status (TaskStatusNewSchema): Схема статуса новой задачи.
        """
        pass

    @abstractmethod
    def get(
        self,
        uuid: Annotated[
            Union[UUID, str],
            Doc(
                """
                    UUID задачи.
                    """
            ),
        ],
    ) -> Union[Task, None]:
        """Получение информации о задаче.

        Args:
            uuid (UUID|str): UUID задачи.

        Returns:
            Task|None: Если есть информация о задаче, возвращает `schemas.task.Task`, иначе `None`.
        """
        pass

    @abstractmethod
    def get_all(self) -> List[Task]:
        """Получить все задачи.

        Returns:
            List[Task]: Массив задач.
        """
        pass

    @abstractmethod
    def update(
        self,
        **kwargs: Annotated[
            dict,
            Doc(
                """
                    Аргументы обновления типа kwargs.
                    """
            ),
        ],
    ) -> None:
        """Обновляет информацию о задаче.

        Args:
            kwargs (dict, optional): данные задачи типа kwargs.
        """
        pass

    @abstractmethod
    def start(self):
        """Запускает хранилище. Эта функция задействуется основным экземпляром `QueueTasks` после запуске функции `run_forever`."""
        pass

    @abstractmethod
    def stop(self):
        """Останавливает хранилище. Эта функция задействуется основным экземпляром `QueueTasks` после завершения функции `run_forever`."""
        pass

    def add_plugin(
        self,
        plugin: Annotated[
            "BasePlugin",
            Doc(
                """
                    Плагин.
                    """
            ),
        ],
        trigger_names: Annotated[
            Optional[List[str]],
            Doc(
                """
                    Имя триггеров для плагина.

                    По умолчанию: По умолчанию: будет добавлен в `Globals`.
                    """
            ),
        ] = None,
    ) -> None:
        """Добавить плагин в класс.

        Args:
            plugin (BasePlugin): Плагин
            trigger_names (List[str], optional): Имя триггеров для плагина. По умолчанию: будет добавлен в `Globals`.
        """
        trigger_names = trigger_names or ["Globals"]

        for name in trigger_names:
            if name not in self.plugins:
                self.plugins.update({name: [plugin]})
            else:
                self.plugins[name].append(plugin)
        return

    def remove_finished_task(
        self,
        task_broker: Annotated[
            TaskPrioritySchema,
            Doc(
                """
                    Схема приоритетной задачи.
                    """
            ),
        ],
        model: Annotated[
            Union[
                TaskStatusProcessSchema, TaskStatusErrorSchema, TaskStatusCancelSchema
            ],
            Doc(
                """
                    Модель результата задачи.
                    """
            ),
        ],
    ) -> None:
        """Обновляет данные хранилища через функцию `self.storage.remove_finished_task`.

        Args:
            task_broker (TaskPrioritySchema): Схема приоритетной задачи.
            model (TaskStatusNewSchema | TaskStatusErrorSchema): Модель результата задачи.
        """
        pass

    def _delete_finished_tasks(self, **kwargs) -> None:
        pass

    def _running_older_tasks(self, **kwargs) -> None:
        pass

    def update_config(
        self,
        config: Annotated[
            QueueConfig,
            Doc(
                """
                    Конфиг.
                    """
            ),
        ],
    ) -> None:
        """Обновляет конфиг брокера.

        Args:
            config (QueueConfig): Конфиг.
        """
        self.config = config
        return

    def flush_all(self) -> None:
        """Удалить все данные."""
        pass

    def _build_task(self, uuid, result: dict) -> Task:
        # Сначала собираем стандартные аргументы Task

        base_kwargs = dict(
            status=result["status"],
            uuid=uuid,
            priority=int(result["priority"]),
            task_name=result["task_name"],
            args=json.loads(result["args"]),
            kwargs=json.loads(result["kwargs"]),
            created_at=datetime.datetime.fromtimestamp(float(result["created_at"])),
            updated_at=datetime.datetime.fromtimestamp(float(result["updated_at"])),
        )

        # Вычисляем имена стандартных полей
        task_field_names = {f.name for f in fields(Task)}

        # Ищем дополнительные ключи
        extra_fields = []
        extra_values = {}

        for key, value in result.items():
            if key not in task_field_names:
                # Типизация примитивная — можно улучшить
                field_type = self._infer_type(value)
                extra_fields.append((key, field_type, field(default=None)))

                # Можно привести значение к типу
                if field_type is bool:
                    extra_values[key] = value.lower() == "true"
                elif field_type is int:
                    extra_values[key] = int(value)
                elif field_type is float:
                    extra_values[key] = float(value)
                else:
                    extra_values[key] = value

        # Создаем новый dataclass с дополнительными полями
        if extra_fields:
            NewTask = make_dataclass("Task", extra_fields, bases=(Task,))
        else:
            NewTask = Task

        # Объединяем все аргументы
        task = NewTask(**base_kwargs, **extra_values)
        if hasattr(task, "returning"):
            with contextlib.suppress(BaseException):
                task.returning = json.loads(task.returning)
        return task

    def _infer_type(self, value: str):
        """Пытается определить реальный тип из строки."""
        if value.lower() in {"true", "false"}:
            return bool
        try:
            int(value)
            return int
        except ValueError:
            pass
        try:
            float(value)
            return float
        except ValueError:
            pass
        return str

    def init_plugins(self):
        """Инициализация плагинов."""
        pass

__init__(name=None, global_config=None, log=None, config=None, events=None)

Инициализация базового хранилища.

Parameters:

Name	Type	Description	Default
name	str	Имя проекта. По умолчанию: None.	None
global_config	BaseGlobalConfig	Глобальный конфиг. По умолчанию: None.	None
log	Logger	Логгер. По умолчанию: qtasks.logs.Logger.	None
config	QueueConfig	Конфиг. По умолчанию: qtasks.configs.config.QueueConfig.	None
events	BaseEvents	События. По умолчанию: None.	None

Source code in src/qtasks/storages/base.py

def __init__(
    self,
    name: Annotated[
        Optional[str],
        Doc(
            """
                Имя проекта. Это имя можно использовать для тегов для Storage.

                По умолчанию: `None`.
                """
        ),
    ] = None,
    global_config: Annotated[
        Optional["BaseGlobalConfig"],
        Doc(
            """
                Глобальный конфиг.

                По умолчанию: `None`.
                """
        ),
    ] = None,
    log: Annotated[
        Optional[Logger],
        Doc(
            """
                Логгер.

                По умолчанию: `qtasks.logs.Logger`.
                """
        ),
    ] = None,
    config: Annotated[
        Optional[QueueConfig],
        Doc(
            """
                Конфиг.

                По умолчанию: `qtasks.configs.config.QueueConfig`.
                """
        ),
    ] = None,
    events: Annotated[
        Optional["BaseEvents"],
        Doc(
            """
                События.

                По умолчанию: `None`.
                """
        ),
    ] = None,
):
    """Инициализация базового хранилища.

    Args:
        name (str, optional): Имя проекта. По умолчанию: `None`.
        global_config (BaseGlobalConfig, optional): Глобальный конфиг. По умолчанию: `None`.
        log (Logger, optional): Логгер. По умолчанию: `qtasks.logs.Logger`.
        config (QueueConfig, optional): Конфиг. По умолчанию: `qtasks.configs.config.QueueConfig`.
        events (BaseEvents, optional): События. По умолчанию: `None`.
    """
    self.name = name
    self.global_config: Union["BaseGlobalConfig", None] = global_config

    self.config = config or QueueConfig()
    self.log = (
        log.with_subname("Storage")
        if log
        else Logger(
            name=self.name,
            subname="Storage",
            default_level=self.config.logs_default_level_server,
            format=self.config.logs_format,
        )
    )
    self.events = events

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

    self.init_plugins()

add(uuid, task_status) abstractmethod

Добавление задачи в хранилище.

Parameters:

Name	Type	Description	Default
uuid	UUID | str	UUID задачи.	required
task_status	TaskStatusNewSchema	Схема статуса новой задачи.	required

Source code in src/qtasks/storages/base.py

@abstractmethod
def add(
    self,
    uuid: Annotated[
        Union[UUID, str],
        Doc(
            """
                UUID задачи.
                """
        ),
    ],
    task_status: Annotated[
        TaskStatusNewSchema,
        Doc(
            """
                Схема статуса новой задачи.
                """
        ),
    ],
) -> None:
    """Добавление задачи в хранилище.

    Args:
        uuid (UUID | str): UUID задачи.
        task_status (TaskStatusNewSchema): Схема статуса новой задачи.
    """
    pass

add_plugin(plugin, trigger_names=None)

Добавить плагин в класс.

Parameters:

Name	Type	Description	Default
plugin	BasePlugin	Плагин	required
trigger_names	List[str]	Имя триггеров для плагина. По умолчанию: будет добавлен в Globals.	None

Source code in src/qtasks/storages/base.py

def add_plugin(
    self,
    plugin: Annotated[
        "BasePlugin",
        Doc(
            """
                Плагин.
                """
        ),
    ],
    trigger_names: Annotated[
        Optional[List[str]],
        Doc(
            """
                Имя триггеров для плагина.

                По умолчанию: По умолчанию: будет добавлен в `Globals`.
                """
        ),
    ] = None,
) -> None:
    """Добавить плагин в класс.

    Args:
        plugin (BasePlugin): Плагин
        trigger_names (List[str], optional): Имя триггеров для плагина. По умолчанию: будет добавлен в `Globals`.
    """
    trigger_names = trigger_names or ["Globals"]

    for name in trigger_names:
        if name not in self.plugins:
            self.plugins.update({name: [plugin]})
        else:
            self.plugins[name].append(plugin)
    return

flush_all()

Удалить все данные.

Source code in src/qtasks/storages/base.py

def flush_all(self) -> None:
    """Удалить все данные."""
    pass

get(uuid) abstractmethod

Получение информации о задаче.

Parameters:

Name	Type	Description	Default
uuid	UUID | str	UUID задачи.	required

Returns:

Type	Description
Union[Task, None]	Task|None: Если есть информация о задаче, возвращает schemas.task.Task, иначе None.

Source code in src/qtasks/storages/base.py

@abstractmethod
def get(
    self,
    uuid: Annotated[
        Union[UUID, str],
        Doc(
            """
                UUID задачи.
                """
        ),
    ],
) -> Union[Task, None]:
    """Получение информации о задаче.

    Args:
        uuid (UUID|str): UUID задачи.

    Returns:
        Task|None: Если есть информация о задаче, возвращает `schemas.task.Task`, иначе `None`.
    """
    pass

get_all() abstractmethod

Получить все задачи.

Returns:

Type	Description
List[Task]	List[Task]: Массив задач.

Source code in src/qtasks/storages/base.py

@abstractmethod
def get_all(self) -> List[Task]:
    """Получить все задачи.

    Returns:
        List[Task]: Массив задач.
    """
    pass

init_plugins()

Инициализация плагинов.

Source code in src/qtasks/storages/base.py

def init_plugins(self):
    """Инициализация плагинов."""
    pass

remove_finished_task(task_broker, model)

Обновляет данные хранилища через функцию self.storage.remove_finished_task.

Parameters:

Name	Type	Description	Default
task_broker	TaskPrioritySchema	Схема приоритетной задачи.	required
model	TaskStatusNewSchema | TaskStatusErrorSchema	Модель результата задачи.	required

Source code in src/qtasks/storages/base.py

def remove_finished_task(
    self,
    task_broker: Annotated[
        TaskPrioritySchema,
        Doc(
            """
                Схема приоритетной задачи.
                """
        ),
    ],
    model: Annotated[
        Union[
            TaskStatusProcessSchema, TaskStatusErrorSchema, TaskStatusCancelSchema
        ],
        Doc(
            """
                Модель результата задачи.
                """
        ),
    ],
) -> None:
    """Обновляет данные хранилища через функцию `self.storage.remove_finished_task`.

    Args:
        task_broker (TaskPrioritySchema): Схема приоритетной задачи.
        model (TaskStatusNewSchema | TaskStatusErrorSchema): Модель результата задачи.
    """
    pass

start() abstractmethod

Запускает хранилище. Эта функция задействуется основным экземпляром QueueTasks после запуске функции run_forever.

Source code in src/qtasks/storages/base.py

@abstractmethod
def start(self):
    """Запускает хранилище. Эта функция задействуется основным экземпляром `QueueTasks` после запуске функции `run_forever`."""
    pass

stop() abstractmethod

Останавливает хранилище. Эта функция задействуется основным экземпляром QueueTasks после завершения функции run_forever.

Source code in src/qtasks/storages/base.py

@abstractmethod
def stop(self):
    """Останавливает хранилище. Эта функция задействуется основным экземпляром `QueueTasks` после завершения функции `run_forever`."""
    pass

update(**kwargs) abstractmethod

Обновляет информацию о задаче.

Parameters:

Name	Type	Description	Default
kwargs	dict	данные задачи типа kwargs.	{}

Source code in src/qtasks/storages/base.py

@abstractmethod
def update(
    self,
    **kwargs: Annotated[
        dict,
        Doc(
            """
                Аргументы обновления типа kwargs.
                """
        ),
    ],
) -> None:
    """Обновляет информацию о задаче.

    Args:
        kwargs (dict, optional): данные задачи типа kwargs.
    """
    pass

update_config(config)

Обновляет конфиг брокера.

Parameters:

Name	Type	Description	Default
config	QueueConfig	Конфиг.	required

Source code in src/qtasks/storages/base.py

def update_config(
    self,
    config: Annotated[
        QueueConfig,
        Doc(
            """
                Конфиг.
                """
        ),
    ],
) -> None:
    """Обновляет конфиг брокера.

    Args:
        config (QueueConfig): Конфиг.
    """
    self.config = config
    return