Spaces:
Sleeping
Sleeping
File size: 2,884 Bytes
c06ce83 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 | from typing import Iterator
from trackio.sqlite_storage import SQLiteStorage
class Run:
def __init__(self, project: str, name: str, run_id: str | None = None):
self.project = project
self.name = name
self._id = run_id or name
self._config = None
@property
def id(self) -> str:
return self._id
@property
def config(self) -> dict | None:
if self._config is None:
self._config = SQLiteStorage.get_run_config(
self.project, self.name, run_id=self.id
)
return self._config
def alerts(self, level: str | None = None, since: str | None = None) -> list[dict]:
return SQLiteStorage.get_alerts(
self.project, run_name=self.name, run_id=self.id, level=level, since=since
)
def delete(self) -> bool:
return SQLiteStorage.delete_run(self.project, self.name, run_id=self.id)
def move(self, new_project: str) -> bool:
success = SQLiteStorage.move_run(
self.project, self.name, new_project, run_id=self.id
)
if success:
self.project = new_project
return success
def rename(self, new_name: str) -> "Run":
SQLiteStorage.rename_run(self.project, self.name, new_name, run_id=self.id)
self.name = new_name
return self
def __repr__(self) -> str:
return f"<Run {self.name} in project {self.project}>"
class Runs:
def __init__(self, project: str):
self.project = project
self._runs = None
def _load_runs(self):
if self._runs is None:
records = SQLiteStorage.get_run_records(self.project)
self._runs = [
Run(self.project, str(record["name"]), run_id=str(record["id"]))
for record in records
]
def __iter__(self) -> Iterator[Run]:
self._load_runs()
return iter(self._runs)
def __getitem__(self, index: int) -> Run:
self._load_runs()
return self._runs[index]
def __len__(self) -> int:
self._load_runs()
return len(self._runs)
def __repr__(self) -> str:
self._load_runs()
return f"<Runs project={self.project} count={len(self._runs)}>"
class Api:
def runs(self, project: str) -> Runs:
if not SQLiteStorage.get_project_db_path(project).exists():
raise ValueError(f"Project '{project}' does not exist")
return Runs(project)
def alerts(
self,
project: str,
run: str | None = None,
level: str | None = None,
since: str | None = None,
) -> list[dict]:
if not SQLiteStorage.get_project_db_path(project).exists():
raise ValueError(f"Project '{project}' does not exist")
return SQLiteStorage.get_alerts(project, run_name=run, level=level, since=since)
|