diff --git a/CHANGELOG.md b/CHANGELOG.md index 349dc1f..5836666 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,14 @@ All notable changes to this project will be documented in this file - replace naming of classes and methods using 'Workbench' to 'Bench' - replace create_sample, create_container long argument lists with new XXXPost objects +## Unreleased + +- Added support for importing Inventory CSV files (issue #32): `parse_csv_import_file`, + `import_csv_files`, `import_samples_csv`, `import_containers_csv` and + `import_subsamples_csv`, plus a new `ImportRecordType` enum. Supports samples, + subsamples and LIST containers; Instruments and Instrument Templates cannot be + imported from CSV. See `examples/import_inventory_csv.py`. + ## 2.7.2 2026-07-13 - Added support for the Inventory "Link" extra-field type (server PR #803 / diff --git a/examples/import_inventory_csv.py b/examples/import_inventory_csv.py new file mode 100644 index 0000000..cf75e41 --- /dev/null +++ b/examples/import_inventory_csv.py @@ -0,0 +1,67 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +Imports samples, subsamples and containers into Inventory from CSV files. + +The Inventory API supports importing samples, subsamples and (simple LIST) +containers from CSV. Instruments and Instrument Templates cannot be imported +from CSV. + +The typical workflow is: + 1. Parse the CSV to discover its columns and the suggested field mappings / + sample template (parse_csv_import_file). + 2. Adjust the mappings, then import (import_samples_csv / import_containers_csv + / import_subsamples_csv, or import_csv_files for multi-file imports). + +A field mapping maps a CSV column name to an RSpace field name, e.g. "name", +"description", "quantity", "expiry date", "identifier", "import identifier", +"parent container import id", "parent container global id", +"parent sample import id", "parent sample global id". +""" +import io + +import rspace_client + +inv_api = rspace_client.utils.createInventoryClient() + + +def csv_stream(content: str) -> io.BytesIO: + stream = io.BytesIO(content.encode("utf-8")) + stream.name = "import.csv" + return stream + + +# 1. Parse a samples CSV to see what the server suggests +samples_csv = "Name,Supplier,Comment\nAntibody A,Acme,first\nAntibody B,Acme,second" +parse_result = inv_api.parse_csv_import_file(csv_stream(samples_csv), "SAMPLES") +print(f"Columns found: {parse_result['columnNames']}") +print(f"Suggested field mappings: {parse_result['fieldMappings']}") + +# 2. Import the samples, creating a new Sample Template from the CSV columns +result = inv_api.import_samples_csv( + csv_stream(samples_csv), + field_mappings={"Name": "name", "Comment": "description"}, + template_info={"name": "Antibodies imported from CSV"}, +) +print(f"Samples import status: {result['status']}") +print(f" created {result['sampleResults']['successCount']} samples") + +# Import containers (LIST containers only) +containers_csv = "Name\nFreezer shelf 1\nFreezer shelf 2" +container_result = inv_api.import_containers_csv( + csv_stream(containers_csv), field_mappings={"Name": "name"} +) +print(f"Containers import status: {container_result['status']}") + +# Import subsamples into a pre-existing sample, referencing it by Global Id +sample = inv_api.create_sample("Sample for CSV subsamples") +subsamples_csv = ( + "Name,Parent\n" + f"Aliquot 1,{sample['globalId']}\n" + f"Aliquot 2,{sample['globalId']}" +) +subsample_result = inv_api.import_subsamples_csv( + csv_stream(subsamples_csv), + field_mappings={"Name": "name", "Parent": "parent sample global id"}, +) +print(f"Subsamples import status: {subsample_result['status']}") diff --git a/rspace_client/inv/inv.py b/rspace_client/inv/inv.py index 5ea0b9d..2f78aa3 100644 --- a/rspace_client/inv/inv.py +++ b/rspace_client/inv/inv.py @@ -58,6 +58,19 @@ class FillingStrategy(Enum): EXACT = 3 +class ImportRecordType(str, Enum): + """ + Type of Inventory record held in a CSV file being imported. + + Note that Inventory CSV import only supports these three record types. + Instruments and Instrument Templates cannot be imported from CSV. + """ + + SAMPLES = "SAMPLES" + SUBSAMPLES = "SUBSAMPLES" + CONTAINERS = "CONTAINERS" + + class Sample: """ Wraps a dict of Sample data returned from samples/{id} GET API call @@ -1975,6 +1988,266 @@ def get_workbenches(self) -> Sequence[dict]: result = self.retrieve_api_results("/workbenches") return [wb for wb in result["containers"]] + ## --------------------------------------------------------------------- + ## CSV import + ## --------------------------------------------------------------------- + ## The Inventory API can import CSV files of samples, subsamples and + ## containers (only simple LIST containers). Importing Instruments or + ## Instrument Templates from CSV is NOT supported by the API. + + def _multipart_post(self, endpoint: str, files: dict, data: dict) -> dict: + """ + Helper for multipart/form-data POSTs. ``requests`` sets the correct + multipart Content-Type (with boundary) when ``files`` is supplied, and + adds each entry of ``data`` as an additional form field. + """ + response = requests.post( + self._get_api_url() + endpoint, + files=files, + data=data, + headers=self._get_headers(), + ) + return self._handle_response(response) + + def parse_csv_import_file( + self, file: BinaryIO, record_type: Union[str, ImportRecordType] + ) -> dict: + """ + Uploads a CSV file for analysis prior to import, without creating + anything. Use this to discover the file's columns and the field + mappings/sample template that the server suggests, then adjust them + before calling one of the import methods. + + Parameters + ---------- + file : an open file + An open file stream of the CSV file to analyse. + record_type : Union[str, ImportRecordType] + The type of record held in the file: SAMPLES, SUBSAMPLES or + CONTAINERS. + + Returns + ------- + dict + Parse result containing 'columnNames', 'fieldNameForColumnName', + 'fieldMappings', 'columnsWithoutBlankValue' and 'rowsCount'. For + SAMPLES it also includes a suggested 'templateInfo', plus + 'radioOptionsForColumn' and 'quantityUnitForColumn'. + """ + record_type = ImportRecordType(record_type).value + return self._multipart_post( + "/import/parseFile", + files={"file": file}, + data={"recordType": record_type}, + ) + + def import_csv_files( + self, + import_settings: dict, + containers_file: BinaryIO = None, + samples_file: BinaryIO = None, + subsamples_file: BinaryIO = None, + ) -> dict: + """ + Imports one or more CSV files (containers, samples, subsamples) in a + single operation, using the supplied import settings. This is the + low-level method that mirrors the API directly; see + import_samples_csv/import_containers_csv/import_subsamples_csv for + convenience wrappers. + + Parameters + ---------- + import_settings : dict + The full import settings, e.g.:: + + { + "containerSettings": {"fieldMappings": {...}}, + "sampleSettings": {"fieldMappings": {...}, + "templateInfo": {"id": 1234}}, + "subSampleSettings": {"fieldMappings": {...}}, + } + + Each ``fieldMappings`` maps a CSV column name to an RSpace field + name (e.g. "name", "description", "quantity", "expiry date", + "identifier", "import identifier", "parent container import id", + "parent container global id", "parent sample import id", + "parent sample global id"). Map a column to ``None`` to ignore it. + containers_file : an open file, optional + samples_file : an open file, optional + subsamples_file : an open file, optional + The CSV files to import. Provide the files matching the settings + keys you populated. At least one file must be supplied. + + Returns + ------- + dict + Import result with a top-level 'status' and per-type + 'containerResults', 'sampleResults' and 'subSampleResults', plus + a 'defaultContainer' if one was created for the imported items. + """ + files = {} + if containers_file is not None: + files["containersFile"] = containers_file + if samples_file is not None: + files["samplesFile"] = samples_file + if subsamples_file is not None: + files["subSamplesFile"] = subsamples_file + if not files: + raise ValueError( + "At least one of containers_file, samples_file or " + "subsamples_file must be supplied" + ) + return self._multipart_post( + "/import/importFiles", + files=files, + data={"importSettings": json.dumps(import_settings)}, + ) + + @staticmethod + def _require_single_mapping( + field_mappings: dict, target: str, record: str + ) -> None: + """ + Ensures exactly one CSV column is mapped to ``target``. Mapping zero + columns leaves a required field unset; mapping more than one is + ambiguous (the server would silently use only one of them). + """ + count = ( + 0 if field_mappings is None else list(field_mappings.values()).count(target) + ) + if count != 1: + raise ValueError( + f"{record} field_mappings must map exactly one column to " + f"'{target}', but {count} were found" + ) + + @staticmethod + def _require_name_mapping(field_mappings: dict, record: str) -> None: + InventoryClient._require_single_mapping(field_mappings, "name", record) + + @staticmethod + def _sample_settings( + field_mappings: dict, template_id=None, template_info: dict = None + ) -> dict: + InventoryClient._require_name_mapping(field_mappings, "sample") + settings = {"fieldMappings": field_mappings} + if template_info is not None: + settings["templateInfo"] = template_info + elif template_id is not None: + settings["templateInfo"] = {"id": Id(template_id).as_id()} + else: + raise ValueError( + "Either template_id (to reuse an existing Sample Template) or " + "template_info (to create a new one) must be provided" + ) + return settings + + def import_samples_csv( + self, + samples_file: BinaryIO, + field_mappings: dict, + template_id: Union[str, int] = None, + template_info: dict = None, + ) -> dict: + """ + Imports a CSV file of samples. + + Parameters + ---------- + samples_file : an open file + An open file stream of the samples CSV file. + field_mappings : dict + Maps CSV column names to RSpace sample field names. Exactly one + column must map to "name". Every CSV column must either map to a + built-in field ("name", "description", "tags", "source", + "expiry date", "quantity", "identifier", "import identifier", + "parent container import id", "parent container global id") or be + left unmapped so that it corresponds, in order, to a custom field + of the template. The number of unmapped columns must exactly equal + the number of custom fields in the template, otherwise the server + rejects the import with a PREVALIDATION_ERROR status. + template_id : Union[str, int], optional + The id or global id of an existing Sample Template to create the + samples from. + template_info : dict, optional + A Sample Template definition to create a new template from, + typically the 'templateInfo' returned by parse_csv_import_file. + Provide either template_id or template_info. + + Returns + ------- + dict + The import result (see import_csv_files). + """ + settings = { + "sampleSettings": self._sample_settings( + field_mappings, template_id, template_info + ) + } + return self.import_csv_files(settings, samples_file=samples_file) + + def import_containers_csv( + self, containers_file: BinaryIO, field_mappings: dict + ) -> dict: + """ + Imports a CSV file of (LIST) containers. + + Parameters + ---------- + containers_file : an open file + An open file stream of the containers CSV file. + field_mappings : dict + Maps CSV column names to RSpace container field names. Exactly one + column must map to "name". + + Returns + ------- + dict + The import result (see import_csv_files). + """ + self._require_name_mapping(field_mappings, "container") + settings = {"containerSettings": {"fieldMappings": field_mappings}} + return self.import_csv_files(settings, containers_file=containers_file) + + def import_subsamples_csv( + self, subsamples_file: BinaryIO, field_mappings: dict + ) -> dict: + """ + Imports a CSV file of subsamples into pre-existing samples. + + Each subsample must reference its parent sample, so field_mappings must + map a column to "parent sample global id" (the Global Id of an existing + sample). To import subsamples for samples created in the same operation + (via "parent sample import id"), use import_csv_files with both + sampleSettings and subSampleSettings instead. + + Parameters + ---------- + subsamples_file : an open file + An open file stream of the subsamples CSV file. + field_mappings : dict + Maps CSV column names to RSpace subsample field names. Exactly one + column must map to "name" and one to "parent sample global id". + + Returns + ------- + dict + The import result (see import_csv_files). + """ + self._require_name_mapping(field_mappings, "subsample") + if "parent sample global id" not in (field_mappings or {}).values(): + raise ValueError( + "subsample field_mappings must map one column to " + "'parent sample global id'. To link subsamples to samples " + "imported in the same operation, use import_csv_files with " + "both sampleSettings and subSampleSettings." + ) + self._require_single_mapping( + field_mappings, "parent sample global id", "subsample" + ) + settings = {"subSampleSettings": {"fieldMappings": field_mappings}} + return self.import_csv_files(settings, subsamples_file=subsamples_file) + def bulk_create_container(self, *container_posts): """ Create up to MAX_BULK containers at once. diff --git a/rspace_client/tests/invapi_test.py b/rspace_client/tests/invapi_test.py index dd0d883..98364cb 100644 --- a/rspace_client/tests/invapi_test.py +++ b/rspace_client/tests/invapi_test.py @@ -6,6 +6,7 @@ @author: richard """ import sys, os +import io import json import datetime as dt import pprint @@ -1214,6 +1215,107 @@ def test_create_sample_template_with_link_field(self): ] self.assertEqual(1, len(link_fields)) + @staticmethod + def _csv(content: str) -> io.BytesIO: + f = io.BytesIO(content.encode("utf-8")) + f.name = "test.csv" + return f + + def test_parse_samples_csv(self): + parse_result = self.invapi.parse_csv_import_file( + self._csv("Name,Comment\nSample1,hello\nSample2,world"), + inv.ImportRecordType.SAMPLES, + ) + self.assertEqual(["Name", "Comment"], parse_result["columnNames"]) + self.assertEqual(2, parse_result["rowsCount"]) + # samples parse returns a suggested template + self.assertIsNotNone(parse_result["templateInfo"]) + + def test_import_samples_csv_creates_template(self): + # Every CSV column must either map to a built-in field or correspond to + # a template field; here both columns map to built-in fields, so a + # zero-field template is compatible. + template_name = base.random_string(8) + result = self.invapi.import_samples_csv( + self._csv("Name,Comment\nSample1,hello\nSample2,world"), + field_mappings={"Name": "name", "Comment": "description"}, + template_info={"name": template_name}, + ) + self.assertEqual("COMPLETED", result["status"]) + sample_result = result["sampleResults"] + self.assertTrue(sample_result["templateCreated"]) + self.assertEqual(2, sample_result["successCount"]) + self.assertEqual(0, sample_result["errorCount"]) + + def test_import_samples_csv_requires_name_mapping(self): + with self.assertRaises(ValueError): + self.invapi.import_samples_csv( + self._csv("Name\nSample1"), + field_mappings={"Name": "description"}, + template_info={"name": base.random_string(8)}, + ) + + def test_import_samples_csv_requires_template(self): + with self.assertRaises(ValueError): + self.invapi.import_samples_csv( + self._csv("Name\nSample1"), field_mappings={"Name": "name"} + ) + + def test_import_samples_csv_rejects_duplicate_name_mapping(self): + with self.assertRaises(ValueError): + self.invapi.import_samples_csv( + self._csv("Name,Alias\nSample1,S1"), + field_mappings={"Name": "name", "Alias": "name"}, + template_info={"name": base.random_string(8)}, + ) + + def test_import_containers_csv(self): + result = self.invapi.import_containers_csv( + self._csv("Name\nContainer1\nContainer2"), + field_mappings={"Name": "name"}, + ) + self.assertEqual("COMPLETED", result["status"]) + container_result = result["containerResults"] + self.assertEqual(2, container_result["successCount"]) + self.assertEqual(0, container_result["errorCount"]) + + def test_import_subsamples_into_existing_sample(self): + sample = self.invapi.create_sample(base.random_string(8)) + global_id = sample["globalId"] + result = self.invapi.import_subsamples_csv( + self._csv( + "Name,Parent sample\n" + f"SubSample1,{global_id}\n" + f"SubSample2,{global_id}" + ), + field_mappings={ + "Name": "name", + "Parent sample": "parent sample global id", + }, + ) + self.assertEqual("COMPLETED", result["status"]) + subsample_result = result["subSampleResults"] + self.assertEqual(2, subsample_result["successCount"]) + self.assertEqual(0, subsample_result["errorCount"]) + + def test_import_subsamples_csv_requires_parent_sample(self): + with self.assertRaises(ValueError): + self.invapi.import_subsamples_csv( + self._csv("Name\nSubSample1"), + field_mappings={"Name": "name"}, + ) + + def test_import_subsamples_csv_rejects_duplicate_parent_mapping(self): + with self.assertRaises(ValueError): + self.invapi.import_subsamples_csv( + self._csv("Name,P1,P2\nSubSample1,SA1,SA2"), + field_mappings={ + "Name": "name", + "P1": "parent sample global id", + "P2": "parent sample global id", + }, + ) + class LinkFieldUnitTest(unittest.TestCase): """