Source code for pyobs.robotic.storage.taskarchive

from abc import ABCMeta, abstractmethod
from collections.abc import Callable, Coroutine
from typing import Any

from pyobs.object import Object
from pyobs.robotic.task import Project, Task
from pyobs.utils.time import Time


[docs] class TaskArchive(Object, metaclass=ABCMeta): def __init__(self, on_tasks_changed: Callable[[], Coroutine[Any, Any, None]] | None = None, **kwargs: Any): Object.__init__(self, **kwargs) self._on_tasks_changed = on_tasks_changed
[docs] @abstractmethod async def last_changed(self) -> Time | None: """Returns time when last time any tasks changed.""" ...
[docs] @abstractmethod async def get_projects(self) -> list[Project]: """Returns list of projects. Returns: List of projects. """ ...
[docs] @abstractmethod async def get_schedulable_tasks(self) -> list[Task]: """Returns list of schedulable tasks. Returns: List of schedulable tasks """ ...
[docs] @abstractmethod async def get_task(self, id: Any) -> Task | None: """Returns the task with the given ID. Returns: Task with given ID. """ ...
__all__ = ["TaskArchive"]