Services¶

services ¶

Service facades for typed SHM entities.

Classes¶

ApiShmRepository ¶

ApiShmRepository(api=None)

Repository adapter built on top of :class:ShmAPI.

Source code in src/owi/metadatabase/shm/services/core.py

def __init__(self, api: ShmAPI | None = None) -> None:
    self.api = api or ShmAPI(token="dummy")
    self._list_methods = {
        ShmEntityName.SENSOR_TYPE: self.api.list_sensor_types,
        ShmEntityName.SENSOR: self.api.list_sensors,
        ShmEntityName.SENSOR_CALIBRATION: self.api.list_sensor_calibrations,
        ShmEntityName.SIGNAL: self.api.list_signals,
        ShmEntityName.SIGNAL_HISTORY: self.api.list_signal_history,
        ShmEntityName.SIGNAL_CALIBRATION: self.api.list_signal_calibrations,
        ShmEntityName.DERIVED_SIGNAL: self.api.list_derived_signals,
        ShmEntityName.DERIVED_SIGNAL_HISTORY: self.api.list_derived_signal_history,
        ShmEntityName.DERIVED_SIGNAL_CALIBRATION: self.api.list_derived_signal_calibrations,
    }
    self._get_methods = {
        ShmEntityName.SENSOR_TYPE: self.api.get_sensor_type,
        ShmEntityName.SENSOR: self.api.get_sensor,
        ShmEntityName.SENSOR_CALIBRATION: self.api.get_sensor_calibration,
        ShmEntityName.SIGNAL: self.api.get_signal,
        ShmEntityName.SIGNAL_HISTORY: self.api.get_signal_history,
        ShmEntityName.SIGNAL_CALIBRATION: self.api.get_signal_calibration,
        ShmEntityName.DERIVED_SIGNAL: self.api.get_derived_signal,
        ShmEntityName.DERIVED_SIGNAL_HISTORY: self.api.get_derived_signal_history,
        ShmEntityName.DERIVED_SIGNAL_CALIBRATION: self.api.get_derived_signal_calibration,
    }
    self._create_methods = {
        ShmEntityName.SENSOR_TYPE: self.api.create_sensor_type,
        ShmEntityName.SENSOR: self.api.create_sensor,
        ShmEntityName.SENSOR_CALIBRATION: self.api.create_sensor_calibration,
        ShmEntityName.SIGNAL: self.api.create_signal,
        ShmEntityName.SIGNAL_HISTORY: self.api.create_signal_history,
        ShmEntityName.SIGNAL_CALIBRATION: self.api.create_signal_calibration,
        ShmEntityName.DERIVED_SIGNAL: self.api.create_derived_signal,
        ShmEntityName.DERIVED_SIGNAL_HISTORY: self.api.create_derived_signal_history,
        ShmEntityName.DERIVED_SIGNAL_CALIBRATION: self.api.create_derived_signal_calibration,
    }

Functions¶

list_records ¶

list_records(entity_name, **filters)

Return backend rows for a collection query.

Source code in src/owi/metadatabase/shm/services/core.py

def list_records(self, entity_name: ShmEntityName | str, **filters: Any) -> pd.DataFrame:
    """Return backend rows for a collection query."""
    resolved_name = self._resolve_name(entity_name)
    return cast(pd.DataFrame, self._list_methods[resolved_name](**filters)["data"])

get_record ¶

get_record(entity_name, **filters)

Return the raw backend response for a single-resource query.

Source code in src/owi/metadatabase/shm/services/core.py

def get_record(self, entity_name: ShmEntityName | str, **filters: Any) -> Mapping[str, Any]:
    """Return the raw backend response for a single-resource query."""
    resolved_name = self._resolve_name(entity_name)
    if resolved_name is ShmEntityName.SIGNAL and "signal_id" in filters:
        signal_id = str(filters.pop("signal_id"))
        return self.api.get_signal(signal_id, **filters)
    return self._get_methods[resolved_name](**filters)

create_record ¶

create_record(entity_name, payload, files=None)

Create a resource through the configured SHM API client.

Source code in src/owi/metadatabase/shm/services/core.py

def create_record(
    self,
    entity_name: ShmEntityName | str,
    payload: Mapping[str, Any],
    files: Mapping[str, Any] | None = None,
) -> Mapping[str, Any]:
    """Create a resource through the configured SHM API client."""
    resolved_name = self._resolve_name(entity_name)
    if files is not None:
        if resolved_name is ShmEntityName.SENSOR_TYPE:
            return self.api.create_sensor_type(dict(payload), files=files)
        if resolved_name is ShmEntityName.SENSOR_CALIBRATION:
            return self.api.create_sensor_calibration(dict(payload), files=files)
    create_method = self._create_methods[resolved_name]
    return create_method(dict(payload))

SensorService ¶

SensorService(entity_service=None)

Convenience service for sensor-domain SHM entities.

Source code in src/owi/metadatabase/shm/services/core.py

def __init__(self, entity_service: ShmEntityService | None = None) -> None:
    self.entity_service = entity_service or ShmEntityService()

ShmEntityService ¶

ShmEntityService(repository=None, registry=None)

Facade for typed SHM retrieval and creation.

Source code in src/owi/metadatabase/shm/services/core.py

def __init__(
    self,
    repository: ShmRepositoryProtocol | None = None,
    registry: EntityRegistryProtocol | None = None,
) -> None:
    self.repository = repository or ApiShmRepository()
    self.registry = registry or default_registry

Functions¶

list_records ¶

list_records(entity_name, filters=None)

Return typed resources for a collection query.

Source code in src/owi/metadatabase/shm/services/core.py

def list_records(
    self,
    entity_name: ShmEntityName | str,
    filters: ShmQuery | Mapping[str, Any] | None = None,
) -> list[ShmResourceRecord]:
    """Return typed resources for a collection query."""
    query = self._coerce_query(entity_name, filters)
    definition = self.registry.get(query.entity or self._resolve_name(entity_name))
    frame = self.repository.list_records(definition.name, **query.to_backend_filters())
    return [
        cast(
            ShmResourceRecord,
            definition.serializer.from_mapping(cast(dict[str, Any], row)),
        )
        for row in frame.to_dict(orient="records")
    ]

get_record ¶

get_record(entity_name, filters=None)

Return a single typed resource when it exists.

Source code in src/owi/metadatabase/shm/services/core.py

def get_record(
    self,
    entity_name: ShmEntityName | str,
    filters: ShmQuery | Mapping[str, Any] | None = None,
) -> ShmResourceRecord | None:
    """Return a single typed resource when it exists."""
    query = self._coerce_query(entity_name, filters)
    definition = self.registry.get(query.entity or self._resolve_name(entity_name))
    result = self.repository.get_record(definition.name, **query.to_backend_filters())
    frame = result.get("data")
    if not result.get("exists") or not isinstance(frame, pd.DataFrame) or frame.empty:
        return None
    return cast(ShmResourceRecord, definition.serializer.from_mapping(frame.iloc[0].to_dict()))

create_record ¶

create_record(entity_name, payload, files=None)

Create and deserialize one SHM resource.

Source code in src/owi/metadatabase/shm/services/core.py

def create_record(
    self,
    entity_name: ShmEntityName | str,
    payload: Mapping[str, Any] | ShmResourceRecord,
    files: Mapping[str, Any] | None = None,
) -> ShmResourceRecord | None:
    """Create and deserialize one SHM resource."""
    resolved_name = self._resolve_name(entity_name)
    definition = self.registry.get(resolved_name)
    serialized_payload = definition.serializer.to_payload(payload)
    result = self.repository.create_record(resolved_name, serialized_payload, files=files)
    frame = result.get("data")
    if not result.get("exists") or not isinstance(frame, pd.DataFrame) or frame.empty:
        return None
    return cast(ShmResourceRecord, definition.serializer.from_mapping(frame.iloc[0].to_dict()))

SignalService ¶

SignalService(entity_service=None)

Convenience service for signal-domain SHM entities.

Source code in src/owi/metadatabase/shm/services/core.py

def __init__(self, entity_service: ShmEntityService | None = None) -> None:
    self.entity_service = entity_service or ShmEntityService()