"""Abstract backend contracts and dependency-error normalization."""
from __future__ import annotations
from abc import ABC, abstractmethod
from typing import TYPE_CHECKING, Any
from .errors import (
BackendCloseError,
BackendError,
BackendExportError,
BackendLoadError,
BackendPreparationError,
BackendSaveError,
QHarborToolsError,
RemoteAccessError,
)
if TYPE_CHECKING:
from collections.abc import Mapping, Sequence
from .models import (
AnnouncementRequest,
Artifact,
BackendContext,
RemoteDataset,
RunReference,
SaveRequest,
StoredRun,
)
from .receipts import Publication
[docs]
class RemoteBackend(ABC):
"""Abstract remote dataset service used by :class:`Experiment`."""
name = "remote"
[docs]
def announce(self, request: AnnouncementRequest) -> RemoteDataset:
"""Create a remote dataset and normalize dependency failures."""
return self._remote_call("announce", lambda: self._announce(request))
[docs]
def open(self, uuid: str) -> RemoteDataset:
"""Open a remote dataset and normalize dependency failures."""
return self._remote_call("open", lambda: self._open(uuid), dataset_uuid=uuid)
[docs]
def publish(self, dataset: RemoteDataset, artifact: Artifact) -> Publication:
"""Publish one artifact."""
return self._remote_call(
"publish",
lambda: self._publish(dataset, artifact),
dataset_uuid=dataset.uuid,
artifact_name=artifact.name,
)
[docs]
def close(self) -> None:
"""Release remote resources."""
self._remote_call("close", self._close)
def _remote_call(self, operation: str, call, **context):
try:
return call()
except QHarborToolsError:
raise
except Exception as error:
raise RemoteAccessError(
f"Remote operation '{operation}' failed.",
operation=operation,
backend_name=self.name,
**context,
) from error
@abstractmethod
def _announce(self, request: AnnouncementRequest) -> RemoteDataset: ...
@abstractmethod
def _open(self, uuid: str) -> RemoteDataset: ...
@abstractmethod
def _update_metadata(self, dataset: RemoteDataset, metadata: Mapping[str, Any]) -> RemoteDataset: ...
@abstractmethod
def _publish(self, dataset: RemoteDataset, artifact: Artifact) -> Publication: ...
def _close(self) -> None:
return None
[docs]
class MeasurementBackend(ABC):
"""Abstract local measurement storage and export backend."""
name = "measurement"
[docs]
def prepare(self, context: BackendContext) -> None:
"""Prepare the backend for an experiment."""
try:
self._prepare(context)
except BackendError:
raise
except Exception as error:
raise BackendPreparationError(
"The measurement backend could not be prepared.",
operation="prepare",
backend_name=self.name,
dataset_uuid=context.dataset_uuid,
) from error
[docs]
def save(self, request: SaveRequest) -> StoredRun:
"""Persist one run and normalize dependency failures."""
try:
return self._save(request)
except BackendError:
raise
except Exception as error:
raise BackendSaveError(
"The measurement backend could not save the run.",
operation="save",
backend_name=self.name,
) from error
[docs]
def load(self, run: RunReference) -> StoredRun:
"""Reload a run and normalize dependency failures."""
if run.backend != self.name:
raise BackendLoadError(
f"Run belongs to backend '{run.backend}', not '{self.name}'.",
operation="load",
backend_name=self.name,
run_reference=run,
)
try:
return self._load(run)
except BackendError:
raise
except Exception as error:
raise BackendLoadError(
"The measurement backend could not load the run.",
operation="load",
backend_name=self.name,
run_reference=run,
) from error
[docs]
def export(self, run: StoredRun, *, variables: Sequence[str] | None = None) -> list[Artifact]:
"""Export a run as portable artifacts."""
try:
return self._export(run, variables=variables)
except BackendError:
raise
except Exception as error:
raise BackendExportError(
"The measurement backend could not export the run.",
operation="export",
backend_name=self.name,
run_reference=run.reference,
) from error
[docs]
def close(self) -> None:
"""Release resources and normalize dependency failures."""
try:
self._close()
except BackendError:
raise
except Exception as error:
raise BackendCloseError(
"The measurement backend could not close cleanly.",
operation="close",
backend_name=self.name,
) from error
@abstractmethod
def _prepare(self, context: BackendContext) -> None: ...
@abstractmethod
def _save(self, request: SaveRequest) -> StoredRun: ...
@abstractmethod
def _load(self, run: RunReference) -> StoredRun: ...
@abstractmethod
def _export(self, run: StoredRun, *, variables: Sequence[str] | None = None) -> list[Artifact]: ...
def _close(self) -> None:
return None