From 8e555808977b89b229199269d385b6ee2c2cada3 Mon Sep 17 00:00:00 2001 From: David Silaghi Date: Tue, 8 Sep 2026 12:16:42 +0300 Subject: [PATCH] [MCC-1547223] Python SDK - Include R dataset properties in get_datasets response Preserve nullable metadata and empty collections, attach lazy frames through existing fetch_data, and retain per-page listing and dataset-version behavior. TDD: baseline 186 passed / 1 unrelated logging-header failure; new regression tests failed before implementation; final focused suite 24 passed, full suite 203 passed / same baseline failure. Ruff and pre-commit pass. Mypy has the same three transport errors as the baseline commit. --- dataconnect/models.py | 41 +++- dataconnect/service/default.py | 10 +- dataconnect/service/mappers.py | 9 + readme/README-v1.1.0.md | 12 +- tests/test_get_datasets_paginated.py | 281 ++++++++++++++++++++++++++- 5 files changed, 347 insertions(+), 6 deletions(-) diff --git a/dataconnect/models.py b/dataconnect/models.py index a770277..4bd1387 100644 --- a/dataconnect/models.py +++ b/dataconnect/models.py @@ -1,7 +1,8 @@ from __future__ import annotations +from collections.abc import Callable from dataclasses import dataclass, field -from typing import Generic, TypeVar +from typing import Generic, Self, TypeVar from uuid import UUID import pandas as pd @@ -37,6 +38,34 @@ class DatasetVersion: dataset_version: str +class DatasetFrame: + """Lazy dataset reference; fetching requires the originating client to remain open.""" + + __slots__ = ("_dataset_uuid", "_fetch_data") + + def __init__(self, dataset_uuid: str, fetch_data: Callable[[UUID, int | None], pd.DataFrame]) -> None: + self._dataset_uuid = dataset_uuid + self._fetch_data = fetch_data + + def head(self, count: int = 6) -> pd.DataFrame: + """Fetch the first count rows, matching R's default of six rows.""" + return self._fetch_data(UUID(self._dataset_uuid), count) + + def collect(self) -> pd.DataFrame: + """Fetch the complete dataset without retaining a previous head limit.""" + return self._fetch_data(UUID(self._dataset_uuid), None) + + def __repr__(self) -> str: + return f"DatasetFrame(dataset_uuid={self._dataset_uuid!r})" + + def __copy__(self) -> Self: + return self + + def __deepcopy__(self, memo: dict[int, object]) -> Self: + """Keep this opaque reference intact when copying metadata with asdict().""" + return self + + @dataclass(frozen=True) class Dataset: """A dataset belonging to a study environment.""" @@ -45,6 +74,16 @@ class Dataset: study_uuid: str study_env_uuid: str dataset_name: str + dataset_short_name: str | None = None + type: str | None = None + source: str | None = None + activation_status: str | None = None + dataset_status: str | None = None + collection: list[str] | None = field(default=None, hash=False) + last_updated: str | None = None + version: str | None = None + other_versions: list[dict[str, str]] | None = field(default=None, hash=False) + frame: DatasetFrame | None = field(default=None, repr=False, compare=False) @dataclass(frozen=True) diff --git a/dataconnect/service/default.py b/dataconnect/service/default.py index fc5ffa3..571b7b5 100644 --- a/dataconnect/service/default.py +++ b/dataconnect/service/default.py @@ -3,6 +3,7 @@ from __future__ import annotations import json +from dataclasses import replace from datetime import UTC, datetime from uuid import UUID @@ -11,6 +12,7 @@ from dataconnect.exceptions import ValidationError from dataconnect.models import ( Dataset, + DatasetFrame, DatasetVersion, DatetimeFormat, DatetimeFormatsResult, @@ -151,7 +153,13 @@ def get_datasets( try: resources = self._transport.list_resources(request) - items = [resource_to_dataset(r) for r in resources] + items = [] + for resource in resources: + dataset = resource_to_dataset(resource) + frame = ( + DatasetFrame(dataset.dataset_uuid, self.fetch_data) if dataset.dataset_uuid is not None else None + ) + items.append(replace(dataset, frame=frame)) total_records = resources[0].total_records if resources else 0 total_pages = (total_records + page_size - 1) // page_size if page_size > 0 else 0 return PaginatedResponse( diff --git a/dataconnect/service/mappers.py b/dataconnect/service/mappers.py index bd01e67..0a6ac25 100644 --- a/dataconnect/service/mappers.py +++ b/dataconnect/service/mappers.py @@ -109,6 +109,15 @@ def resource_to_dataset(resource: ResourceInfo) -> Dataset: study_uuid=data.get("study_uuid", ""), study_env_uuid=data.get("study_env_uuid", ""), dataset_name=data.get("dataset_name", ""), + dataset_short_name=data.get("dataset_short_name"), + type=data.get("type"), + source=data.get("source"), + activation_status=data.get("activation_status"), + dataset_status=data.get("dataset_status"), + collection=data.get("collection"), + last_updated=data.get("last_updated"), + version=data.get("version"), + other_versions=data.get("other_versions"), ) diff --git a/readme/README-v1.1.0.md b/readme/README-v1.1.0.md index a778ec8..1d2a821 100644 --- a/readme/README-v1.1.0.md +++ b/readme/README-v1.1.0.md @@ -148,7 +148,15 @@ Retrieves datasets for a specific study environment and returns paginated result | page_size | int | Optional. Number of results per page. Default: 50 | #### Output -Returns a list containing `total_records` (total datasets available across all pages), `pagination` and `datasets` array. +Returns a `PaginatedResponse` containing `total_records` (total datasets available across all pages), `pagination`, and an `items` list. Each call retrieves one requested page; it does not automatically fetch subsequent pages. + +Each dataset retains `dataset_uuid`, `study_uuid`, `study_env_uuid`, and `dataset_name`, and also exposes `dataset_short_name`, `type`, `source`, `activation_status`, `dataset_status`, `collection`, `last_updated`, `version`, `other_versions`, and `frame`. + +Missing or null metadata is represented as `None`; empty collections remain `[]` and empty strings remain `""`. `collection` is a list of strings, and `other_versions` is a list of dictionaries with `version` and `dataset_uuid` keys, or `None`. Version labels and timestamps remain strings without conversion. + +`dataset.frame` is a lazy reference: listing datasets does not fetch their rows. `dataset.frame.head(10)` returns the first ten rows as a pandas DataFrame (default: six), while `dataset.frame.collect()` fetches the complete dataset regardless of previous previews. Keep the originating client open while using its frames. A null dataset UUID has no usable frame and returns `frame=None`. + +`dataclasses.asdict(dataset)` retains the frame as an opaque reference without copying its connection. Exclude `frame` when JSON-serializing the metadata. --- @@ -273,7 +281,7 @@ Returns a result object with the following attributes: | Validations | Description | |:---------------------|:---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | **Invalid Input** | Required argument is missing | -| **project_token** | 1. Project Token is valid and generated from the Data Connect > Transformations > Custom Code project type.
2. More than one dataset cannot be published into a project
3. Authorized collaborators and project owners can publish datasets into a project. | +| **project_token** | 1. Project Token is valid and generated from the Data Connect > Transformations > Custom Code project type.
2. More than one dataset cannot be published into a project
3. Authorized collaborators and project owners can publish datasets into a project. | | **dataset_name** | 1. Maximum length of 15 characters and must only contain alphanumeric characters and underscores
2. This is the new name of the resulting dataset created by the user | | **key_columns** | 1. Key columns are valid column names from the data frame being published
2. Key columns must not contain null/missing values (for example, `None`) in any row
3. Maps directly to the server-side metrics payload: `valid_record_count`, `duplicate_record_count`, and `invalid_record_count` without double-penalizing overlapping row states. | | **source_datasets** | 1. Source Dataset is a valid dataset UUID
2. Source Dataset is from the same study environment. | diff --git a/tests/test_get_datasets_paginated.py b/tests/test_get_datasets_paginated.py index f16d9cf..bea7bc7 100644 --- a/tests/test_get_datasets_paginated.py +++ b/tests/test_get_datasets_paginated.py @@ -3,14 +3,20 @@ from __future__ import annotations import json +from copy import copy, deepcopy +from dataclasses import asdict, fields +from threading import Lock from uuid import UUID +import pandas as pd +import pyarrow as pa import pytest +from dataconnect.client import DataConnectClient from dataconnect.models import Dataset, PaginatedResponse, Pagination from dataconnect.service.default import DefaultDataConnectService from dataconnect.transport.errors import TransportError -from dataconnect.transport.models import DataRef, ResourceInfo, ResourceQuery +from dataconnect.transport.models import DataRef, DatasetTicket, DataTable, ResourceInfo, ResourceQuery class _FakeTransport: @@ -36,7 +42,7 @@ def close(self) -> None: return None -def _dataset_resource(payload: dict[str, str], total_records: int = 1) -> ResourceInfo: +def _dataset_resource(payload: dict[str, object], total_records: int = 1) -> ResourceInfo: return ResourceInfo( descriptor=b"", endpoints=[DataRef(ticket=json.dumps(payload).encode("utf-8"))], @@ -170,3 +176,274 @@ def test_multiple_items_returned(self) -> None: assert result.total_records == 2 assert result.items[0].dataset_name == "labs" assert result.items[1].dataset_name == "vitals" + + +_DATASET_UUID = "073410b6-79be-3e7d-ae37-92f6e054013e" +_OTHER_DATASET_UUID = "006e4963-aba4-5d3b-93c4-e01354077219" +_IDENTIFIERS = { + "dataset_uuid": _DATASET_UUID, + "study_uuid": "64a98a9b-1512-44c8-92af-e4cab0183670", + "study_env_uuid": str(_STUDY_ENV_UUID), + "dataset_name": "LBHEM2", +} +_METADATA = { + "dataset_short_name": "LB", + "type": "Derived Dataset", + "source": "JL_templ_upgrd2", + "activation_status": "Activated", + "dataset_status": "Warning", + "collection": ["clinical", "labs"], + "last_updated": "2024-06-15 09:30:00", + "version": "1", + "other_versions": [{"version": "0", "dataset_uuid": _OTHER_DATASET_UUID}], +} +_EXPECTED_FIELDS = { + "dataset_uuid", + "study_uuid", + "study_env_uuid", + "dataset_name", + "dataset_short_name", + "type", + "source", + "activation_status", + "dataset_status", + "collection", + "last_updated", + "version", + "other_versions", + "frame", +} + + +class _PagedFetchingTransport(_FakeTransport): + def __init__(self, pages: dict[int, list[dict[str, object]]]) -> None: + super().__init__() + self.pages = pages + self.requests: list[ResourceQuery] = [] + self.tickets: list[DatasetTicket] = [] + self.closed = False + self.lock = Lock() + self.fetch_error: TransportError | None = None + + def list_resources(self, request: ResourceQuery) -> list[ResourceInfo]: + self.requests.append(request) + body = json.loads(request.body) + total_records = sum(len(payloads) for payloads in self.pages.values()) + return [_dataset_resource(payload, total_records) for payload in self.pages[body["page"]]] + + def get_ticket(self, ticket: DatasetTicket) -> DataTable: + if self.closed: + raise TransportError(error_code="CLOSED", message="transport closed") + if self.fetch_error is not None: + raise self.fetch_error + self.tickets.append(ticket) + table = pa.table({"dataset_uuid": [ticket.dataset_uuid] * 8, "value": list(range(8))}) + if ticket.limit is not None: + table = table.slice(0, ticket.limit) + buffer = pa.BufferOutputStream() + with pa.ipc.new_stream(buffer, table.schema) as writer: + writer.write_table(table) + return DataTable(schema_bytes=table.schema.serialize().to_pybytes(), ipc_bytes=buffer.getvalue().to_pybytes()) + + def close(self) -> None: + self.closed = True + + +def test_complete_metadata_and_frames_survive_independent_pages() -> None: + pages = { + 1: [{**_IDENTIFIERS, **_METADATA}], + 2: [{**_IDENTIFIERS, **_METADATA, "dataset_uuid": _OTHER_DATASET_UUID, "version": "41, 51"}], + } + transport = _PagedFetchingTransport(pages) + client = DataConnectClient(DefaultDataConnectService(transport)) + + for page_number, payloads in pages.items(): + result = client.get_datasets(_STUDY_ENV_UUID, search_dataset_name="LB", page=page_number, page_size=1) + assert len(transport.requests) == page_number + assert transport.requests[-1].action == "datasets.list" + assert json.loads(transport.requests[-1].body) == { + "study_environment_uuid": str(_STUDY_ENV_UUID), + "search_dataset_name": "LB", + "page": page_number, + "page_size": 1, + } + assert result.total_records == 2 + assert result.pagination == Pagination(page=page_number, page_size=1, total_pages=2) + dataset = result.items[0] + assert {item.name for item in fields(dataset)} == _EXPECTED_FIELDS + for name, value in payloads[0].items(): + assert getattr(dataset, name) == value + assert dataset.frame is not None + + assert transport.tickets == [] + + +@pytest.mark.parametrize("scenario", ["missing", "null", "empty"]) +def test_metadata_preserves_missing_null_and_empty_values(scenario: str) -> None: + metadata = {} + if scenario == "null": + metadata = dict.fromkeys(_METADATA) + elif scenario == "empty": + metadata = {name: [] if name in {"collection", "other_versions"} else "" for name in _METADATA} + transport = _FakeTransport([_dataset_resource({**_IDENTIFIERS, **metadata})]) + + dataset = DefaultDataConnectService(transport).get_datasets(_STUDY_ENV_UUID).items[0] + + for name in _METADATA: + assert getattr(dataset, name) == metadata.get(name) + assert dataset.frame is not None + + +@pytest.mark.parametrize("version", ["3", "v1,v2", "41, 51, 88", "", None]) +def test_version_labels_are_preserved_without_coercion(version: str | None) -> None: + transport = _FakeTransport([_dataset_resource({**_IDENTIFIERS, "version": version})]) + + dataset = DefaultDataConnectService(transport).get_datasets(_STUDY_ENV_UUID).items[0] + + assert dataset.version == version + + +def test_frame_fetches_only_on_demand_and_binds_each_dataset() -> None: + transport = _PagedFetchingTransport( + { + 1: [ + {**_IDENTIFIERS, **_METADATA}, + {**_IDENTIFIERS, "dataset_uuid": _OTHER_DATASET_UUID}, + ] + } + ) + client = DataConnectClient(DefaultDataConnectService(transport)) + first, second = client.get_datasets(_STUDY_ENV_UUID).items + assert transport.tickets == [] + assert first.frame is not None + assert second.frame is not None + assert first.frame is not second.frame + + preview = first.frame.head(3) + assert isinstance(preview, pd.DataFrame) + assert preview["value"].tolist() == [0, 1, 2] + assert preview["dataset_uuid"].tolist() == [_DATASET_UUID] * 3 + assert len(first.frame.head()) == 6 + assert len(first.frame.collect()) == 8 + assert second.frame.head(1)["dataset_uuid"].tolist() == [_OTHER_DATASET_UUID] + assert transport.tickets == [ + DatasetTicket(dataset_uuid=_DATASET_UUID, limit=3), + DatasetTicket(dataset_uuid=_DATASET_UUID, limit=6), + DatasetTicket(dataset_uuid=_DATASET_UUID, limit=None), + DatasetTicket(dataset_uuid=_OTHER_DATASET_UUID, limit=1), + ] + + +def test_frame_copy_and_inspection_do_not_copy_or_fetch_connection() -> None: + transport = _PagedFetchingTransport({1: [{**_IDENTIFIERS, **_METADATA}]}) + dataset = DefaultDataConnectService(transport).get_datasets(_STUDY_ENV_UUID).items[0] + assert dataset.frame is not None + + encoded = asdict(dataset) + assert set(encoded) == _EXPECTED_FIELDS + assert encoded["frame"] is dataset.frame + assert deepcopy(dataset).frame is dataset.frame + assert copy(dataset.frame) is dataset.frame + assert "_fetch_data" not in repr(dataset.frame) + assert "_PagedFetchingTransport" not in repr(dataset) + for name, value in _METADATA.items(): + assert encoded[name] == value + assert transport.tickets == [] + assert len(encoded["frame"].head(1)) == 1 + + +def test_dataset_constructor_and_equality_remain_independent_of_frame() -> None: + transport = _FakeTransport([_dataset_resource(_IDENTIFIERS)]) + dataset = DefaultDataConnectService(transport).get_datasets(_STUDY_ENV_UUID).items[0] + legacy = Dataset(**_IDENTIFIERS) + + assert legacy.frame is None + assert dataset == legacy + assert dataset.frame is not None + + +def test_frame_uses_existing_fetch_error_translation_and_client_lifetime() -> None: + transport = _PagedFetchingTransport({1: [_IDENTIFIERS]}) + service = DefaultDataConnectService(transport) + client = DataConnectClient(service) + dataset = client.get_datasets(_STUDY_ENV_UUID).items[0] + assert dataset.frame is not None + transport.fetch_error = TransportError(error_code="FETCH", message="fetch failed") + + with pytest.raises(Exception) as direct: + service.fetch_data(UUID(_DATASET_UUID)) + with pytest.raises(type(direct.value), match="fetch failed"): + dataset.frame.collect() + + transport.fetch_error = None + client.close() + with pytest.raises(Exception, match="transport closed"): + dataset.frame.collect() + + +def test_r_screenshot_metadata_preserves_empty_collections_and_null_timestamp() -> None: + payload = { + **_IDENTIFIERS, + **_METADATA, + "dataset_short_name": "", + "collection": [], + "last_updated": None, + "other_versions": [], + } + transport = _FakeTransport([_dataset_resource(payload)]) + dataset = DefaultDataConnectService(transport).get_datasets(_STUDY_ENV_UUID).items[0] + + assert dataset.dataset_short_name == "" + assert dataset.collection == [] + assert dataset.last_updated is None + assert dataset.other_versions == [] + assert dataset.frame is not None + + +def test_null_dataset_identifier_is_preserved_without_attaching_frame() -> None: + transport = _FakeTransport([_dataset_resource({**_IDENTIFIERS, "dataset_uuid": None})]) + dataset = DefaultDataConnectService(transport).get_datasets(_STUDY_ENV_UUID).items[0] + + assert dataset.dataset_uuid is None + assert dataset.frame is None + + +def test_dataset_remains_hashable_with_collection_metadata() -> None: + transport = _FakeTransport([_dataset_resource({**_IDENTIFIERS, **_METADATA})]) + dataset = DefaultDataConnectService(transport).get_datasets(_STUDY_ENV_UUID).items[0] + + assert hash(dataset) == hash(deepcopy(dataset)) + assert {dataset: "cached metadata"}[deepcopy(dataset)] == "cached metadata" + + +def test_dataset_versions_response_is_unchanged_before_and_after_listing() -> None: + older = {**_IDENTIFIERS, "dataset_version": "1", "dataset_uuid": _OTHER_DATASET_UUID} + newer = {**_IDENTIFIERS, "dataset_version": "2"} + versions = [_dataset_resource(older), _dataset_resource(newer)] + transport = _FakeTransport(versions) + client = DataConnectClient(DefaultDataConnectService(transport)) + expected = [ + { + "study_uuid": UUID(_IDENTIFIERS["study_uuid"]), + "study_environment_uuid": _STUDY_ENV_UUID, + "dataset_uuid": UUID(_DATASET_UUID), + "dataset_name": "LBHEM2", + "dataset_version": "2", + }, + { + "study_uuid": UUID(_IDENTIFIERS["study_uuid"]), + "study_environment_uuid": _STUDY_ENV_UUID, + "dataset_uuid": UUID(_OTHER_DATASET_UUID), + "dataset_name": "LBHEM2", + "dataset_version": "1", + }, + ] + + assert [asdict(item) for item in client.get_dataset_versions(UUID(_DATASET_UUID))] == expected + transport._resources = [_dataset_resource({**_IDENTIFIERS, **_METADATA})] + assert client.get_datasets(_STUDY_ENV_UUID).items[0].frame is not None + transport._resources = versions + assert [asdict(item) for item in client.get_dataset_versions(UUID(_DATASET_UUID))] == expected + assert transport.last_request is not None + assert transport.last_request.action == "dataset_versions.list" + assert json.loads(transport.last_request.body) == {"dataset_uuid": _DATASET_UUID}