python(feat): add artifacts resource

python/changelog.d/845.feat.md

@@ -0,0 +1 @@
+Added `client.artifacts` for finding artifacts by the runs and assets they link to, browsing version history through `client.artifacts.versions`, and downloading any version's file.

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

@@ -0,0 +1,258 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any, cast
+
+from sift.artifacts.v1.artifacts_pb2 import (
+    ArchiveArtifactRequest,
+    CreateArtifactResponse,
+    GetArtifactRequest,
+    GetArtifactResponse,
+    ListArtifactsRequest,
+    ListArtifactsResponse,
+    ListArtifactVersionsRequest,
+    ListArtifactVersionsResponse,
+    PromoteArtifactVersionRequest,
+    PromoteArtifactVersionResponse,
+    UnarchiveArtifactRequest,
+    UpdateArtifactResponse,
+)
+from sift.artifacts.v1.artifacts_pb2_grpc import ArtifactServiceStub
+
+from sift_client._internal.low_level_wrappers.base import DEFAULT_PAGE_SIZE, LowLevelClientBase
+from sift_client.sift_types.artifact import (
+    Artifact,
+    ArtifactCreate,
+    ArtifactUpdate,
+    ArtifactVersion,
+)
+from sift_client.transport import WithGrpcClient
+
+if TYPE_CHECKING:
+    from sift_client.transport.grpc_transport import GrpcClient
+
+# The update mask path that declares new bytes for the version the update writes.
+REMOTE_FILE_ID_PATH = "artifact_version.remote_file_id"
+
+
+class ArtifactsLowLevelClient(LowLevelClientBase, WithGrpcClient):
+    """Low-level client for the ArtifactsAPI.
+
+    A thin wrapper around the autogenerated bindings for the ArtifactService.
+    """
+
+    def __init__(self, grpc_client: GrpcClient):
+        """Initialize the ArtifactsLowLevelClient.
+
+        Args:
+            grpc_client: The gRPC client to use for making API calls.
+        """
+        super().__init__(grpc_client)
+
+    async def create_artifact(self, create: ArtifactCreate) -> Artifact:
+        """Create an artifact and its first version.
+
+        Args:
+            create: The artifact to create.
+
+        Returns:
+            The new Artifact, resolved to version 1.
+        """
+        request = create.to_proto()
+        response = await self._grpc_client.get_stub(ArtifactServiceStub).CreateArtifact(request)
+        return Artifact._from_proto(cast("CreateArtifactResponse", response).artifact)
+
+    async def update_artifact(self, update: ArtifactUpdate, *, new_file: bool = False) -> Artifact:
+        """Write a new version that carries the fields the update sets.
+
+        Args:
+            update: The fields to write. Its resource ID names the artifact.
+            new_file: Declare that new bytes will be uploaded for the new version, so the
+                previous bytes and their SOURCE and DERIVED_FROM links do not carry forward.
+
+        Returns:
+            The Artifact, resolved to the new version.
+        """
+        request, mask = update.to_proto_with_mask()
+        if new_file:
+            mask.paths.append(REMOTE_FILE_ID_PATH)
+        request.update_mask.CopyFrom(mask)
+        response = await self._grpc_client.get_stub(ArtifactServiceStub).UpdateArtifact(request)
+        return Artifact._from_proto(cast("UpdateArtifactResponse", response).artifact)
+
+    async def get_artifact(
+        self, artifact_id: str, artifact_version_id: str | None = None
+    ) -> Artifact:
+        """Get an artifact, resolved to its latest version unless one is pinned.
+
+        Args:
+            artifact_id: The artifact to read.
+            artifact_version_id: The version to resolve to instead of the latest.
+
+        Returns:
+            The Artifact.
+        """
+        request = GetArtifactRequest(
+            artifact_id=artifact_id, artifact_version_id=artifact_version_id
+        )
+        response = await self._grpc_client.get_stub(ArtifactServiceStub).GetArtifact(request)
+        return Artifact._from_proto(cast("GetArtifactResponse", response).artifact)
+
+    async def list_artifacts(
+        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[Artifact], str]:
+        """List one page of artifacts, each resolved to its latest version.
+
+        Args:
+            page_size: The maximum number of artifacts to return.
+            page_token: A page token for pagination.
+            query_filter: A CEL filter string.
+            order_by: How to order the retrieved artifacts.
+
+        Returns:
+            A tuple of (artifacts, 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 = ListArtifactsRequest(**request_kwargs)
+        response = await self._grpc_client.get_stub(ArtifactServiceStub).ListArtifacts(request)
+        response = cast("ListArtifactsResponse", response)
+        return [Artifact._from_proto(a) for a in response.artifacts], response.next_page_token
+
+    async def list_all_artifacts(
+        self,
+        *,
+        query_filter: str | None = None,
+        order_by: str | None = None,
+        max_results: int | None = None,
+        page_size: int | None = DEFAULT_PAGE_SIZE,
+    ) -> list[Artifact]:
+        """List all artifacts matching the given query.
+
+        Args:
+            query_filter: A CEL filter string.
+            order_by: How to order the retrieved artifacts.
+            max_results: The maximum number of artifacts to return.
+            page_size: The number of artifacts to fetch per request.
+
+        Returns:
+            The matching Artifacts.
+        """
+        return await self._handle_pagination(
+            self.list_artifacts,
+            kwargs={"query_filter": query_filter},
+            page_size=page_size,
+            order_by=order_by,
+            max_results=max_results,
+        )
+
+    async def list_artifact_versions(
+        self,
+        *,
+        page_size: int | None = DEFAULT_PAGE_SIZE,
+        page_token: str | None = None,
+        query_filter: str | None = None,
+        order_by: str | None = None,
+        artifact_id: str | None = None,
+    ) -> tuple[list[ArtifactVersion], str]:
+        """List one page of artifact versions.
+
+        Args:
+            page_size: The maximum number of versions to return.
+            page_token: A page token for pagination.
+            query_filter: A CEL filter string, evaluated against each version.
+            order_by: How to order the retrieved versions.
+            artifact_id: Only this artifact's versions. Unset searches every artifact.
+
+        Returns:
+            A tuple of (versions, 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
+        if artifact_id is not None:
+            request_kwargs["artifact_id"] = artifact_id
+
+        request = ListArtifactVersionsRequest(**request_kwargs)
+        response = await self._grpc_client.get_stub(ArtifactServiceStub).ListArtifactVersions(
+            request
+        )
+        response = cast("ListArtifactVersionsResponse", response)
+        return [ArtifactVersion._from_proto(v) for v in response.versions], response.next_page_token
+
+    async def list_all_artifact_versions(
+        self,
+        *,
+        query_filter: str | None = None,
+        order_by: str | None = None,
+        artifact_id: str | None = None,
+        max_results: int | None = None,
+        page_size: int | None = DEFAULT_PAGE_SIZE,
+    ) -> list[ArtifactVersion]:
+        """List all artifact versions matching the given query.
+
+        Args:
+            query_filter: A CEL filter string, evaluated against each version.
+            order_by: How to order the retrieved versions.
+            artifact_id: Only this artifact's versions. Unset searches every artifact.
+            max_results: The maximum number of versions to return.
+            page_size: The number of versions to fetch per request.
+
+        Returns:
+            The matching ArtifactVersions.
+        """
+        return await self._handle_pagination(
+            self.list_artifact_versions,
+            kwargs={"query_filter": query_filter, "artifact_id": artifact_id},
+            page_size=page_size,
+            order_by=order_by,
+            max_results=max_results,
+        )
+
+    async def promote_artifact_version(
+        self, artifact_id: str, artifact_version_id: str
+    ) -> Artifact:
+        """Write a new latest version with the content of an earlier version.
+
+        Args:
+            artifact_id: The artifact that owns the version.
+            artifact_version_id: The version whose content to promote.
+
+        Returns:
+            The Artifact, resolved to the version the promotion wrote.
+        """
+        request = PromoteArtifactVersionRequest(
+            artifact_id=artifact_id, artifact_version_id=artifact_version_id
+        )
+        response = await self._grpc_client.get_stub(ArtifactServiceStub).PromoteArtifactVersion(
+            request
+        )
+        return Artifact._from_proto(cast("PromoteArtifactVersionResponse", response).artifact)
+
+    async def archive_artifact(self, artifact_id: str) -> None:
+        """Archive an artifact."""
+        request = ArchiveArtifactRequest(artifact_id=artifact_id)
+        await self._grpc_client.get_stub(ArtifactServiceStub).ArchiveArtifact(request)
+
+    async def unarchive_artifact(self, artifact_id: str) -> None:
+        """Unarchive an artifact."""
+        request = UnarchiveArtifactRequest(artifact_id=artifact_id)
+        await self._grpc_client.get_stub(ArtifactServiceStub).UnarchiveArtifact(request)

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

@@ -2,8 +2,9 @@
 
 import logging
 import mimetypes
+from contextlib import nullcontext
 from pathlib import Path
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, BinaryIO, ContextManager
 
 from requests_toolbelt import MultipartEncoder
 
@@ -40,7 +41,7 @@ class UploadLowLevelClient(LowLevelClientBase, WithRestClient):
 
         # Upload a file
         remote_file_id = await upload_client.upload_attachment(
-            path="path/to/file.mp4",
+            source="path/to/file.mp4",
             entity_id="run_12345",
             entity_type="runs",
             description="Video of test run",
@@ -61,48 +62,50 @@ def __init__(self, rest_client: RestClient):
 
     async def upload_attachment(
         self,
-        path: str | Path,
+        source: str | Path | BinaryIO,
         entity_id: str,
         entity_type: str,
         metadata: dict[str, Any] | None = None,
         description: str | None = None,
         organization_id: str | None = None,
+        file_name: str | None = None,
     ) -> str:
         """Upload a file attachment to an entity.
 
         Args:
-            path: Path to the file to upload.
+            source: Path to the file to upload, or an open binary file such as `io.BytesIO`.
             entity_id: The ID of the entity to attach the file to.
             entity_type: The type of entity (e.g., "runs", "annotations", "annotation_logs").
             metadata: Optional metadata for the file (e.g., video/image metadata).
             description: Optional description of the file.
             organization_id: Optional organization ID.
+            file_name: The stored file name when `source` is a file object; its extension sets
+                the MIME type. A path upload uses the file's own name.
 
         Returns:
             The remote file ID of the uploaded file.
 
         Raises:
-            ValueError: If the path doesn't point to a regular file or MIME type cannot be determined.
+            ValueError: If the path doesn't point to a regular file, or a file object has no
+                file_name.
             Exception: If the upload fails.
         """
-        posix_path = Path(path) if isinstance(path, str) else path
-
-        if not posix_path.is_file():
-            raise ValueError(f"Provided path, '{path}', does not point to a regular file.")
-
-        file_name, mimetype, content_encoding = self._mime_and_content_type_from_path(posix_path)
-
-        if not mimetype:
-            extension = posix_path.suffix
-            if extension:
-                mimetype = f"application/x-{extension.lstrip('.')}"
-            else:
-                mimetype = "application/octet-stream"  # fallback to generic 'binary data' MIME type
+        upload: Path | BinaryIO
+        if isinstance(source, (str, Path)):
+            upload = Path(source)
+            if not upload.is_file():
+                raise ValueError(f"Provided path, '{source}', does not point to a regular file.")
+            file_name, mimetype, content_encoding = self._resolve_mime(upload)
+        else:
+            if not file_name:
+                raise ValueError("file_name is required when uploading a file object.")
+            upload = source
+            _, mimetype, content_encoding = self._resolve_mime(Path(file_name))
 
         # Run the synchronous file upload in a thread pool to avoid blocking the event loop
         return await run_sync_function(
             self._upload_file_sync,
-            posix_path,
+            upload,
             file_name,
             mimetype,
             content_encoding,
@@ -115,7 +118,7 @@ async def upload_attachment(
 
     def _upload_file_sync(
         self,
-        path: Path,
+        source: Path | BinaryIO,
         file_name: str,
         mimetype: str,
         content_encoding: str | None,
@@ -129,7 +132,12 @@ def _upload_file_sync(
 
         This is called from a thread pool to avoid blocking the async event loop.
         """
-        with open(path, "rb") as file:
+        opened: ContextManager[BinaryIO]
+        if isinstance(source, Path):
+            opened = open(source, "rb")
+        else:
+            opened = nullcontext(source)
+        with opened as file:
             form_fields: dict[str, Any] = {
                 "entityId": entity_id,
                 "entityType": entity_type,
@@ -179,6 +187,18 @@ def _upload_file_sync(
             response_data = response.json()
             return response_data.get("remoteFile", {}).get("remoteFileId")
 
+    @classmethod
+    def _resolve_mime(cls, path: Path) -> tuple[str, str, str | None]:
+        """Like `_mime_and_content_type_from_path`, with a fallback when the type is unknown."""
+        file_name, mimetype, content_encoding = cls._mime_and_content_type_from_path(path)
+        if not mimetype:
+            extension = path.suffix
+            if extension:
+                mimetype = f"application/x-{extension.lstrip('.')}"
+            else:
+                mimetype = "application/octet-stream"  # fallback to generic 'binary data' MIME type
+        return file_name, mimetype, content_encoding
+
     @staticmethod
     def _mime_and_content_type_from_path(path: Path) -> tuple[str, str | None, str | None]:
         """Determine the MIME type and content encoding from a file path.