"""
Manager for interacting with target resources.
"""
__all__ = ["TargetManager"]
import logging
from pathlib import Path
from typing import Any, Optional
from gpp_client.api.custom_fields import (
CloneTargetResultFields,
CreateTargetResultFields,
DeclinationArcFields,
DeclinationFields,
OpportunityFields,
ProgramFields,
RegionFields,
RightAscensionArcFields,
RightAscensionFields,
SiderealFields,
TargetFields,
TargetSelectResultFields,
UpdateTargetsResultFields,
)
from gpp_client.api.custom_mutations import Mutation
from gpp_client.api.custom_queries import Query
from gpp_client.api.enums import Existence
from gpp_client.api.input_types import (
CloneTargetInput,
CreateTargetInput,
TargetPropertiesInput,
UpdateTargetsInput,
WhereOrderTargetId,
WhereTarget,
)
from gpp_client.managers.base import BaseManager
logger = logging.getLogger(__name__)
[docs]
class TargetManager(BaseManager):
"""
Manager for interacting with target resources.
"""
_OP_CLONE: str = "cloneTarget"
_OP_CREATE: str = "createTarget"
_OP_UPDATE: str = "updateTargets"
_OP_GET: str = "target"
_OP_LIST: str = "targets"
@staticmethod
def _build_where_for_id(*, target_id: str) -> WhereTarget:
"""Build a ``WhereTarget`` filter for a target ID."""
return WhereTarget(id=WhereOrderTargetId(eq=target_id))
[docs]
async def clone(
self,
*,
target_id: str,
properties: Optional[TargetPropertiesInput] = None,
from_json: Optional[str | Path | dict[str, Any]] = None,
replace_in: Optional[list[str]] = None,
) -> dict[str, Any]:
"""
Clone an existing target, optionally modifying some properties.
Parameters
----------
target_id : str
Unique identifier of the target to clone.
properties : TargetPropertiesInput, optional
Properties to override in the cloned target.
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.
replace_in : list[str], optional
List of observation IDs where the cloned target should replace the
original target.
Returns
-------
dict[str, Any]
The original and newly created target data.
Raises
------
GPPValidationError
If an error is encountered validating the input properties.
GPPClientError
If an unexpected error occurs unpacking the response.
"""
logger.debug("Cloning target from target with ID: %s", target_id)
properties = self.load_properties(
properties=properties, from_json=from_json, cls=TargetPropertiesInput
)
input_data = CloneTargetInput(
target_id=target_id,
set_=properties,
replace_in=replace_in,
)
fields = Mutation.clone_target(input=input_data).fields(
CloneTargetResultFields.new_target().fields(*self._fields()),
)
result = await self.client.mutation(fields, operation_name=self._OP_CLONE)
return self.get_result(result, self._OP_CLONE)
[docs]
async def create(
self,
*,
properties: Optional[TargetPropertiesInput] = None,
from_json: Optional[str | Path | dict[str, Any]] = None,
program_id: Optional[str] = None,
proposal_reference: Optional[str] = None,
program_reference: Optional[str] = None,
) -> dict[str, Any]:
"""
Create a new target using a fully defined ``TargetPropertiesInput``.
Parameters
----------
properties : TargetPropertiesInput, optional
Full target definition. 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
Program ID to associate with. Must be provided if neither
`proposal_reference` nor `program_reference` is given.
proposal_reference : str, optional
Reference to a proposal; used as an alternative to `program_id`.
program_reference : str, optional
Reference to a program label; used as an alternative to `program_id`.
Returns
-------
dict[str, Any]
The created target and its associated metadata.
Raises
------
GPPValidationError
If an error is encountered validating the input properties or identifiers.
GPPClientError
If an unexpected error occurs unpacking the response.
Notes
-----
At least one of `program_id`, `proposal_reference`, or `program_reference`
must be specified to associate with a valid program.
Exactly one of ``properties`` or ``from_json`` must be supplied. Supplying
both or neither raises ``GPPValidationError``.
"""
logger.debug("Creating a new target")
self.validate_single_identifier(
program_id=program_id,
proposal_reference=proposal_reference,
program_reference=program_reference,
)
properties = self.load_properties(
properties=properties, from_json=from_json, cls=TargetPropertiesInput
)
input_data = CreateTargetInput(
program_id=program_id,
proposal_reference=proposal_reference,
program_reference=program_reference,
set_=properties,
)
fields = Mutation.create_target(input=input_data).fields(
CreateTargetResultFields.target().fields(*self._fields()),
)
result = await self.client.mutation(fields, operation_name=self._OP_CREATE)
return self.get_result(result, self._OP_CREATE)
[docs]
async def update_all(
self,
*,
properties: Optional[TargetPropertiesInput] = None,
from_json: Optional[str | Path | dict[str, Any]] = None,
where: Optional[WhereTarget] = None,
limit: Optional[int] = None,
include_deleted: bool = False,
) -> dict[str, Any]:
"""
Update one or more targets using a partial or complete
``TargetPropertiesInput``.
Parameters
----------
properties : TargetPropertiesInput, optional
New values to apply to matching targets. 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 : WhereTarget, optional
Query filters to select which targets to update. If omitted, all targets
are eligible.
limit : int, optional
Maximum number of targets to update. If omitted, all matches are updated.
include_deleted : bool, default=False
Whether to include soft-deleted targets in the update.
Returns
-------
dict[str, Any]
A dictionary containing update results and the updated targets.
Raises
------
GPPValidationError
If an error is encountered validating the input properties.
GPPClientError
If an unexpected error occurs unpacking the response.
Notes
-----
Exactly one of ``properties`` or ``from_json`` must be supplied. Supplying
both or neither raises ``GPPValidationError``.
"""
logger.debug("Updating target(s)")
properties = self.load_properties(
properties=properties, from_json=from_json, cls=TargetPropertiesInput
)
input_data = UpdateTargetsInput(
set_=properties,
where=where,
limit=limit,
include_deleted=include_deleted,
)
fields = Mutation.update_targets(input=input_data).fields(
UpdateTargetsResultFields.has_more,
UpdateTargetsResultFields.targets().fields(
*self._fields(include_deleted=include_deleted)
),
)
result = await self.client.mutation(fields, operation_name=self._OP_UPDATE)
return self.get_result(result, self._OP_UPDATE)
[docs]
async def update_by_id(
self,
target_id: str,
*,
properties: Optional[TargetPropertiesInput] = None,
from_json: Optional[str | Path | dict[str, Any]] = None,
include_deleted: bool = False,
) -> dict[str, Any]:
"""
Update a single target by its unique identifier.
Parameters
----------
target_id : str
Unique identifier of the target. This or ``from_json`` must be
supplied.
properties : TargetPropertiesInput, optional
New values to apply to the target.
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 targets in the update.
Returns
-------
dict[str, Any]
Dictionary containing update result, including updated target data.
Raises
------
GPPValidationError
If an error is encountered validating the input properties.
GPPClientError
If an unexpected error occurs unpacking the response.
Notes
-----
Exactly one of ``properties`` or ``from_json`` must be supplied. Supplying
both or neither raises ``GPPValidationError``.
"""
logger.debug("Updating target with ID: %s", target_id)
where = self._build_where_for_id(target_id=target_id)
results = await self.update_all(
where=where,
limit=1,
properties=properties,
from_json=from_json,
include_deleted=include_deleted,
)
# Since it returns one item, discard the 'matches' and return the item.
return self.get_single_result(results, "targets")
[docs]
async def get_by_id(
self, target_id: str, *, include_deleted: bool = False
) -> dict[str, Any]:
"""
Fetch a single resource by ID.
Parameters
----------
target_id : str
Unique identifier of the target.
include_deleted : bool, default=False
Whether to include soft-deleted targets when retrieving.
Returns
-------
dict[str, Any]
Retrieved data.
Raises
------
GPPClientError
If an unexpected error occurs unpacking the response.
"""
fields = Query.target(target_id=target_id).fields(
*self._fields(include_deleted=include_deleted)
)
result = await self.client.query(fields, operation_name=self._OP_GET)
return self.get_result(result, self._OP_GET)
[docs]
async def get_all(
self,
*,
include_deleted: bool = False,
where: WhereTarget | None = None,
offset: int | None = None,
limit: int | None = None,
) -> dict[str, Any]:
"""
Get all targets with optional filtering and pagination.
Parameters
----------
include_deleted : bool, default=False
Whether to include deleted resources.
where : WhereTarget, optional
Optional filter criteria.
offset : int, optional
Cursor-based offset (by ID).
limit : int, optional
Maximum number of results.
Returns
-------
dict[str, Any]
Dictionary with `matches` and `hasMore` keys.
Raises
------
GPPClientError
If an unexpected error occurs unpacking the response.
"""
logger.debug("Fetching target(s)")
fields = Query.targets(
include_deleted=include_deleted, where=where, offset=offset, limit=limit
).fields(
TargetSelectResultFields.has_more,
TargetSelectResultFields.matches().fields(
*self._fields(include_deleted=include_deleted)
),
)
result = await self.client.query(fields, operation_name=self._OP_LIST)
return self.get_result(result, self._OP_LIST)
[docs]
async def restore_by_id(self, target_id: str) -> dict[str, Any]:
"""
Restore a soft-deleted resource by setting its existence to PRESENT.
Parameters
----------
target_id : str
Unique identifier of the target.
Returns
-------
dict[str, Any]
The restore result payload.
Raises
------
GPPValidationError
If a validation error occurs.
GPPClientError
If an unexpected error occurs unpacking the response.
"""
logger.debug("Restoring target with ID: %s", target_id)
properties = TargetPropertiesInput(existence=Existence.PRESENT)
return await self.update_by_id(
target_id, properties=properties, include_deleted=True
)
[docs]
async def delete_by_id(self, target_id: str) -> dict[str, Any]:
"""
Soft-delete a resource by setting its existence to DELETED.
Parameters
----------
target_id : str
Unique identifier of the target.
Returns
-------
dict[str, Any]
The delete result payload.
Raises
------
GPPValidationError
If a validation error occurs.
GPPClientError
If an unexpected error occurs unpacking the response.
"""
logger.debug("Deleting target with ID: %s", target_id)
properties = TargetPropertiesInput(existence=Existence.DELETED)
return await self.update_by_id(
target_id,
properties=properties,
include_deleted=False,
)
@staticmethod
def _fields(include_deleted: bool = False) -> tuple:
"""Generate the fields to return."""
return (
TargetFields.id,
TargetFields.existence,
TargetFields.name,
TargetFields.calibration_role,
TargetFields.program(include_deleted=include_deleted).fields(
ProgramFields.id,
ProgramFields.name,
ProgramFields.description,
ProgramFields.existence,
),
TargetFields.opportunity().fields(
OpportunityFields.region().fields(
RegionFields.right_ascension_arc().fields(
RightAscensionArcFields.start().fields(
RightAscensionFields.degrees,
),
RightAscensionArcFields.end().fields(
RightAscensionFields.degrees,
),
),
RegionFields.declination_arc().fields(
DeclinationArcFields.start().fields(
DeclinationFields.degrees,
),
DeclinationArcFields.end().fields(
DeclinationFields.degrees,
),
),
)
),
TargetFields.sidereal().fields(
SiderealFields.ra().fields(
RightAscensionFields.hours,
RightAscensionFields.hms,
RightAscensionFields.degrees,
),
SiderealFields.dec().fields(
DeclinationFields.degrees,
DeclinationFields.dms,
),
SiderealFields.epoch,
),
)