import asyncio
import logging
from typing import Any
from urllib.parse import urljoin
import aiohttp
from pyobs.robotic.storage.taskarchive import TaskArchive
from pyobs.robotic.task import Project, Task
from pyobs.utils.http import http_request_with_retries
from pyobs.utils.time import Time
log = logging.getLogger(__name__)
[docs]
class BackendTaskArchive(TaskArchive):
"""Task archive based on pyobs-robotic-backend."""
def __init__(self, url: str, token: str, auto_update: bool = True, **kwargs: Any):
"""Creates a new task archive.
Args:
url: URL of pyobs-robotic-backend.
token: Auth token.
"""
TaskArchive.__init__(self, **kwargs)
self._url = url
self._token = token
self._aiohttp_session: aiohttp.ClientSession | None = None
self._last_update: Time | None = None
self._projects: list[Project] = list()
self._tasks: list[Task] = list()
if auto_update:
self.add_background_task(self._check_for_changes)
[docs]
async def open(self) -> None:
"""Opens the backend task archive."""
self._aiohttp_session = aiohttp.ClientSession(headers={"Authorization": f"Token {self._token}"})
await TaskArchive.open(self)
[docs]
async def close(self) -> None:
"""Closes the backend observation archive."""
await TaskArchive.close(self)
if self._aiohttp_session is not None:
await self._aiohttp_session.close()
self._aiohttp_session = None
@property
def _session(self) -> aiohttp.ClientSession:
if self._aiohttp_session is None:
raise ValueError("No session available.")
return self._aiohttp_session
async def _check_for_changes(self) -> None:
"""Update tasks in background."""
while True:
try:
last_update = await self.last_update_time()
if self._last_update is None or self._last_update < last_update:
self._projects = await self._get_projects()
self._tasks = await self._get_tasks()
log.info("Downloaded new tasks/projects.")
self._last_update = last_update
if self._on_tasks_changed is not None:
await self._on_tasks_changed()
except Exception as e:
log.error("Failed to update tasks from backend: %s", e)
await asyncio.sleep(5)
[docs]
async def last_update_time(self) -> Time:
"""Fetches last schedule update time."""
res = await http_request_with_retries(self._session, urljoin(self._url, "/api/last_task_update/"))
return Time(res["last_task_update"])
async def _get_projects(self) -> list[Project]:
"""Fetch projects from backend."""
projects = await http_request_with_retries(self._session, urljoin(self._url, "/api/projects/"))
return [self.pyobs_model_validate(Project, project) for project in projects["results"]]
async def _get_tasks(self) -> list[Task]:
"""Fetch tasks from backend."""
tasks = await http_request_with_retries(self._session, urljoin(self._url, "/api/tasks/"))
return [self.pyobs_model_validate(Task, task) for task in tasks["results"]]
[docs]
async def last_changed(self) -> Time | None:
"""Returns time when last time any tasks changed."""
return self._last_update
[docs]
async def get_projects(self) -> list[Project]:
"""Returns list of projects.
Returns:
List of projects.
"""
return self._projects
[docs]
async def get_schedulable_tasks(self) -> list[Task]:
"""Returns list of schedulable tasks.
Returns:
List of schedulable tasks
"""
return self._tasks
[docs]
async def get_task(self, id: Any) -> Task | None:
"""Returns the task with the given ID.
Returns:
Task with given ID.
"""
for task in self._tasks:
if task.id == id:
return task
else:
return None
__all__ = ["BackendTaskArchive"]