From 5ca890ac1a556b2759c3ba74f264612ed7acbb27 Mon Sep 17 00:00:00 2001 From: Wei Qi Lu Date: Fri, 6 Mar 2026 18:31:59 -0800 Subject: [PATCH] python(feature): sift_client low level wrapper for exports --- .../_internal/low_level_wrappers/exports.py | 65 +++++++++++++++++++ 1 file changed, 65 insertions(+) create mode 100644 python/lib/sift_client/_internal/low_level_wrappers/exports.py diff --git a/python/lib/sift_client/_internal/low_level_wrappers/exports.py b/python/lib/sift_client/_internal/low_level_wrappers/exports.py new file mode 100644 index 000000000..9c1c43970 --- /dev/null +++ b/python/lib/sift_client/_internal/low_level_wrappers/exports.py @@ -0,0 +1,65 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING, cast + +from sift.exports.v1.exports_pb2 import ( + ExportDataRequest, + ExportDataResponse, + GetDownloadUrlRequest, + GetDownloadUrlResponse, +) +from sift.exports.v1.exports_pb2_grpc import ExportServiceStub + +from sift_client._internal.low_level_wrappers.base import LowLevelClientBase +from sift_client.transport import WithGrpcClient + +if TYPE_CHECKING: + from sift_client.transport.grpc_transport import GrpcClient + + +class ExportsLowLevelClient(LowLevelClientBase, WithGrpcClient): + """Low-level client for the ExportsAPI. + + This class provides a thin wrapper around the autogenerated gRPC bindings for the ExportsAPI. + """ + + def __init__(self, grpc_client: GrpcClient): + """Initialize the ExportsLowLevelClient. + + Args: + grpc_client: The gRPC client to use for making API calls. + """ + super().__init__(grpc_client) + + async def export_data(self, request: ExportDataRequest) -> ExportDataResponse: + """Initiate a data export. + + Returns a presigned_url if the export completes immediately, or a job_id + if it's processed in the background. Use get_download_url() to retrieve + the URL for background jobs. + + Args: + request: The ExportDataRequest proto message. + + Returns: + The ExportDataResponse containing either a presigned_url or a job_id. + """ + response = await self._grpc_client.get_stub(ExportServiceStub).ExportData(request) + return cast("ExportDataResponse", response) + + async def get_download_url(self, job_id: str) -> str: + """Get the download URL for a background export job. + + If the job is still processing, the server will return an error. + Polling/retry logic should be handled. + + Args: + job_id: The job ID returned from export_data(). + + Returns: + The presigned URL to download the exported zip file. + """ + request = GetDownloadUrlRequest(job_id=job_id) + response = await self._grpc_client.get_stub(ExportServiceStub).GetDownloadUrl(request) + response = cast("GetDownloadUrlResponse", response) + return response.presigned_url