Observation

Manager for interacting with observation resources.

class gpp_client.managers.observation.ObservationManager(client: GPPClient)[source]

Bases: BaseManager

Manager for interacting with observation resources.

async clone(*, observation_id: str | None = None, observation_reference: str | None = None, properties: ObservationPropertiesInput | None = None, from_json: str | Path | dict[str, Any] | None = None) → dict[str, Any][source]

Clone an existing observation to create a new one.

Parameters:

  • observation_id (str, optional) – Unique internal ID of the observation to clone.

  • observation_reference (str, optional) – Human-readable reference label (e.g., “G-2025A-1234-Q-0001”) of the observation to clone.

  • properties (ObservationPropertiesInput, optional) – Properties to override in the cloned observation. This or from_json may be supplied.

  • from_json (str | Path | dict[str, Any], optional) – JSON representation of the properties. May be a path-like object (str or Path) to a JSON file, or a dict already containing the JSON data.

Returns:

A dictionary containing details of the original and new cloned observations.

Return type:

dict[str, Any]

Raises:

Notes

Exactly one of observation_id or observation_reference must be provided to identify the observation to clone. Additionally, either properties or from_json may be supplied to specify overrides for the cloned observation.

async create(*, properties: ObservationPropertiesInput | None = None, from_json: str | Path | dict[str, Any] | None = None, program_id: str | None = None, proposal_reference: str | None = None, program_reference: str | None = None) → dict[str, Any][source]

Create a new observation under a specified program.

Parameters:

  • properties (ObservationPropertiesInput, optional) – Observation definition to use in creation. This or from_json must be supplied.

  • from_json (str | Path | dict[str, Any], optional) – JSON representation of the properties. May be a path-like object (str or Path) to a JSON file, or a dict already containing the JSON data.

  • program_id (str, optional) – Direct program identifier. Must be provided if proposal_reference and program_reference are omitted.

  • proposal_reference (str, optional) – Proposal label alternative to program_id.

  • program_reference (str, optional) – Program label alternative to program_id.

Returns:

The created observation and its metadata.

Return type:

dict[str, Any]

Raises:

Notes

Exactly one of properties or from_json must be supplied. Supplying both or neither raises GPPValidationError.

async delete_by_id(*, observation_id: str | None = None, observation_reference: str | None = None) → dict[str, Any][source]

Soft-delete an observation using ID or reference.

Parameters:

  • observation_id (str, optional) – Unique internal ID of the observation to delete.

  • observation_reference (str, optional) – Human-readable reference label (e.g., “G-2025A-1234-Q-0001”).

Returns:

The deleted observation with existence set to DELETED.

Return type:

dict[str, Any]

Raises:
async get_all(*, include_deleted: bool = False, where: WhereObservation | None = None, offset: int | None = None, limit: int | None = None) → dict[str, Any][source]

Retrieve all observations with optional filters.

Parameters:

  • include_deleted (bool, default=False) – Whether to include soft-deleted observations.

  • where (WhereObservation, optional) – Filter criteria.

  • offset (int, optional) – Cursor offset (by ID).

  • limit (int, optional) – Maximum number of observations.

Returns:

A dictionary with the results.

Return type:

dict[str, Any]

Raises:

GPPClientError – If an unexpected error occurs unpacking the response.

async get_by_id(*, observation_id: str | None = None, observation_reference: str | None = None, include_deleted: bool = False) → dict[str, Any][source]

Fetch a single observation by ID or reference.

This method retrieves a single observation using either the internal ID or the reference label. Exactly one of observation_id or observation_reference must be provided.

Parameters:

  • observation_id (str, optional) – The unique internal identifier of the observation.

  • observation_reference (str, optional) – The human-readable reference label (e.g., “G-2024B-1234-Q-0001”).

  • include_deleted (bool, default=False) – Whether to include soft-deleted observations in the query.

Returns:

The retrieved observation.

Return type:

dict[str, Any]

Raises:
get_result(result: dict[str, Any] | None, operation_name: str | None = None) → dict[str, Any]

Extract the payload for a given GraphQL operation.

Parameters:

  • result (dict[str, Any] | None) – The full GraphQL result.

  • operation_name (str | None, optional) – The name of the operation to extract. If None, and the result contains exactly one top-level key, that key is used.

Returns:

The extracted payload for the specified operation.

Return type:

dict[str, Any]

Raises:

GPPClientError – If the result is empty or invalid, or if the specified operation name is not found in the result.

get_single_result(payload: dict[str, Any], key: str) → dict[str, Any]

Extract exactly one item from a list-valued field in a GraphQL payload.

Parameters:

  • payload (dict[str, Any]) – The GraphQL payload containing the field.

  • key (str) – The key of the field to extract.

Returns:

The single item extracted from the field.

Return type:

dict[str, Any]

Raises:

GPPClientError – If the specified field is missing from the payload, is not a list, or does not contain exactly one item.

load_properties(*, properties: Any | None, from_json: str | Path | dict[str, Any] | None, cls: type[T]) → T

Return a validated properties object from exactly one data source.

Parameters:

  • properties (T, optional) – Preconstructed properties instance. Returned unchanged when provided.

  • from_json (str | Path | dict[str, Any], optional) – Path to a JSON file or a dictionary containing the JSON data.

  • cls (type[T]) – Concrete PropertiesInput class for validation. Required.

Returns:

Instance of cls representing the validated properties.

Return type:

T

Raises:

GPPValidationError – If validation fails or an error occurs loading the properties.

raise_error(exc_class: type[GPPError], exc: Exception, *, include_traceback: bool = False) → NoReturn

Raise a structured exception with contextual information.

Parameters:

  • exc_class (type[GPPError]) – The exception class to raise (must accept a string message).

  • exc (Exception) – The original exception to wrap.

  • include_traceback (bool, default=False) – Whether to include the original traceback using from exc.

Raises:

type[GPPError] – The raised exception of the specified type.

async raise_for_status(response, *, ok_statuses: set[int], default_message: str = 'Request failed') → None

Raise an exception if the HTTP response status is not OK.

Parameters:

  • response (aiohttp.ClientResponse) – The HTTP response to check.

  • ok_statuses (set[int]) – Set of acceptable HTTP status codes.

  • default_message (str, default="Request failed") – Default error message if the response has no content.

Raises:

GPPResponseError – If the response status is not in ok_statuses.

resolve_content(*, file_path: str | Path | None, content: bytes | None) → bytes

Resolve upload content bytes from exactly one source.

Parameters:

  • file_path (str | Path | None) – Path to a local file. If provided, it must exist and be a file.

  • content (bytes | None) – Raw bytes content.

Returns:

The resolved content bytes.

Return type:

bytes

Raises:

  • GPPValidationError – If both or neither of file_path and content are provided, or if file_path is invalid.

  • GPPClientError – If reading the file fails due to an unexpected I/O error.

async restore_by_id(*, observation_id: str | None = None, observation_reference: str | None = None) → dict[str, Any][source]

Restore a soft-deleted observation using ID or reference.

Parameters:

  • observation_id (str, optional) – Unique internal ID of the observation to restore.

  • observation_reference (str, optional) – Human-readable reference label (e.g., “G-2025A-1234-Q-0001”).

Returns:

The restored observation with existence set to PRESENT.

Return type:

dict[str, Any]

Raises:
async update_all(*, properties: ObservationPropertiesInput | None = None, from_json: str | Path | dict[str, Any] | None = None, where: WhereObservation | None = None, limit: int | None = None, include_deleted: bool = False) → dict[str, Any][source]

Update one or more observations with new properties.

Parameters:

  • properties (ObservationPropertiesInput, optional) – Fields to update. This or from_json must be supplied.

  • from_json (str | Path | dict[str, Any], optional) – JSON representation of the properties. May be a path-like object (str or Path) to a JSON file, or a dict already containing the JSON data.

  • where (WhereObservation, optional) – Filter expression to limit which observations are updated.

  • limit (int, optional) – Maximum number of observations to update.

  • include_deleted (bool, default=False) – Whether to include soft-deleted observations.

Returns:

The update result and updated records.

Return type:

dict[str, Any]

Raises:

Notes

Exactly one of properties or from_json must be supplied. Supplying both or neither raises GPPValidationError.

async update_by_id(*, observation_id: str | None = None, observation_reference: str | None = None, properties: ObservationPropertiesInput | None = None, from_json: str | Path | dict[str, Any] | None = None, include_deleted: bool = False) → dict[str, Any][source]

Update a single observation by ID or reference.

Parameters:

  • observation_id (str, optional) – Unique internal ID of the observation.

  • observation_reference (str, optional) – Human-readable reference label (e.g., “G-2025A-1234-Q-0001”).

  • properties (ObservationPropertiesInput, optional) – New values to apply to the observation. This or from_json must be supplied.

  • from_json (str | Path | dict[str, Any], optional) – JSON representation of the properties. May be a path-like object (str or Path) to a JSON file, or a dict already containing the JSON data.

  • include_deleted (bool, default=False) – Whether to include soft-deleted observations in the match.

Returns:

The updated observation.

Return type:

dict[str, Any]

Raises:

Notes

Exactly one of properties or from_json must be supplied. Supplying both or neither raises GPPValidationError.

validate_single_identifier(**kwargs) → None

Validate that exactly one identifier is provided.

This helper checks that exactly one of the provided keyword arguments is non-None. It raises a ValueError otherwise.

Parameters:

**kwargs (dict[str, Optional[str]]) – A dictionary of identifier keyword arguments to validate.

Raises:

GPPValidationError – If none or more than one identifiers are provided.