diff --git a/python/CHANGELOG.md b/python/CHANGELOG.md index b568ef5131..55abf79acc 100644 --- a/python/CHANGELOG.md +++ b/python/CHANGELOG.md @@ -5,6 +5,23 @@ This project adheres to [Semantic Versioning](http://semver.org/). ## [Unreleased] +### What's New + +#### List and get data imports + +New in `client.data_import`: `list_` and `get` (plus `find`), and a `run.data_imports` property. + +```python +# Imports for a run, e.g. to check on a batch +imports = client.data_import.list_(runs=[run]) +imports = run.data_imports # same query, keyed off the run + +# One import by ID +data_import = client.data_import.get("data-import-id") +``` + +Each `DataImport` carries the import's `status`, `error_message`, and `warning_messages`. `job.get_data_import()` returns it for the `Job` that `import_from_path` hands back. + ## [v0.20.0] - August 25, 2026 ### What's New diff --git a/python/lib/sift_client/_internal/low_level_wrappers/data_imports.py b/python/lib/sift_client/_internal/low_level_wrappers/data_imports.py index cd430a773d..4ce2d8bdbd 100644 --- a/python/lib/sift_client/_internal/low_level_wrappers/data_imports.py +++ b/python/lib/sift_client/_internal/low_level_wrappers/data_imports.py @@ -1,6 +1,6 @@ from __future__ import annotations -from typing import TYPE_CHECKING, cast +from typing import TYPE_CHECKING, Any, cast from sift.data_imports.v2.data_imports_pb2 import ( CreateDataImportFromUploadRequest, @@ -9,12 +9,15 @@ DetectConfigResponse, GetDataImportRequest, GetDataImportResponse, + ListDataImportsRequest, + ListDataImportsResponse, ) from sift.data_imports.v2.data_imports_pb2_grpc import DataImportServiceStub -from sift_client._internal.low_level_wrappers.base import LowLevelClientBase +from sift_client._internal.low_level_wrappers.base import DEFAULT_PAGE_SIZE, LowLevelClientBase from sift_client.sift_types.data_import import ( CsvImportConfig, + DataImport, Hdf5ImportConfig, ImportConfig, ParquetFlatDatasetImportConfig, @@ -77,18 +80,82 @@ async def create_from_upload(self, config: ImportConfig) -> tuple[str, str]: response = cast("CreateDataImportFromUploadResponse", response) return response.data_import_id, response.upload_url - async def get(self, data_import_id: str) -> GetDataImportResponse: + async def get(self, data_import_id: str) -> DataImport: """Get a data import by ID. Args: data_import_id: The ID of the data import. Returns: - The GetDataImportResponse proto. + The DataImport. """ request = GetDataImportRequest(data_import_id=data_import_id) response = await self._grpc_client.get_stub(DataImportServiceStub).GetDataImport(request) - return cast("GetDataImportResponse", response) + response = cast("GetDataImportResponse", response) + return DataImport._from_proto(response.data_import) + + async def list_data_imports( + 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[DataImport], str]: + """List data imports with optional filtering and pagination. + + Args: + page_size: The maximum number of data imports to return. + page_token: A page token from a previous list call. + query_filter: A CEL filter string. + order_by: How to order the retrieved data imports. + + Returns: + A tuple of (data_imports, 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 = ListDataImportsRequest(**request_kwargs) + response = await self._grpc_client.get_stub(DataImportServiceStub).ListDataImports(request) + response = cast("ListDataImportsResponse", response) + + data_imports = [DataImport._from_proto(di) for di in response.data_imports] + return data_imports, response.next_page_token + + async def list_all_data_imports( + self, + *, + query_filter: str | None = None, + order_by: str | None = None, + page_size: int | None = DEFAULT_PAGE_SIZE, + max_results: int | None = None, + ) -> list[DataImport]: + """List all data imports, handling pagination automatically. + + Args: + query_filter: A CEL filter string. + order_by: How to order the retrieved data imports. + page_size: The number of results to fetch per request. + max_results: Maximum number of results to return across all pages. + + Returns: + A list of all matching data imports. + """ + return await self._handle_pagination( + self.list_data_imports, + kwargs={"query_filter": query_filter}, + order_by=order_by, + max_results=max_results, + page_size=page_size, + ) async def detect_config( self, data: bytes, data_type_key: DataTypeKey.ValueType diff --git a/python/lib/sift_client/_tests/conftest.py b/python/lib/sift_client/_tests/conftest.py index 125f846aaf..2272cc2998 100644 --- a/python/lib/sift_client/_tests/conftest.py +++ b/python/lib/sift_client/_tests/conftest.py @@ -74,6 +74,8 @@ def mock_client(): client.test_results = MagicMock() client.file_attachments = MagicMock() client.jobs = MagicMock() + client.data_import = MagicMock() + client.data_export = MagicMock() client.async_ = MagicMock(spec=AsyncAPIs) client.async_.ingestion = MagicMock() return client diff --git a/python/lib/sift_client/_tests/resources/test_data_imports.py b/python/lib/sift_client/_tests/resources/test_data_imports.py index c960437a4c..0be5d898eb 100644 --- a/python/lib/sift_client/_tests/resources/test_data_imports.py +++ b/python/lib/sift_client/_tests/resources/test_data_imports.py @@ -4,15 +4,20 @@ from datetime import datetime, timezone from typing import TYPE_CHECKING, cast +from unittest.mock import AsyncMock, MagicMock, patch import pytest from sift.common.type.v1.channel_config_pb2 import ChannelConfig as ChannelConfigProto from sift.data_imports.v2.data_imports_pb2 import ( + DATA_IMPORT_STATUS_FAILED, + DATA_IMPORT_STATUS_SUCCEEDED, + DATA_IMPORT_STATUS_UNSPECIFIED, ParquetColumn, ParquetConfig, ParquetFlatDatasetConfig, ParquetSingleChannelPerRowConfig, ) +from sift.data_imports.v2.data_imports_pb2 import DataImport as DataImportProto from sift.data_imports.v2.data_imports_pb2 import ( ParquetDataColumn as ParquetDataColumnProto, ) @@ -31,6 +36,8 @@ CsvDataColumn, CsvImportConfig, CsvTimeColumn, + DataImport, + DataImportStatus, DataTypeKey, Hdf5DataColumn, Hdf5ImportConfig, @@ -44,11 +51,16 @@ UlogImportConfig, UlogParseErrorPolicy, ) +from sift_client.sift_types.job import Job, JobStatus, JobType +from sift_client.sift_types.run import Run if TYPE_CHECKING: from sift.common.type.v1.channel_data_type_pb2 import ( ChannelDataType as ChannelDataTypeProto, ) + from sift.data_imports.v2.data_imports_pb2 import ( + DataImportStatus as DataImportStatusProto, + ) @pytest.mark.integration @@ -685,3 +697,265 @@ def test_scpr_infers_int64_time_column(self): config = _parse_parquet_detect_response(proto, "file.parquet", 0, 0) assert isinstance(config, ParquetSingleChannelPerRowImportConfig) assert config.time_column.path == "timestamp" + + +def _data_import_proto( + data_import_id: str = "data-import-1", + status: DataImportStatusProto.ValueType = DATA_IMPORT_STATUS_SUCCEEDED, +) -> DataImportProto: + proto = DataImportProto( + data_import_id=data_import_id, + source_url="https://example.com/data.csv", + status=status, + warning_messages=["skipped 3 records"], + run_id="run-1", + report_id="report-1", + asset_id="asset-1", + ) + proto.created_date.FromDatetime(datetime(2026, 1, 1, tzinfo=timezone.utc)) + proto.modified_date.FromDatetime(datetime(2026, 1, 2, tzinfo=timezone.utc)) + proto.data_start_time.FromDatetime(datetime(2025, 12, 31, tzinfo=timezone.utc)) + proto.data_stop_time.FromDatetime(datetime(2025, 12, 31, 1, tzinfo=timezone.utc)) + return proto + + +def _data_imports_api(mock_client) -> DataImportAPIAsync: + api = DataImportAPIAsync(mock_client) + api._low_level_client = MagicMock() + return api + + +class TestDataImportStatus: + def test_unspecified_maps_instead_of_raising(self): + # A status the server never set must not break list_() parsing. + assert ( + DataImportStatus.from_proto(DATA_IMPORT_STATUS_UNSPECIFIED) + == DataImportStatus.UNSPECIFIED + ) + + def test_unknown_future_value_falls_back_to_unspecified(self): + assert DataImportStatus.from_proto(100) == DataImportStatus.UNSPECIFIED + + +class TestDataImportFromProto: + def test_all_fields(self): + data_import = DataImport._from_proto(_data_import_proto()) + + assert data_import.id_ == "data-import-1" + assert data_import.source_url == "https://example.com/data.csv" + assert data_import.status == DataImportStatus.SUCCEEDED + assert data_import.error_message is None + assert data_import.warning_messages == ["skipped 3 records"] + assert data_import.created_date == datetime(2026, 1, 1, tzinfo=timezone.utc) + assert data_import.modified_date == datetime(2026, 1, 2, tzinfo=timezone.utc) + assert data_import.run_id == "run-1" + assert data_import.report_id == "report-1" + assert data_import.asset_id == "asset-1" + assert data_import.data_start_time == datetime(2025, 12, 31, tzinfo=timezone.utc) + assert data_import.data_stop_time == datetime(2025, 12, 31, 1, tzinfo=timezone.utc) + + def test_unset_optionals_are_none(self): + proto = DataImportProto( + data_import_id="data-import-2", + status=DATA_IMPORT_STATUS_FAILED, + error_message="boom", + ) + data_import = DataImport._from_proto(proto) + + assert data_import.source_url is None + assert data_import.error_message == "boom" + assert data_import.warning_messages == [] + assert data_import.run_id is None + assert data_import.report_id is None + assert data_import.asset_id is None + assert data_import.data_start_time is None + assert data_import.data_stop_time is None + + +class TestDataImportsGet: + @pytest.mark.asyncio + async def test_returns_data_import(self, mock_client): + api = _data_imports_api(mock_client) + api._low_level_client.get = AsyncMock( + return_value=DataImport._from_proto(_data_import_proto()) + ) + + data_import = await api.get("data-import-1") + + api._low_level_client.get.assert_awaited_once_with("data-import-1") + assert isinstance(data_import, DataImport) + assert data_import.id_ == "data-import-1" + assert data_import.client is mock_client + + +class TestDataImportsList: + @pytest.mark.asyncio + async def test_no_filters_sends_no_query(self, mock_client): + api = _data_imports_api(mock_client) + api._low_level_client.list_all_data_imports = AsyncMock( + return_value=[DataImport._from_proto(_data_import_proto())] + ) + + data_imports = await api.list_() + + api._low_level_client.list_all_data_imports.assert_awaited_once_with( + query_filter=None, order_by=None, max_results=None + ) + assert data_imports[0].client is mock_client + + @pytest.mark.asyncio + async def test_builds_cel_filter(self, mock_client): + api = _data_imports_api(mock_client) + api._low_level_client.list_all_data_imports = AsyncMock(return_value=[]) + + await api.list_( + data_import_ids=["di-1", "di-2"], + source_url_contains="s3://bucket", + status=DataImportStatus.FAILED, + runs=["run-1"], + ) + + query_filter = api._low_level_client.list_all_data_imports.await_args.kwargs["query_filter"] + assert query_filter == ( + "data_import_id in ['di-1','di-2'] && " + "source_url.contains('s3://bucket') && " + "status == 'DATA_IMPORT_STATUS_FAILED' && " + "run_id in ['run-1']" + ) + + @pytest.mark.asyncio + async def test_accepts_run_objects(self, mock_client): + api = _data_imports_api(mock_client) + api._low_level_client.list_all_data_imports = AsyncMock(return_value=[]) + run = MagicMock(spec=Run) + run._id_or_error = "run-42" + + await api.list_(runs=[run]) + + assert ( + api._low_level_client.list_all_data_imports.await_args.kwargs["query_filter"] + == "run_id in ['run-42']" + ) + + @pytest.mark.asyncio + async def test_forwards_pagination_and_ordering(self, mock_client): + api = _data_imports_api(mock_client) + api._low_level_client.list_all_data_imports = AsyncMock(return_value=[]) + + await api.list_(order_by="created_date desc", limit=5, page_size=2) + + api._low_level_client.list_all_data_imports.assert_awaited_once_with( + query_filter=None, order_by="created_date desc", max_results=5, page_size=2 + ) + + +class TestDataImportsFind: + @pytest.mark.asyncio + async def test_raises_on_multiple_matches(self, mock_client): + api = _data_imports_api(mock_client) + api._low_level_client.list_all_data_imports = AsyncMock( + return_value=[ + DataImport._from_proto(_data_import_proto("di-1")), + DataImport._from_proto(_data_import_proto("di-2")), + ] + ) + + with pytest.raises(ValueError, match="Multiple data imports found"): + await api.find() + + +class TestDataImportsGetRun: + @pytest.mark.asyncio + async def test_resolves_run_id(self, mock_client): + api = _data_imports_api(mock_client) + api._low_level_client.get = AsyncMock( + return_value=DataImport._from_proto(_data_import_proto()) + ) + run = MagicMock(spec=Run) + mock_client.async_.runs.get = AsyncMock(return_value=run) + + assert await api.get_run("data-import-1") is run + mock_client.async_.runs.get.assert_awaited_once_with(run_id="run-1") + + @pytest.mark.asyncio + async def test_raises_without_run(self, mock_client): + api = _data_imports_api(mock_client) + proto = _data_import_proto() + proto.ClearField("run_id") + api._low_level_client.get = AsyncMock(return_value=DataImport._from_proto(proto)) + + with pytest.raises(ValueError, match="does not have an associated run"): + await api.get_run("data-import-1") + + +class TestImportFromPathDataImportId: + """import_from_path must hand the returned job a way back to the import.""" + + @staticmethod + def _job(job_details): + return Job( + proto=MagicMock(), + id_="job-1", + organization_id="org-1", + created_by_user_id="user-1", + modified_by_user_id="user-1", + created_date=datetime(2026, 1, 1, tzinfo=timezone.utc), + modified_date=datetime(2026, 1, 1, tzinfo=timezone.utc), + started_date=None, + completed_date=None, + job_type=JobType.DATA_IMPORT, + job_status=JobStatus.RUNNING, + job_status_details=None, + job_details=job_details, + ) + + async def _run_import(self, mock_client, tmp_path): + api = _data_imports_api(mock_client) + api._low_level_client.create_from_upload = AsyncMock( + return_value=("di-from-create", "https://upload.example/put") + ) + job = self._job(job_details=None) + job._apply_client_to_instance(mock_client) + mock_client.async_.jobs.get = AsyncMock(return_value=job) + + csv_path = tmp_path / "data.csv" + csv_path.write_text("time,value\n") + config = CsvImportConfig( + asset_name="test_asset", + time_column=CsvTimeColumn(column=1, format=TimeFormat.ABSOLUTE_RFC3339), + data_columns=[CsvDataColumn(column=2, name="value", data_type=ChannelDataType.DOUBLE)], + ) + + with patch( + "sift_client.resources.data_imports.upload_file", + return_value={"jobId": "job-1"}, + ): + return await api.import_from_path(csv_path, config=config, show_progress=False) + + @pytest.mark.asyncio + async def test_records_id_when_job_details_missing(self, mock_client, tmp_path): + """The create call's ID reaches the job even when the server hasn't set details.""" + job = await self._run_import(mock_client, tmp_path) + + job.get_data_import() + + mock_client.data_import.get.assert_called_once_with("di-from-create") + + @pytest.mark.asyncio + async def test_missing_job_id_raises(self, mock_client, tmp_path): + """A response without a job ID is still an error.""" + api = _data_imports_api(mock_client) + api._low_level_client.create_from_upload = AsyncMock( + return_value=("di-from-create", "https://upload.example/put") + ) + csv_path = tmp_path / "data.csv" + csv_path.write_text("time,value\n") + config = CsvImportConfig( + asset_name="test_asset", + time_column=CsvTimeColumn(column=1, format=TimeFormat.ABSOLUTE_RFC3339), + data_columns=[CsvDataColumn(column=2, name="value", data_type=ChannelDataType.DOUBLE)], + ) + + with patch("sift_client.resources.data_imports.upload_file", return_value={}): + with pytest.raises(RuntimeError, match="did not include a job ID"): + await api.import_from_path(csv_path, config=config, show_progress=False) diff --git a/python/lib/sift_client/_tests/sift_types/test_job.py b/python/lib/sift_client/_tests/sift_types/test_job.py index 847514fdf6..9f9b74b7c5 100644 --- a/python/lib/sift_client/_tests/sift_types/test_job.py +++ b/python/lib/sift_client/_tests/sift_types/test_job.py @@ -214,6 +214,84 @@ def test_retry_calls_client_and_updates_self(self, mock_failed_job, mock_client) assert result is mock_failed_job +class TestJobDataImportAccessors: + """Unit tests for the data-import accessors on Job.""" + + def test_get_data_import_calls_client(self, mock_job, mock_client): + """get_data_import() looks the import up by the ID in job_details.""" + data_import = MagicMock() + mock_client.data_import.get.return_value = data_import + + assert mock_job.get_data_import() is data_import + mock_client.data_import.get.assert_called_once_with("import123") + + def test_get_import_run_calls_client(self, mock_job, mock_client): + """get_import_run() resolves the run through the data import API.""" + run = MagicMock() + mock_client.data_import.get_run.return_value = run + + assert mock_job.get_import_run() is run + mock_client.data_import.get_run.assert_called_once_with("import123") + + @pytest.mark.parametrize("method", ["get_data_import", "get_import_run"]) + def test_rejects_non_import_jobs(self, mock_job, method): + """Both accessors refuse jobs that are not data imports.""" + object.__setattr__(mock_job, "job_type", JobType.DATA_EXPORT) + + with pytest.raises(ValueError, match=f"{method}\\(\\) is only valid for data import"): + getattr(mock_job, method)() + + @pytest.mark.parametrize("method", ["get_data_import", "get_import_run"]) + def test_rejects_missing_details(self, mock_job, method): + """Both accessors refuse a data import job with no import details.""" + object.__setattr__(mock_job, "job_details", None) + + with pytest.raises(ValueError, match="does not have data import details"): + getattr(mock_job, method)() + + def test_falls_back_to_recorded_id_when_details_unset(self, mock_job, mock_client): + """The ID recorded at import time covers a job whose details aren't set yet.""" + object.__setattr__(mock_job, "job_details", None) + mock_job._set_data_import_id("import-from-create") + + mock_job.get_data_import() + + mock_client.data_import.get.assert_called_once_with("import-from-create") + + def test_server_details_win_over_recorded_id(self, mock_job, mock_client): + """Once the server reports the import details, they are authoritative.""" + mock_job._set_data_import_id("import-from-create") + + mock_job.get_data_import() + + mock_client.data_import.get.assert_called_once_with("import123") + + def test_recorded_id_survives_update(self, mock_job, mock_client): + """A refresh that still lacks details must not lose the recorded ID.""" + mock_job._set_data_import_id("import-from-create") + refreshed = Job( + proto=MagicMock(), + id_=mock_job.id_, + organization_id="org1", + created_by_user_id="user1", + modified_by_user_id="user1", + created_date=mock_job.created_date, + modified_date=datetime.now(timezone.utc), + started_date=None, + completed_date=None, + job_type=JobType.DATA_IMPORT, + job_status=JobStatus.RUNNING, + job_status_details=None, + job_details=None, + ) + + mock_job._update(refreshed) + + assert mock_job.job_details is None + mock_job.get_data_import() + mock_client.data_import.get.assert_called_once_with("import-from-create") + + class TestJobType: """Unit tests for JobType enum.""" diff --git a/python/lib/sift_client/_tests/sift_types/test_run.py b/python/lib/sift_client/_tests/sift_types/test_run.py index 64c6b8cad3..58695efdf1 100644 --- a/python/lib/sift_client/_tests/sift_types/test_run.py +++ b/python/lib/sift_client/_tests/sift_types/test_run.py @@ -144,6 +144,14 @@ def test_assets_property_calls_client(self, mock_run, mock_client): # Verify client method was called with correct asset_ids mock_client.assets.list_.assert_called_once_with(asset_ids=["asset1", "asset2"]) + def test_data_imports_property_calls_client(self, mock_run, mock_client): + """Test that data_imports property filters by this run's ID.""" + mock_client.data_import.list_.return_value = [] + + _ = mock_run.data_imports + + mock_client.data_import.list_.assert_called_once_with(runs=["test_run_id"]) + def test_archive_calls_client_and_updates_self(self, mock_run, mock_client): """Test that archive() calls client.runs.archive and calls _update.""" archived_run = MagicMock() diff --git a/python/lib/sift_client/resources/data_imports.py b/python/lib/sift_client/resources/data_imports.py index c2a19c0970..794d03b5a9 100644 --- a/python/lib/sift_client/resources/data_imports.py +++ b/python/lib/sift_client/resources/data_imports.py @@ -13,6 +13,7 @@ DATA_TYPE_KEY_TO_PROTO, EXTENSION_TO_DATA_TYPE_KEY, CsvImportConfig, + DataImport, DataTypeKey, Hdf5ImportConfig, ImportConfig, @@ -24,11 +25,13 @@ UlogImportConfig, ) from sift_client.sift_types.run import Run +from sift_client.util import cel_utils as cel if TYPE_CHECKING: from collections.abc import Iterable from sift_client.client import SiftClient + from sift_client.sift_types.data_import import DataImportStatus from sift_client.sift_types.job import Job @@ -76,14 +79,14 @@ async def import_from_path( Examples: Import a CSV file with auto-detected config: - job = client.data_imports.import_from_path( + job = client.data_import.import_from_path( "data.csv", asset=my_asset, ) Auto-detect config, inspect and patch before importing: - config = client.data_imports.detect_config("data.csv") + config = client.data_import.detect_config("data.csv") # Fix a column data type config["temperature"].data_type = ChannelDataType.FLOAT @@ -93,7 +96,7 @@ async def import_from_path( dc for dc in config.data_columns if dc.name != "internal_id" ] - job = client.data_imports.import_from_path( + job = client.data_import.import_from_path( "data.csv", asset=my_asset, config=config, @@ -127,7 +130,9 @@ async def import_from_path( Defaults to True for sync, False for async. Returns: - A ``Job`` handle for the pending import. + A ``Job`` handle for the pending import. Call + ``job.get_data_import()`` on it for the import's status, error + message, and warnings, which are not on the job. Raises: FileNotFoundError: If the file does not exist. @@ -170,7 +175,7 @@ async def import_from_path( if show_progress is None: show_progress = self._show_progress() - _, upload_url = await self._low_level_client.create_from_upload(config) + data_import_id, upload_url = await self._low_level_client.create_from_upload(config) response = await run_sync_function( lambda: upload_file( @@ -184,7 +189,110 @@ async def import_from_path( if not job_id: raise RuntimeError("Upload succeeded but server response did not include a job ID.") - return await self.client.async_.jobs.get(job_id=job_id) + job = await self.client.async_.jobs.get(job_id=job_id) + # The job's details are filled in server-side and may not be set yet. + job._set_data_import_id(data_import_id) + return job + + async def get(self, data_import_id: str) -> DataImport: + """Get a data import by ID. + + The ``data_import_id`` is available on the job returned by + ``import_from_path`` via ``job.job_details.data_import_id``. + For a more ergonomic approach, use ``job.get_data_import()`` + which calls this method internally. + + Args: + data_import_id: The ID of the data import. + + Returns: + The DataImport. + """ + data_import = await self._low_level_client.get(data_import_id) + return self._apply_client_to_instance(data_import) + + async def list_( + self, + *, + # Self ids + data_import_ids: list[str] | None = None, + # Resource-specific filters + source_url: str | None = None, + source_url_contains: str | None = None, + status: DataImportStatus | None = None, + runs: list[Run | str] | None = None, + # Common filters + filter_query: str | None = None, + # Ordering and pagination + order_by: str | None = None, + limit: int | None = None, + page_size: int | None = None, + ) -> list[DataImport]: + """List data imports with optional filtering. + + The server only supports filtering on ``data_import_id``, + ``source_url``, ``status``, and ``run_id``, and only supports + ordering by ``created_date`` and ``modified_date``. + + Args: + data_import_ids: Filter to data imports with any of these IDs. + source_url: Filter to data imports with exactly this source url. + source_url_contains: Filter to data imports whose source url + contains this substring. + status: Filter to data imports with this status. + runs: Filter to data imports that ingested into any of these Runs + or run IDs. + filter_query: Explicit CEL query to filter data imports. + order_by: Field and direction to order results by, e.g. + ``"created_date desc"``. Defaults to oldest-first by + ``created_date``. + limit: Maximum number of data imports to return. If None, returns + all matches. + page_size: Number of results to fetch per request. Lower this if you hit gRPC + message size limits on responses. If None, uses the server default. + + Returns: + A list of DataImport objects that match the filter criteria. + """ + filter_parts = [*self._build_common_cel_filters(filter_query=filter_query)] + if data_import_ids: + filter_parts.append(cel.in_("data_import_id", data_import_ids)) + if source_url: + filter_parts.append(cel.equals("source_url", source_url)) + if source_url_contains: + filter_parts.append(cel.contains("source_url", source_url_contains)) + if status: + filter_parts.append(cel.equals("status", status.to_filter_str())) + if runs: + run_ids = [r._id_or_error if isinstance(r, Run) else r for r in runs] + filter_parts.append(cel.in_("run_id", run_ids)) + + query_filter = cel.and_(*filter_parts) + + data_imports = await self._low_level_client.list_all_data_imports( + query_filter=query_filter or None, + order_by=order_by, + max_results=limit, + **({"page_size": page_size} if page_size is not None else {}), + ) + return self._apply_client_to_instances(data_imports) + + async def find(self, **kwargs) -> DataImport | None: + """Find a single data import matching the given query. Takes the same arguments as + `list_`. If more than one data import is found, raises an error. + + Args: + **kwargs: Keyword arguments to pass to `list_`. + + Returns: + The DataImport found or None. + """ + data_imports = await self.list_(**kwargs) + if len(data_imports) > 1: + raise ValueError("Multiple data imports found for query") + elif len(data_imports) == 1: + return data_imports[0] + return None async def get_run(self, data_import_id: str) -> Run: """Get the run associated with a data import. @@ -192,7 +300,8 @@ async def get_run(self, data_import_id: str) -> Run: The ``data_import_id`` is available on the job returned by ``import_from_path`` via ``job.job_details.data_import_id``. For a more ergonomic approach, use ``job.get_import_run()`` - which calls this method internally. + which calls this method internally, or ``job.get_data_import()`` + followed by the import's ``run_id``. Args: data_import_id: The ID of the data import. @@ -203,11 +312,10 @@ async def get_run(self, data_import_id: str) -> Run: Raises: ValueError: If the data import has no associated run. """ - response = await self._low_level_client.get(data_import_id) - run_id = response.data_import.run_id - if not run_id: + data_import = await self._low_level_client.get(data_import_id) + if not data_import.run_id: raise ValueError("Data import does not have an associated run.") - return await self.client.async_.runs.get(run_id=run_id) + return await self.client.async_.runs.get(run_id=data_import.run_id) async def detect_config( self, diff --git a/python/lib/sift_client/resources/sync_stubs/__init__.pyi b/python/lib/sift_client/resources/sync_stubs/__init__.pyi index e92fbd35d5..a030f42069 100644 --- a/python/lib/sift_client/resources/sync_stubs/__init__.pyi +++ b/python/lib/sift_client/resources/sync_stubs/__init__.pyi @@ -25,6 +25,8 @@ if TYPE_CHECKING: ) from sift_client.sift_types.channel import Channel, ChannelUpdate from sift_client.sift_types.data_import import ( + DataImport, + DataImportStatus, DataTypeKey, ImportConfig, TimeFormat, @@ -769,13 +771,42 @@ class DataImportAPI: """ ... + def find(self, **kwargs) -> DataImport | None: + """Find a single data import matching the given query. Takes the same arguments as + `list_`. If more than one data import is found, raises an error. + + Args: + **kwargs: Keyword arguments to pass to `list_`. + + Returns: + The DataImport found or None. + """ + ... + + def get(self, data_import_id: str) -> DataImport: + """Get a data import by ID. + + The ``data_import_id`` is available on the job returned by + ``import_from_path`` via ``job.job_details.data_import_id``. + For a more ergonomic approach, use ``job.get_data_import()`` + which calls this method internally. + + Args: + data_import_id: The ID of the data import. + + Returns: + The DataImport. + """ + ... + def get_run(self, data_import_id: str) -> Run: """Get the run associated with a data import. The ``data_import_id`` is available on the job returned by ``import_from_path`` via ``job.job_details.data_import_id``. For a more ergonomic approach, use ``job.get_import_run()`` - which calls this method internally. + which calls this method internally, or ``job.get_data_import()`` + followed by the import's ``run_id``. Args: data_import_id: The ID of the data import. @@ -818,14 +849,14 @@ class DataImportAPI: Examples: Import a CSV file with auto-detected config: - job = client.data_imports.import_from_path( + job = client.data_import.import_from_path( "data.csv", asset=my_asset, ) Auto-detect config, inspect and patch before importing: - config = client.data_imports.detect_config("data.csv") + config = client.data_import.detect_config("data.csv") # Fix a column data type config["temperature"].data_type = ChannelDataType.FLOAT @@ -835,7 +866,7 @@ class DataImportAPI: dc for dc in config.data_columns if dc.name != "internal_id" ] - job = client.data_imports.import_from_path( + job = client.data_import.import_from_path( "data.csv", asset=my_asset, config=config, @@ -869,13 +900,56 @@ class DataImportAPI: Defaults to True for sync, False for async. Returns: - A ``Job`` handle for the pending import. + A ``Job`` handle for the pending import. Call + ``job.get_data_import()`` on it for the import's status, error + message, and warnings, which are not on the job. Raises: FileNotFoundError: If the file does not exist. """ ... + def list_( + self, + *, + data_import_ids: list[str] | None = None, + source_url: str | None = None, + source_url_contains: str | None = None, + status: DataImportStatus | None = None, + runs: list[Run | str] | None = None, + filter_query: str | None = None, + order_by: str | None = None, + limit: int | None = None, + page_size: int | None = None, + ) -> list[DataImport]: + """List data imports with optional filtering. + + The server only supports filtering on ``data_import_id``, + ``source_url``, ``status``, and ``run_id``, and only supports + ordering by ``created_date`` and ``modified_date``. + + Args: + data_import_ids: Filter to data imports with any of these IDs. + source_url: Filter to data imports with exactly this source url. + source_url_contains: Filter to data imports whose source url + contains this substring. + status: Filter to data imports with this status. + runs: Filter to data imports that ingested into any of these Runs + or run IDs. + filter_query: Explicit CEL query to filter data imports. + order_by: Field and direction to order results by, e.g. + ``"created_date desc"``. Defaults to oldest-first by + ``created_date``. + limit: Maximum number of data imports to return. If None, returns + all matches. + page_size: Number of results to fetch per request. Lower this if you hit gRPC + message size limits on responses. If None, uses the server default. + + Returns: + A list of DataImport objects that match the filter criteria. + """ + ... + class FileAttachmentsAPI: """Sync counterpart to `FileAttachmentsAPIAsync`. diff --git a/python/lib/sift_client/sift_types/__init__.py b/python/lib/sift_client/sift_types/__init__.py index 5cb05a2ae4..45589c9f0e 100644 --- a/python/lib/sift_client/sift_types/__init__.py +++ b/python/lib/sift_client/sift_types/__init__.py @@ -144,6 +144,7 @@ ChannelReference, ChannelUpdate, ) +from sift_client.sift_types.data_import import DataImport, DataImportStatus from sift_client.sift_types.ingestion import ( ChannelConfig, Flow, @@ -246,7 +247,9 @@ "ChannelUpdate", "DataExportDetails", "DataExportStatusDetails", + "DataImport", "DataImportDetails", + "DataImportStatus", "DataImportStatusDetails", "Flow", "FlowConfig", diff --git a/python/lib/sift_client/sift_types/data_import.py b/python/lib/sift_client/sift_types/data_import.py index 870596b331..94292e69bb 100644 --- a/python/lib/sift_client/sift_types/data_import.py +++ b/python/lib/sift_client/sift_types/data_import.py @@ -1,9 +1,9 @@ from __future__ import annotations from abc import ABC -from datetime import datetime # noqa: TC003 +from datetime import datetime, timezone from enum import Enum -from typing import Union +from typing import TYPE_CHECKING, Union from pydantic import BaseModel, model_validator from sift.common.type.v1.channel_config_pb2 import ChannelConfig as ChannelConfigProto @@ -30,6 +30,8 @@ ) from sift.data_imports.v2.data_imports_pb2 import CsvConfig as CsvConfigProto from sift.data_imports.v2.data_imports_pb2 import CsvTimeColumn as CsvTimeColumnProto +from sift.data_imports.v2.data_imports_pb2 import DataImport as DataImportProto +from sift.data_imports.v2.data_imports_pb2 import DataImportStatus as DataImportStatusProto from sift.data_imports.v2.data_imports_pb2 import Hdf5Config as Hdf5ConfigProto from sift.data_imports.v2.data_imports_pb2 import Hdf5DataConfig as Hdf5DataConfigProto from sift.data_imports.v2.data_imports_pb2 import ParquetConfig as ParquetConfigProto @@ -54,8 +56,12 @@ from sift.data_imports.v2.data_imports_pb2 import UlogDataConfig as UlogDataConfigProto from sift_client._internal.util.timestamp import to_pb_timestamp +from sift_client.sift_types._base import BaseType from sift_client.sift_types.channel import ChannelDataType +if TYPE_CHECKING: + from sift_client.client import SiftClient + class TimeFormat(Enum): """Supported time formats for data import columns.""" @@ -1016,3 +1022,91 @@ def _from_proto(cls, proto: UlogConfigProto) -> UlogImportConfig: Hdf5ImportConfig, UlogImportConfig, ] + + +class DataImportStatus(str, Enum): + """Status of a data import.""" + + UNSPECIFIED = "UNSPECIFIED" + PENDING = "PENDING" + IN_PROGRESS = "IN_PROGRESS" + SUCCEEDED = "SUCCEEDED" + FAILED = "FAILED" + + def to_filter_str(self) -> str: + """Convert to the string used in CEL filters.""" + return f"DATA_IMPORT_STATUS_{self.value}" + + @classmethod + def from_proto(cls, proto_value: int) -> DataImportStatus: + """Create from proto enum value. Unknown values map to UNSPECIFIED.""" + mapping: dict[int, DataImportStatus] = { + DataImportStatusProto.DATA_IMPORT_STATUS_PENDING: DataImportStatus.PENDING, + DataImportStatusProto.DATA_IMPORT_STATUS_IN_PROGRESS: DataImportStatus.IN_PROGRESS, + DataImportStatusProto.DATA_IMPORT_STATUS_SUCCEEDED: DataImportStatus.SUCCEEDED, + DataImportStatusProto.DATA_IMPORT_STATUS_FAILED: DataImportStatus.FAILED, + } + return mapping.get(proto_value, DataImportStatus.UNSPECIFIED) + + +class DataImport(BaseType[DataImportProto, "DataImport"]): + """Model of a Sift data import. + + The format-specific import config is not modeled here; read it off + ``data_import.proto`` (e.g. ``data_import.proto.csv_config``). + """ + + source_url: str | None + status: DataImportStatus + error_message: str | None + # What the import skipped. A succeeding import can still have warnings. + warning_messages: list[str] + created_date: datetime + modified_date: datetime + # Set once the run, report, and asset exist. + run_id: str | None + report_id: str | None + asset_id: str | None + # Time range of the imported data, not of the import itself. + data_start_time: datetime | None + data_stop_time: datetime | None + + @classmethod + def _from_proto( + cls, proto: DataImportProto, sift_client: SiftClient | None = None + ) -> DataImport: + """Create from proto.""" + return cls( + proto=proto, + id_=proto.data_import_id, + source_url=proto.source_url or None, + status=DataImportStatus.from_proto(proto.status), + error_message=proto.error_message or None, + warning_messages=list(proto.warning_messages), + created_date=proto.created_date.ToDatetime(tzinfo=timezone.utc), + modified_date=proto.modified_date.ToDatetime(tzinfo=timezone.utc), + run_id=proto.run_id or None, + report_id=proto.report_id or None, + asset_id=proto.asset_id or None, + data_start_time=( + proto.data_start_time.ToDatetime(tzinfo=timezone.utc) + if proto.HasField("data_start_time") + else None + ), + data_stop_time=( + proto.data_stop_time.ToDatetime(tzinfo=timezone.utc) + if proto.HasField("data_stop_time") + else None + ), + _client=sift_client, + ) + + def refresh(self) -> DataImport: + """Refresh this data import with the latest data from the API. + + Returns: + The updated DataImport object. + """ + updated = self.client.data_import.get(self._id_or_error) + self._update(updated) + return self diff --git a/python/lib/sift_client/sift_types/job.py b/python/lib/sift_client/sift_types/job.py index 676cdc8c28..386e59253d 100644 --- a/python/lib/sift_client/sift_types/job.py +++ b/python/lib/sift_client/sift_types/job.py @@ -19,6 +19,7 @@ from pathlib import Path from sift_client.client import SiftClient + from sift_client.sift_types.data_import import DataImport from sift_client.sift_types.run import Run @@ -187,6 +188,9 @@ class Job(BaseType[JobProto, "Job"]): job_status_details: JobStatusDetails | None job_details: JobDetails | None + # Import ID recorded at create time. Not a model field, so _update keeps it. + _data_import_id: str | None = None + @classmethod def _from_proto(cls, proto: JobProto, sift_client: SiftClient | None = None) -> Job: """Create from proto.""" @@ -316,6 +320,19 @@ def wait_until_complete( self._update(completed_job) return self + def get_data_import(self) -> DataImport: + """Get the data import this job ran. + + The import's error message and warnings live there, not on the job. + + Returns: + The DataImport this job ran. + + Raises: + ValueError: If this is not a data import job. + """ + return self.client.data_import.get(self._data_import_id_or_error("get_data_import()")) + def get_import_run(self) -> Run: """Get the run created by this data import job. @@ -326,11 +343,22 @@ def get_import_run(self) -> Run: ValueError: If this is not a data import job or the import has no associated run. """ + return self.client.data_import.get_run(self._data_import_id_or_error("get_import_run()")) + + def _data_import_id_or_error(self, method: str) -> str: + """Get the data import ID off this job, or raise if it is not an import job.""" if self.job_type != JobType.DATA_IMPORT: - raise ValueError("get_import_run() is only valid for data import jobs.") - if not isinstance(self.job_details, DataImportDetails): - raise ValueError("Job does not have data import details.") - return self.client.data_import.get_run(self.job_details.data_import_id) + raise ValueError(f"{method} is only valid for data import jobs.") + if isinstance(self.job_details, DataImportDetails): + return self.job_details.data_import_id + if self._data_import_id: + return self._data_import_id + raise ValueError("Job does not have data import details.") + + def _set_data_import_id(self, data_import_id: str) -> None: + """Record the import ID, covering a job whose details aren't set yet.""" + # This bypasses the frozen status of the model. + self.__dict__["_data_import_id"] = data_import_id def wait_and_download( self, diff --git a/python/lib/sift_client/sift_types/run.py b/python/lib/sift_client/sift_types/run.py index e912253424..516230153c 100644 --- a/python/lib/sift_client/sift_types/run.py +++ b/python/lib/sift_client/sift_types/run.py @@ -25,7 +25,7 @@ from sift_client.client import SiftClient from sift_client.sift_types.asset import Asset - from sift_client.sift_types.data_import import DataTypeKey, ImportConfig + from sift_client.sift_types.data_import import DataImport, DataTypeKey, ImportConfig from sift_client.sift_types.job import Job @@ -101,6 +101,11 @@ def assets(self) -> list[Asset]: return [] return self.client.assets.list_(asset_ids=self.asset_ids) + @property + def data_imports(self) -> list[DataImport]: + """Return all data imports that ingested into this run.""" + return self.client.data_import.list_(runs=[self._id_or_error]) + def archive(self) -> Run: """Archive the run.""" updated_run = self.client.runs.archive(run=self) @@ -144,7 +149,7 @@ def import_data( ) -> Job: """Import data from a file into this run. - Convenience method that calls ``client.data_imports.import_from_path`` + Convenience method that calls ``client.data_import.import_from_path`` with this run pre-filled. If the run has exactly one asset, ``asset`` is inferred automatically.