python(feat): annotations resource

python/CHANGELOG.md

@@ -5,6 +5,8 @@ This project adheres to [Semantic Versioning](http://semver.org/).
 
 ## [Unreleased]
 
+- Added `client.annotations` for data reviews, phases, and review history.
+
 ### What's New
 
 #### MCAP imports

python/lib/sift_client/_internal/low_level_wrappers/annotations.py

@@ -0,0 +1,352 @@
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING, Any, cast
+
+from sift.annotation_logs.v1.annotation_logs_pb2 import (
+    AnnotationLogAssignedProperties,
+    AnnotationLogCommentProperties,
+    AnnotationLogStateUpdateProperties,
+    CreateAnnotationLogRequest,
+    CreateAnnotationLogResponse,
+    ListAnnotationLogsRequest,
+    ListAnnotationLogsResponse,
+)
+from sift.annotation_logs.v1.annotation_logs_pb2_grpc import AnnotationLogServiceStub
+from sift.annotations.v1.annotations_pb2 import (
+    ArchiveAnnotationRequest,
+    ArchiveAnnotationResponse,
+    BatchArchiveAnnotationsRequest,
+    BatchArchiveAnnotationsResponse,
+    BatchUnarchiveAnnotationsRequest,
+    BatchUnarchiveAnnotationsResponse,
+    CreateAnnotationResponse,
+    GetAnnotationRequest,
+    GetAnnotationResponse,
+    ListAnnotationsRequest,
+    ListAnnotationsResponse,
+    UnarchiveAnnotationRequest,
+    UnarchiveAnnotationResponse,
+    UpdateAnnotationRequest,
+    UpdateAnnotationResponse,
+)
+from sift.annotations.v1.annotations_pb2_grpc import AnnotationServiceStub
+
+from sift_client._internal.low_level_wrappers.base import DEFAULT_PAGE_SIZE, LowLevelClientBase
+from sift_client.sift_types.annotation import (
+    Annotation,
+    AnnotationCommentElement,
+    AnnotationCreateBase,
+    AnnotationLog,
+    AnnotationLogKind,
+    AnnotationLogState,
+    AnnotationUpdate,
+)
+from sift_client.transport import WithGrpcClient
+
+if TYPE_CHECKING:
+    from sift_client.transport.grpc_transport import GrpcClient
+
+# Configure logging
+logger = logging.getLogger(__name__)
+
+
+class AnnotationsLowLevelClient(LowLevelClientBase, WithGrpcClient):
+    """Low-level client for the AnnotationsAPI.
+
+    A thin wrapper around the autogenerated bindings for the AnnotationsAPI and the
+    AnnotationLogsAPI.
+    """
+
+    def __init__(self, grpc_client: GrpcClient):
+        """Initialize the AnnotationsLowLevelClient.
+
+        Args:
+            grpc_client: The gRPC client to use for making API calls.
+        """
+        super().__init__(grpc_client)
+
+    async def get_annotation(self, annotation_id: str) -> Annotation:
+        """Get an annotation by annotation_id.
+
+        Args:
+            annotation_id: The annotation ID to get.
+
+        Returns:
+            The Annotation.
+        """
+        request = GetAnnotationRequest(annotation_id=annotation_id)
+        response = await self._grpc_client.get_stub(AnnotationServiceStub).GetAnnotation(request)
+        return Annotation._from_proto(cast("GetAnnotationResponse", response).annotation)
+
+    async def list_annotations(
+        self,
+        *,
+        page_size: int | None = DEFAULT_PAGE_SIZE,
+        page_token: str | None = None,
+        query_filter: str | None = None,
+        order_by: str | None = None,
+    ) -> tuple[list[Annotation], str]:
+        """List one page of annotations.
+
+        Args:
+            page_size: The maximum number of annotations to return.
+            page_token: A page token for pagination.
+            query_filter: A CEL filter string.
+            order_by: How to order the retrieved annotations.
+
+        Returns:
+            A tuple of (annotations, next_page_token).
+        """
+        request_kwargs: dict[str, Any] = {}
+        if page_size is not None:
+            request_kwargs["page_size"] = page_size
+        if page_token is not None:
+            request_kwargs["page_token"] = page_token
+        if query_filter is not None:
+            request_kwargs["filter"] = query_filter
+        if order_by is not None:
+            request_kwargs["order_by"] = order_by
+
+        request = ListAnnotationsRequest(**request_kwargs)
+        response = await self._grpc_client.get_stub(AnnotationServiceStub).ListAnnotations(request)
+        response = cast("ListAnnotationsResponse", response)
+
+        annotations = [Annotation._from_proto(a) for a in response.annotations]
+        return annotations, response.next_page_token
+
+    async def list_all_annotations(
+        self,
+        *,
+        query_filter: str | None = None,
+        order_by: str | None = None,
+        page_size: int | None = DEFAULT_PAGE_SIZE,
+        max_results: int | None = None,
+    ) -> list[Annotation]:
+        """List every matching annotation.
+
+        Args:
+            query_filter: A CEL filter string.
+            order_by: How to order the retrieved annotations.
+            page_size: The number of results to fetch per request.
+            max_results: Maximum number of results to return.
+
+        Returns:
+            A list of all matching annotations.
+        """
+        return await self._handle_pagination(
+            self.list_annotations,
+            kwargs={"query_filter": query_filter},
+            order_by=order_by,
+            max_results=max_results,
+            page_size=page_size,
+        )
+
+    async def create_annotation(self, *, create: AnnotationCreateBase) -> Annotation:
+        """Create an annotation.
+
+        Args:
+            create: The annotation definition.
+
+        Returns:
+            The created Annotation.
+        """
+        request_proto = create.to_proto()
+        request_proto.linked_channels.extend(create.linked_channels_to_proto())
+        response = await self._grpc_client.get_stub(AnnotationServiceStub).CreateAnnotation(
+            request_proto
+        )
+        return Annotation._from_proto(cast("CreateAnnotationResponse", response).annotation)
+
+    async def update_annotation(self, update: AnnotationUpdate) -> Annotation:
+        """Update select fields of an annotation.
+
+        Args:
+            update: The updates to apply. Its `resource_id` must be set.
+
+        Returns:
+            The updated Annotation.
+        """
+        grpc_annotation, update_mask = update.to_proto_with_mask()
+        if "linked_channels" in update.model_fields_set:
+            grpc_annotation.linked_channels.extend(update.linked_channels_to_proto())
+            update_mask.paths.append("linked_channels")
+        request = UpdateAnnotationRequest(annotation=grpc_annotation, update_mask=update_mask)
+        response = await self._grpc_client.get_stub(AnnotationServiceStub).UpdateAnnotation(request)
+        return Annotation._from_proto(cast("UpdateAnnotationResponse", response).annotation)
+
+    async def archive_annotation(self, annotation_id: str) -> Annotation:
+        """Archive an annotation.
+
+        Args:
+            annotation_id: The ID of the annotation to archive.
+
+        Returns:
+            The archived Annotation.
+        """
+        request = ArchiveAnnotationRequest(annotation_id=annotation_id)
+        response = await self._grpc_client.get_stub(AnnotationServiceStub).ArchiveAnnotation(
+            request
+        )
+        return Annotation._from_proto(cast("ArchiveAnnotationResponse", response).annotation)
+
+    async def unarchive_annotation(self, annotation_id: str) -> Annotation:
+        """Unarchive an annotation.
+
+        Args:
+            annotation_id: The ID of the annotation to unarchive.
+
+        Returns:
+            The unarchived Annotation.
+        """
+        request = UnarchiveAnnotationRequest(annotation_id=annotation_id)
+        response = await self._grpc_client.get_stub(AnnotationServiceStub).UnarchiveAnnotation(
+            request
+        )
+        return Annotation._from_proto(cast("UnarchiveAnnotationResponse", response).annotation)
+
+    async def batch_archive_annotations(self, annotation_ids: list[str]) -> list[Annotation]:
+        """Archive many annotations in one call.
+
+        Args:
+            annotation_ids: The IDs of the annotations to archive.
+
+        Returns:
+            The archived Annotations.
+        """
+        request = BatchArchiveAnnotationsRequest(annotation_ids=annotation_ids)
+        response = await self._grpc_client.get_stub(AnnotationServiceStub).BatchArchiveAnnotations(
+            request
+        )
+        response = cast("BatchArchiveAnnotationsResponse", response)
+        return [Annotation._from_proto(a) for a in response.annotations]
+
+    async def batch_unarchive_annotations(self, annotation_ids: list[str]) -> list[Annotation]:
+        """Unarchive many annotations in one call.
+
+        Args:
+            annotation_ids: The IDs of the annotations to unarchive.
+
+        Returns:
+            The unarchived Annotations.
+        """
+        request = BatchUnarchiveAnnotationsRequest(annotation_ids=annotation_ids)
+        response = await self._grpc_client.get_stub(
+            AnnotationServiceStub
+        ).BatchUnarchiveAnnotations(request)
+        response = cast("BatchUnarchiveAnnotationsResponse", response)
+        return [Annotation._from_proto(a) for a in response.annotations]
+
+    async def list_annotation_logs(
+        self,
+        *,
+        annotation_id: str | None = None,
+        page_size: int | None = DEFAULT_PAGE_SIZE,
+        page_token: str | None = None,
+        query_filter: str | None = None,
+        order_by: str | None = None,
+    ) -> tuple[list[AnnotationLog], str]:
+        """List one page of annotation logs.
+
+        Args:
+            annotation_id: Restrict results to this annotation.
+            page_size: The maximum number of logs to return.
+            page_token: A page token for pagination.
+            query_filter: A CEL filter string.
+            order_by: How to order the retrieved logs.
+
+        Returns:
+            A tuple of (logs, next_page_token).
+        """
+        request_kwargs: dict[str, Any] = {}
+        if annotation_id is not None:
+            request_kwargs["annotation_id"] = annotation_id
+        if page_size is not None:
+            request_kwargs["page_size"] = page_size
+        if page_token is not None:
+            request_kwargs["page_token"] = page_token
+        if query_filter is not None:
+            request_kwargs["filter"] = query_filter
+        if order_by is not None:
+            request_kwargs["order_by"] = order_by
+
+        request = ListAnnotationLogsRequest(**request_kwargs)
+        response = await self._grpc_client.get_stub(AnnotationLogServiceStub).ListAnnotationLogs(
+            request
+        )
+        response = cast("ListAnnotationLogsResponse", response)
+
+        logs = [AnnotationLog._from_proto(log) for log in response.annotation_logs]
+        return logs, response.next_page_token
+
+    async def list_all_annotation_logs(
+        self,
+        *,
+        annotation_id: str | None = None,
+        query_filter: str | None = None,
+        order_by: str | None = None,
+        page_size: int | None = DEFAULT_PAGE_SIZE,
+        max_results: int | None = None,
+    ) -> list[AnnotationLog]:
+        """List every matching annotation log.
+
+        Args:
+            annotation_id: Restrict results to this annotation.
+            query_filter: A CEL filter string.
+            order_by: How to order the retrieved logs.
+            page_size: The number of results to fetch per request.
+            max_results: Maximum number of results to return.
+
+        Returns:
+            A list of all matching logs.
+        """
+        return await self._handle_pagination(
+            self.list_annotation_logs,
+            kwargs={"annotation_id": annotation_id, "query_filter": query_filter},
+            order_by=order_by,
+            max_results=max_results,
+            page_size=page_size,
+        )
+
+    async def create_annotation_log(
+        self,
+        *,
+        annotation_id: str,
+        kind: AnnotationLogKind,
+        assigned_to_user_id: str | None = None,
+        state: AnnotationLogState | None = None,
+        comment: list[AnnotationCommentElement] | None = None,
+    ) -> AnnotationLog:
+        """Create an annotation log.
+
+        Args:
+            annotation_id: The annotation to attach the log to.
+            kind: Which kind of log to create.
+            assigned_to_user_id: The assignee, for ASSIGNED logs.
+            state: The new state, for STATE_UPDATE logs.
+            comment: The comment body, for COMMENT logs.
+
+        Returns:
+            The created AnnotationLog.
+        """
+        request_kwargs: dict[str, Any] = {"annotation_id": annotation_id, "kind": kind.value}
+        if kind is AnnotationLogKind.ASSIGNED:
+            request_kwargs["assigned"] = AnnotationLogAssignedProperties(
+                assigned_to_user_id=assigned_to_user_id or ""
+            )
+        elif kind is AnnotationLogKind.STATE_UPDATE:
+            request_kwargs["state_update"] = AnnotationLogStateUpdateProperties(
+                state=state.value if state else 0  # type: ignore[arg-type]
+            )
+        elif kind is AnnotationLogKind.COMMENT:
+            request_kwargs["comment"] = AnnotationLogCommentProperties(
+                body=[e._to_proto() for e in comment or []]
+            )
+
+        request = CreateAnnotationLogRequest(**request_kwargs)
+        response = await self._grpc_client.get_stub(AnnotationLogServiceStub).CreateAnnotationLog(
+            request
+        )
+        return AnnotationLog._from_proto(
+            cast("CreateAnnotationLogResponse", response).annotation_log
+        )

python/lib/sift_client/_tests/conftest.py

@@ -79,6 +79,7 @@ def mock_client():
     """Create a mock SiftClient for unit testing."""
     client = MagicMock(spec=SiftClient)
     # Configure the mock to have the necessary API attributes
+    client.annotations = MagicMock()
     client.assets = MagicMock()
     client.reports = MagicMock()
     client.runs = MagicMock()