Skip to content

Reference

datalab_api special

DuplicateItemError (ValueError)

Raised when the API operation would create a duplicate item.

Source code in datalab_api/__init__.py
class DuplicateItemError(ValueError):
    """Raised when the API operation would create a duplicate item."""

DatalabClient (BaseDatalabClient)

A client for the Datalab API.

The client will keep an HTTP session open for the duration of its life and thus is amenable for use as a context manager, e.g.,

with DatalabClient("https://demo-api.datalab-org.io") as client:
    client.get_items()
Source code in datalab_api/__init__.py
class DatalabClient(BaseDatalabClient):
    """A client for the Datalab API.

    The client will keep an HTTP session open for the duration of its life and thus is
    amenable for use as a context manager, e.g.,

    ```python
    with DatalabClient("https://demo-api.datalab-org.io") as client:
        client.get_items()
    ```

    """

    def get_info(self) -> dict[str, Any]:
        """Fetch metadata associated with this datalab instance.

        The server information is stored on the client and can be accessed
        by the `info` attribute.

        Returns:
            dict: The JSON response from the `/info` endpoint of the Datalab API.

        """
        info_url = f"{self.datalab_api_url}/info"
        info_resp = self.session.get(info_url, follow_redirects=True)
        self.info = info_resp.json()["data"]
        return self.info

    def authenticate(self):
        """Tests authentication of the client with the Datalab API."""
        user_url = f"{self.datalab_api_url}/get-current-user"
        user_resp = self.session.get(user_url, follow_redirects=True)
        if user_resp.status_code != 200:
            raise RuntimeError(
                f"Failed to authenticate to {self.datalab_api_url!r}: {user_resp.status_code=} from {self._headers}. Please check your API key."
            )
        return user_resp.json()

    def get_block_info(self) -> list[dict[str, Any]]:
        """Return the list of available data blocks types for this instance,
        including some details and descriptions of their usage.

        The block information is stored on the client and can be accessed
        by the `block_info` attribute.

        Returns:
            A list of dictionary of block types.

        """
        block_info_url = f"{self.datalab_api_url}/info/blocks"
        block_info_resp = self.session.get(block_info_url, follow_redirects=True)
        if block_info_resp.status_code != 200:
            raise RuntimeError(
                f"Failed to list block information at {block_info_url}: {block_info_resp.status_code=}."
            )
        self.block_info = block_info_resp.json()["data"]
        return self.block_info

    def get_items(self, item_type: str | None = "samples") -> list[dict[str, Any]]:
        """List items of the given type available to the authenticated user.

        Parameters:
            item_type: The type of item to list. Defaults to "samples". Other
            choices will trigger a call to the `/<item_type>` endpoint of the API.

        Returns:
            A list of items of the given type.

        """
        if item_type is None:
            item_type = "samples"
        items_url = f"{self.datalab_api_url}/{item_type}"
        items_resp = self.session.get(items_url, follow_redirects=True)
        if items_resp.status_code != 200:
            raise RuntimeError(
                f"Failed to list items with {item_type=} at {items_url}: {items_resp.status_code=}. Check the item type is correct."
            )
        items = items_resp.json()
        if items["status"] != "success":
            raise RuntimeError(f"Failed to list items at {items_url}: {items['status']!r}.")
        return items[item_type]

    def search_items(
        self, query: str, item_types: Iterable[str] | str = ("samples", "cells")
    ) -> list[dict[str, Any]]:
        """Search for items of the given types that match the query.

        Parameters:
            query: Free-text query to search for.
            item_types: The types of items to search for. Defaults to ["samples", "cells"].

        Returns:
            An ordered list of items of the given types that match the query.

        """
        if isinstance(item_types, str):
            item_types = [item_types]

        search_items_url = (
            f"{self.datalab_api_url}/search-items?query={query}&types={','.join(item_types)}"
        )
        items_resp = self.session.get(search_items_url, follow_redirects=True)
        if items_resp.status_code != 200:
            raise RuntimeError(
                f"Failed to search items with {item_types=} at {search_items_url}: {items_resp.status_code=}"
            )
        items = items_resp.json()
        if items["status"] != "success":
            raise RuntimeError(f"Failed to list items at {search_items_url}: {items['status']!r}.")

        return items["items"]

    def create_item(
        self,
        item_id: str | None = None,
        item_type: str = "samples",
        item_data: dict[str, Any] | None = None,
        collection_id: str | None = None,
    ) -> dict[str, Any]:
        """Create an item with a given ID and item data.

        Parameters:
            item_id: The ID of the item to create, if set to `None`, the server will generate one.
            item_type: The type of item to create, e.g., 'samples', 'cells'.
            item_data: The data for the item.
            collection_id: The ID of the collection to add the item to (optional).
                If such a collection does not exist, one will be made.

        """
        new_item = {}
        if item_data is not None:
            new_item = item_data
        new_item.update({"item_id": item_id, "type": item_type})

        if new_item["item_id"] is None:
            new_item.pop("item_id")

        if collection_id is not None:
            try:
                collection_immutable_id = self.get_collection(collection_id)["immutable_id"]
            except RuntimeError:
                self.create_collection(collection_id)
                collection_immutable_id = self.get_collection(collection_id)["immutable_id"]
            new_item["collections"] = new_item.get("collections", [])
            new_item["collections"].append({"immutable_id": collection_immutable_id})

        create_item_url = f"{self.datalab_api_url}/new-sample/"
        create_item_resp = self.session.post(
            create_item_url,
            json={"new_sample_data": new_item, "generate_id_automatically": item_id is None},
            follow_redirects=True,
        )
        try:
            created_item = create_item_resp.json()
            if create_item_resp.status_code == 409:
                raise DuplicateItemError(
                    f"Item {item_id=} already exists at {create_item_url}: {created_item['status']!r}."
                )
            if created_item["status"] != "success":
                raise RuntimeError(f"Failed to create item at {create_item_url}: {created_item}.")
            return created_item["sample_list_entry"]

        except Exception as exc:
            raise exc.__class__(
                f"Failed to create item {item_id=} with data {item_data=} at {create_item_url}: {create_item_resp.status_code=}, {create_item_resp.content}. Check the item information is correct."
            ) from exc

    def update_item(self, item_id: str, item_data: dict[str, Any]) -> dict[str, Any]:
        """Update an item with the given item data.

        Parameters:
            item_id: The ID of the item to update.
            item_data: The new data for the item.

        Returns:
            A dictionary of the updated item data.

        """
        update_item_data = {"item_id": item_id, "data": item_data}
        update_item_url = f"{self.datalab_api_url}/save-item/"
        update_item_resp = self.session.post(
            update_item_url,
            json=update_item_data,
            follow_redirects=True,
        )
        if update_item_resp.status_code != 200:
            raise RuntimeError(
                f"Failed to update item {item_id=} at {update_item_url}: {update_item_resp.status_code=}. Check the item information is correct."
            )
        updated_item = update_item_resp.json()
        if updated_item["status"] != "success":
            raise RuntimeError(
                f"Failed to update item at {update_item_url}: {updated_item['status']!r}."
            )
        return updated_item

    def get_item(
        self,
        item_id: str | None = None,
        refcode: str | None = None,
        load_blocks: bool = False,
    ) -> dict[str, Any]:
        """Get an item with a given ID or refcode.

        Parameters:
            item_id: The ID of the item to search for.
            refcode: The refcode of the item to search fork
            load_blocks: Whether to load the blocks associated with the item.

        Returns:
            A dictionary of item data for the item with the given ID or refcode.

        """

        if item_id is None and refcode is None:
            raise ValueError("Must provide one of `item_id` or `refcode`.")
        if item_id is not None and refcode is not None:
            raise ValueError("Must provide only one of `item_id` or `refcode`.")

        if refcode is not None:
            raise NotImplementedError("Searching by `refcode` is not yet implemented.")

        item_url = f"{self.datalab_api_url}/get-item-data/{item_id}"
        item_resp = self.session.get(item_url, follow_redirects=True)

        if item_resp.status_code != 200:
            raise RuntimeError(
                f"Failed to find item {item_id=}, {refcode=} {item_url}: {item_resp.status_code=}. Check the item information is correct."
            )

        item = item_resp.json()
        if item["status"] != "success":
            raise RuntimeError(f"Failed to get item at {item_url}: {item['status']!r}.")

        # Filter out any deleted blocks
        item["item_data"]["blocks_obj"] = {
            block_id: block
            for block_id, block in item["item_data"]["blocks_obj"].items()
            if block_id in item["item_data"]["display_order"]
        }

        # Make a call to `/update-block` which will parse/create plots and return as JSON
        if load_blocks:
            ret_item_id = item["item_data"]["item_id"]
            for block_id, block in item["item_data"]["blocks_obj"].items():
                block_data = self.get_block(
                    item_id=ret_item_id, block_id=block_id, block_data=block
                )
                item["item_data"]["blocks_obj"][block_id] = block_data

        return item["item_data"]

    def get_item_files(self, item_id: str) -> None:
        """Download all the files for a given item and save them locally
        in the current working directory.

        Parameters:
            item_id: The ID of the item to search for.

        """

        item_data = self.get_item(item_id)
        for f in item_data.get("files", []):
            url = f"{self.datalab_api_url}/files/{f['immutable_id']}/{f['name']}"
            if Path(f["name"]).exists():
                warnings.warn(f"Will not overwrite existing file {f['name']}")
                continue
            with open(f["name"], "wb") as file:
                response = self.session.get(url, follow_redirects=True)
                file.write(response.content)

    def get_block(self, item_id: str, block_id: str, block_data: dict[str, Any]) -> dict[str, Any]:
        """Get a block with a given ID and block data.
        Should be used in conjunction with `get_item` to load an existing block.

        Parameters:
            item_id: The ID of the item to search for.
            block_id: The ID of the block to search for.
            block_data: Any other block data required by the request.

        Returns:
            A dictionary of block data for the block with the given ID.

        """
        block_url = f"{self.datalab_api_url}/update-block/"
        block_request = {
            "block_data": block_data,
            "item_id": item_id,
            "block_id": block_id,
            "save_to_db": False,
        }
        block_resp = self.session.post(block_url, json=block_request, follow_redirects=True)
        if block_resp.status_code != 200:
            raise RuntimeError(
                f"Failed to find block {block_id=} for item {item_id=} at {block_url}: {block_resp.status_code=}. Check the block information is correct."
            )

        block = block_resp.json()
        if block["status"] != "success":
            raise RuntimeError(f"Failed to get block at {block_url}: {block['status']!r}.")
        return block["new_block_data"]

    def upload_file(self, item_id: str, file_path: Path | str) -> dict[str, Any]:
        """Upload a file to an item with a given ID.

        Parameters:
            item_id: The ID of the item to upload the file to.
            file_path: The path to the file to upload.

        Returns:
            A dictionary of the uploaded file data.

        """
        if isinstance(file_path, str):
            file_path = Path(file_path)
        if not file_path.exists():
            raise FileNotFoundError(f"File {file_path=} does not exist.")

        upload_url = f"{self.datalab_api_url}/upload-file/"
        with open(file_path, "rb") as file:
            files = {"file": (file_path.name, file)}
            upload_resp = self.session.post(
                upload_url,
                files=files,
                data={"item_id": item_id, "replace_file": None},
                follow_redirects=True,
            )
        if upload_resp.status_code != 201:
            raise RuntimeError(
                f"Failed to upload file {file_path=} to item {item_id=} at {upload_url}: {upload_resp.status_code=}. Check the file information is correct."
            )

        upload = upload_resp.json()
        if upload["status"] != "success":
            raise RuntimeError(f"Failed to upload file at {upload_url}: {upload['status']!r}.")

        return upload

    def create_data_block(
        self,
        item_id: str,
        block_type: str,
        file_ids: str | list[str] | None = None,
        file_paths: Path | list[Path] | None = None,
    ) -> dict[str, Any]:
        """Creates a data block for an item with the given block type and file ID.

        Parameters:
            item_id: The ID of the item to create the block for.
            block_type: The type of block to create.
            file_ids: The ID of the file to attach to the block, must already
                be uploaded to the item.
            file_paths: A path, or set of paths, to files to upload and attach to the
                item. Conflicts with `file_id`, only one can be provided.

        Returns:
            The created block data.

        """

        blocks_url = f"{self.datalab_api_url}/add-data-block/"

        # TODO: future validation of block types based on server response
        payload = {
            "item_id": item_id,
            "block_type": block_type,
            "index": None,
        }

        if file_paths:
            raise NotImplementedError(
                "Simultaneously uploading files and creating blocks is not yet supported."
            )

        if file_ids:
            if isinstance(file_ids, str):
                file_ids = [file_ids]
            # check that the file is attached to the item
            item = self.get_item(item_id=item_id, load_blocks=False)
            attached_file_ids = set(item["file_ObjectIds"])
            if not set(file_ids).issubset(attached_file_ids):
                raise RuntimeError(
                    f"Not all IDs {file_ids=} are attached to item {item_id=}: {attached_file_ids=}"
                )

        block_resp = self.session.post(blocks_url, json=payload, follow_redirects=True)
        if block_resp.status_code != 200:
            raise RuntimeError(
                f"Failed to create block {block_type=} for item {item_id=}:\n{block_resp.text}"
            )
        block_data = block_resp.json()["new_block_obj"]

        if file_ids:
            block_data = self._update_data_block(
                block_type=block_type, block_data=block_data, file_ids=file_ids
            )

        return block_data

    def update_data_block(
        self, item_id: str, block_id: str, block_type: str, block_data: dict
    ) -> dict[str, Any]:
        """Attempts to update a block with the given payload of data,
        returning the updated block.

        Parameters:
            item_id: The ID of the item that the block is attached to.
            block_id: The ID of existing block.
            block_type: The type of block to update.
            block_data: The payload of block data to update; will be used to selectively update
                any fields present, with validation performed by the remote block code itself.

        Returns:
            If successful, the updated block data.

        Raises:
            RuntimeError: If the block update fails.

        """
        payload = {k: v for k, v in block_data.items()}
        payload["block_id"] = block_id
        payload["blocktype"] = block_type
        payload["item_id"] = item_id

        return self._update_data_block(block_type=block_type, block_data=payload)

    def _update_data_block(
        self, block_type: str, block_data: dict, file_ids: str | list[str] | None = None
    ) -> dict[str, Any]:
        """Attaches files to blocks: should only be used via wrapper methods."""
        if file_ids and isinstance(file_ids, str):
            file_ids = [file_ids]

        if file_ids:
            if len(file_ids) > 1:
                raise RuntimeError(
                    "API does not currently support attaching multiple files in a block."
                )
            block_data["file_id"] = file_ids[0]

        blocks_url = f"{self.datalab_api_url}/update-block/"
        payload = {
            "block_data": block_data,
            "block_type": block_type,
            "save_to_db": True,
        }

        resp = self.session.post(blocks_url, json=payload, follow_redirects=True)
        if resp.status_code != 200:
            raise RuntimeError(f"Failed to update block {block_type=}:\n{resp.text}")

        return resp.json()["new_block_data"]

    def get_collection(self, collection_id: str) -> dict[str, Any]:
        """Get a collection with a given ID.

        Parameters:
            collection_id: The ID of the collection to search for.

        Returns:
            A dictionary of collection data for the collection with the given ID.

        """
        collection_url = f"{self.datalab_api_url}/collections/{collection_id}"
        collection_resp = self.session.get(collection_url, follow_redirects=True)
        if collection_resp.status_code != 200:
            raise RuntimeError(
                f"Failed to find collection {collection_id=} at {collection_url}: {collection_resp.status_code=}. Check the collection ID is correct."
            )
        collection = collection_resp.json()
        if collection["status"] != "success":
            raise RuntimeError(
                f"Failed to get collection at {collection_url}: {collection['status']!r}."
            )
        return collection["data"]

    def create_collection(
        self, collection_id: str, collection_data: dict | None = None
    ) -> dict[str, Any]:
        """Create a collection with a given ID and collection data.

        Parameters:
            collection_id: The ID of the collection to create.
            collection_data: The data for the collection.

        """
        collection_url = f"{self.datalab_api_url}/collections"

        new_collection = {}
        if collection_data is not None:
            new_collection = collection_data
        new_collection.update({"collection_id": collection_id, "type": "collections"})

        collection_resp = self.session.put(
            collection_url,
            json={"data": new_collection},
            follow_redirects=True,
        )

        if collection_resp.status_code != 201 or collection_resp.json()["status"] != "success":
            raise RuntimeError(
                f"Failed to create collection {collection_id=} at {collection_url}: {collection_resp.status_code=}. Check the collection information is correct: {collection_resp.content}"
            )

        created_collection = collection_resp.json()["data"]
        return created_collection

get_info(self) -> dict[str, Any]

Fetch metadata associated with this datalab instance.

The server information is stored on the client and can be accessed by the info attribute.

Returns:

Type Description
dict

The JSON response from the /info endpoint of the Datalab API.

Source code in datalab_api/__init__.py
def get_info(self) -> dict[str, Any]:
    """Fetch metadata associated with this datalab instance.

    The server information is stored on the client and can be accessed
    by the `info` attribute.

    Returns:
        dict: The JSON response from the `/info` endpoint of the Datalab API.

    """
    info_url = f"{self.datalab_api_url}/info"
    info_resp = self.session.get(info_url, follow_redirects=True)
    self.info = info_resp.json()["data"]
    return self.info

authenticate(self)

Tests authentication of the client with the Datalab API.

Source code in datalab_api/__init__.py
def authenticate(self):
    """Tests authentication of the client with the Datalab API."""
    user_url = f"{self.datalab_api_url}/get-current-user"
    user_resp = self.session.get(user_url, follow_redirects=True)
    if user_resp.status_code != 200:
        raise RuntimeError(
            f"Failed to authenticate to {self.datalab_api_url!r}: {user_resp.status_code=} from {self._headers}. Please check your API key."
        )
    return user_resp.json()

get_block_info(self) -> list[dict[str, Any]]

Return the list of available data blocks types for this instance, including some details and descriptions of their usage.

The block information is stored on the client and can be accessed by the block_info attribute.

Returns:

Type Description
list[dict[str, Any]]

A list of dictionary of block types.

Source code in datalab_api/__init__.py
def get_block_info(self) -> list[dict[str, Any]]:
    """Return the list of available data blocks types for this instance,
    including some details and descriptions of their usage.

    The block information is stored on the client and can be accessed
    by the `block_info` attribute.

    Returns:
        A list of dictionary of block types.

    """
    block_info_url = f"{self.datalab_api_url}/info/blocks"
    block_info_resp = self.session.get(block_info_url, follow_redirects=True)
    if block_info_resp.status_code != 200:
        raise RuntimeError(
            f"Failed to list block information at {block_info_url}: {block_info_resp.status_code=}."
        )
    self.block_info = block_info_resp.json()["data"]
    return self.block_info

get_items(self, item_type: str | None = 'samples') -> list[dict[str, Any]]

List items of the given type available to the authenticated user.

Parameters:

Name Type Description Default
item_type str | None

The type of item to list. Defaults to "samples". Other

'samples'

Returns:

Type Description
list[dict[str, Any]]

A list of items of the given type.

Source code in datalab_api/__init__.py
def get_items(self, item_type: str | None = "samples") -> list[dict[str, Any]]:
    """List items of the given type available to the authenticated user.

    Parameters:
        item_type: The type of item to list. Defaults to "samples". Other
        choices will trigger a call to the `/<item_type>` endpoint of the API.

    Returns:
        A list of items of the given type.

    """
    if item_type is None:
        item_type = "samples"
    items_url = f"{self.datalab_api_url}/{item_type}"
    items_resp = self.session.get(items_url, follow_redirects=True)
    if items_resp.status_code != 200:
        raise RuntimeError(
            f"Failed to list items with {item_type=} at {items_url}: {items_resp.status_code=}. Check the item type is correct."
        )
    items = items_resp.json()
    if items["status"] != "success":
        raise RuntimeError(f"Failed to list items at {items_url}: {items['status']!r}.")
    return items[item_type]

search_items(self, query: str, item_types: Iterable[str] | str = ('samples', 'cells')) -> list[dict[str, Any]]

Search for items of the given types that match the query.

Parameters:

Name Type Description Default
query str

Free-text query to search for.

required
item_types Iterable[str] | str

The types of items to search for. Defaults to ["samples", "cells"].

('samples', 'cells')

Returns:

Type Description
list[dict[str, Any]]

An ordered list of items of the given types that match the query.

Source code in datalab_api/__init__.py
def search_items(
    self, query: str, item_types: Iterable[str] | str = ("samples", "cells")
) -> list[dict[str, Any]]:
    """Search for items of the given types that match the query.

    Parameters:
        query: Free-text query to search for.
        item_types: The types of items to search for. Defaults to ["samples", "cells"].

    Returns:
        An ordered list of items of the given types that match the query.

    """
    if isinstance(item_types, str):
        item_types = [item_types]

    search_items_url = (
        f"{self.datalab_api_url}/search-items?query={query}&types={','.join(item_types)}"
    )
    items_resp = self.session.get(search_items_url, follow_redirects=True)
    if items_resp.status_code != 200:
        raise RuntimeError(
            f"Failed to search items with {item_types=} at {search_items_url}: {items_resp.status_code=}"
        )
    items = items_resp.json()
    if items["status"] != "success":
        raise RuntimeError(f"Failed to list items at {search_items_url}: {items['status']!r}.")

    return items["items"]

create_item(self, item_id: str | None = None, item_type: str = 'samples', item_data: dict[str, Any] | None = None, collection_id: str | None = None) -> dict[str, Any]

Create an item with a given ID and item data.

Parameters:

Name Type Description Default
item_id str | None

The ID of the item to create, if set to None, the server will generate one.

None
item_type str

The type of item to create, e.g., 'samples', 'cells'.

'samples'
item_data dict[str, Any] | None

The data for the item.

None
collection_id str | None

The ID of the collection to add the item to (optional). If such a collection does not exist, one will be made.

None
Source code in datalab_api/__init__.py
def create_item(
    self,
    item_id: str | None = None,
    item_type: str = "samples",
    item_data: dict[str, Any] | None = None,
    collection_id: str | None = None,
) -> dict[str, Any]:
    """Create an item with a given ID and item data.

    Parameters:
        item_id: The ID of the item to create, if set to `None`, the server will generate one.
        item_type: The type of item to create, e.g., 'samples', 'cells'.
        item_data: The data for the item.
        collection_id: The ID of the collection to add the item to (optional).
            If such a collection does not exist, one will be made.

    """
    new_item = {}
    if item_data is not None:
        new_item = item_data
    new_item.update({"item_id": item_id, "type": item_type})

    if new_item["item_id"] is None:
        new_item.pop("item_id")

    if collection_id is not None:
        try:
            collection_immutable_id = self.get_collection(collection_id)["immutable_id"]
        except RuntimeError:
            self.create_collection(collection_id)
            collection_immutable_id = self.get_collection(collection_id)["immutable_id"]
        new_item["collections"] = new_item.get("collections", [])
        new_item["collections"].append({"immutable_id": collection_immutable_id})

    create_item_url = f"{self.datalab_api_url}/new-sample/"
    create_item_resp = self.session.post(
        create_item_url,
        json={"new_sample_data": new_item, "generate_id_automatically": item_id is None},
        follow_redirects=True,
    )
    try:
        created_item = create_item_resp.json()
        if create_item_resp.status_code == 409:
            raise DuplicateItemError(
                f"Item {item_id=} already exists at {create_item_url}: {created_item['status']!r}."
            )
        if created_item["status"] != "success":
            raise RuntimeError(f"Failed to create item at {create_item_url}: {created_item}.")
        return created_item["sample_list_entry"]

    except Exception as exc:
        raise exc.__class__(
            f"Failed to create item {item_id=} with data {item_data=} at {create_item_url}: {create_item_resp.status_code=}, {create_item_resp.content}. Check the item information is correct."
        ) from exc

update_item(self, item_id: str, item_data: dict[str, Any]) -> dict[str, Any]

Update an item with the given item data.

Parameters:

Name Type Description Default
item_id str

The ID of the item to update.

required
item_data dict[str, Any]

The new data for the item.

required

Returns:

Type Description
dict[str, Any]

A dictionary of the updated item data.

Source code in datalab_api/__init__.py
def update_item(self, item_id: str, item_data: dict[str, Any]) -> dict[str, Any]:
    """Update an item with the given item data.

    Parameters:
        item_id: The ID of the item to update.
        item_data: The new data for the item.

    Returns:
        A dictionary of the updated item data.

    """
    update_item_data = {"item_id": item_id, "data": item_data}
    update_item_url = f"{self.datalab_api_url}/save-item/"
    update_item_resp = self.session.post(
        update_item_url,
        json=update_item_data,
        follow_redirects=True,
    )
    if update_item_resp.status_code != 200:
        raise RuntimeError(
            f"Failed to update item {item_id=} at {update_item_url}: {update_item_resp.status_code=}. Check the item information is correct."
        )
    updated_item = update_item_resp.json()
    if updated_item["status"] != "success":
        raise RuntimeError(
            f"Failed to update item at {update_item_url}: {updated_item['status']!r}."
        )
    return updated_item

get_item(self, item_id: str | None = None, refcode: str | None = None, load_blocks: bool = False) -> dict[str, Any]

Get an item with a given ID or refcode.

Parameters:

Name Type Description Default
item_id str | None

The ID of the item to search for.

None
refcode str | None

The refcode of the item to search fork

None
load_blocks bool

Whether to load the blocks associated with the item.

False

Returns:

Type Description
dict[str, Any]

A dictionary of item data for the item with the given ID or refcode.

Source code in datalab_api/__init__.py
def get_item(
    self,
    item_id: str | None = None,
    refcode: str | None = None,
    load_blocks: bool = False,
) -> dict[str, Any]:
    """Get an item with a given ID or refcode.

    Parameters:
        item_id: The ID of the item to search for.
        refcode: The refcode of the item to search fork
        load_blocks: Whether to load the blocks associated with the item.

    Returns:
        A dictionary of item data for the item with the given ID or refcode.

    """

    if item_id is None and refcode is None:
        raise ValueError("Must provide one of `item_id` or `refcode`.")
    if item_id is not None and refcode is not None:
        raise ValueError("Must provide only one of `item_id` or `refcode`.")

    if refcode is not None:
        raise NotImplementedError("Searching by `refcode` is not yet implemented.")

    item_url = f"{self.datalab_api_url}/get-item-data/{item_id}"
    item_resp = self.session.get(item_url, follow_redirects=True)

    if item_resp.status_code != 200:
        raise RuntimeError(
            f"Failed to find item {item_id=}, {refcode=} {item_url}: {item_resp.status_code=}. Check the item information is correct."
        )

    item = item_resp.json()
    if item["status"] != "success":
        raise RuntimeError(f"Failed to get item at {item_url}: {item['status']!r}.")

    # Filter out any deleted blocks
    item["item_data"]["blocks_obj"] = {
        block_id: block
        for block_id, block in item["item_data"]["blocks_obj"].items()
        if block_id in item["item_data"]["display_order"]
    }

    # Make a call to `/update-block` which will parse/create plots and return as JSON
    if load_blocks:
        ret_item_id = item["item_data"]["item_id"]
        for block_id, block in item["item_data"]["blocks_obj"].items():
            block_data = self.get_block(
                item_id=ret_item_id, block_id=block_id, block_data=block
            )
            item["item_data"]["blocks_obj"][block_id] = block_data

    return item["item_data"]

get_item_files(self, item_id: str) -> None

Download all the files for a given item and save them locally in the current working directory.

Parameters:

Name Type Description Default
item_id str

The ID of the item to search for.

required
Source code in datalab_api/__init__.py
def get_item_files(self, item_id: str) -> None:
    """Download all the files for a given item and save them locally
    in the current working directory.

    Parameters:
        item_id: The ID of the item to search for.

    """

    item_data = self.get_item(item_id)
    for f in item_data.get("files", []):
        url = f"{self.datalab_api_url}/files/{f['immutable_id']}/{f['name']}"
        if Path(f["name"]).exists():
            warnings.warn(f"Will not overwrite existing file {f['name']}")
            continue
        with open(f["name"], "wb") as file:
            response = self.session.get(url, follow_redirects=True)
            file.write(response.content)

get_block(self, item_id: str, block_id: str, block_data: dict[str, Any]) -> dict[str, Any]

Get a block with a given ID and block data. Should be used in conjunction with get_item to load an existing block.

Parameters:

Name Type Description Default
item_id str

The ID of the item to search for.

required
block_id str

The ID of the block to search for.

required
block_data dict[str, Any]

Any other block data required by the request.

required

Returns:

Type Description
dict[str, Any]

A dictionary of block data for the block with the given ID.

Source code in datalab_api/__init__.py
def get_block(self, item_id: str, block_id: str, block_data: dict[str, Any]) -> dict[str, Any]:
    """Get a block with a given ID and block data.
    Should be used in conjunction with `get_item` to load an existing block.

    Parameters:
        item_id: The ID of the item to search for.
        block_id: The ID of the block to search for.
        block_data: Any other block data required by the request.

    Returns:
        A dictionary of block data for the block with the given ID.

    """
    block_url = f"{self.datalab_api_url}/update-block/"
    block_request = {
        "block_data": block_data,
        "item_id": item_id,
        "block_id": block_id,
        "save_to_db": False,
    }
    block_resp = self.session.post(block_url, json=block_request, follow_redirects=True)
    if block_resp.status_code != 200:
        raise RuntimeError(
            f"Failed to find block {block_id=} for item {item_id=} at {block_url}: {block_resp.status_code=}. Check the block information is correct."
        )

    block = block_resp.json()
    if block["status"] != "success":
        raise RuntimeError(f"Failed to get block at {block_url}: {block['status']!r}.")
    return block["new_block_data"]

upload_file(self, item_id: str, file_path: Path | str) -> dict[str, Any]

Upload a file to an item with a given ID.

Parameters:

Name Type Description Default
item_id str

The ID of the item to upload the file to.

required
file_path Path | str

The path to the file to upload.

required

Returns:

Type Description
dict[str, Any]

A dictionary of the uploaded file data.

Source code in datalab_api/__init__.py
def upload_file(self, item_id: str, file_path: Path | str) -> dict[str, Any]:
    """Upload a file to an item with a given ID.

    Parameters:
        item_id: The ID of the item to upload the file to.
        file_path: The path to the file to upload.

    Returns:
        A dictionary of the uploaded file data.

    """
    if isinstance(file_path, str):
        file_path = Path(file_path)
    if not file_path.exists():
        raise FileNotFoundError(f"File {file_path=} does not exist.")

    upload_url = f"{self.datalab_api_url}/upload-file/"
    with open(file_path, "rb") as file:
        files = {"file": (file_path.name, file)}
        upload_resp = self.session.post(
            upload_url,
            files=files,
            data={"item_id": item_id, "replace_file": None},
            follow_redirects=True,
        )
    if upload_resp.status_code != 201:
        raise RuntimeError(
            f"Failed to upload file {file_path=} to item {item_id=} at {upload_url}: {upload_resp.status_code=}. Check the file information is correct."
        )

    upload = upload_resp.json()
    if upload["status"] != "success":
        raise RuntimeError(f"Failed to upload file at {upload_url}: {upload['status']!r}.")

    return upload

create_data_block(self, item_id: str, block_type: str, file_ids: str | list[str] | None = None, file_paths: Path | list[Path] | None = None) -> dict[str, Any]

Creates a data block for an item with the given block type and file ID.

Parameters:

Name Type Description Default
item_id str

The ID of the item to create the block for.

required
block_type str

The type of block to create.

required
file_ids str | list[str] | None

The ID of the file to attach to the block, must already be uploaded to the item.

None
file_paths Path | list[Path] | None

A path, or set of paths, to files to upload and attach to the item. Conflicts with file_id, only one can be provided.

None

Returns:

Type Description
dict[str, Any]

The created block data.

Source code in datalab_api/__init__.py
def create_data_block(
    self,
    item_id: str,
    block_type: str,
    file_ids: str | list[str] | None = None,
    file_paths: Path | list[Path] | None = None,
) -> dict[str, Any]:
    """Creates a data block for an item with the given block type and file ID.

    Parameters:
        item_id: The ID of the item to create the block for.
        block_type: The type of block to create.
        file_ids: The ID of the file to attach to the block, must already
            be uploaded to the item.
        file_paths: A path, or set of paths, to files to upload and attach to the
            item. Conflicts with `file_id`, only one can be provided.

    Returns:
        The created block data.

    """

    blocks_url = f"{self.datalab_api_url}/add-data-block/"

    # TODO: future validation of block types based on server response
    payload = {
        "item_id": item_id,
        "block_type": block_type,
        "index": None,
    }

    if file_paths:
        raise NotImplementedError(
            "Simultaneously uploading files and creating blocks is not yet supported."
        )

    if file_ids:
        if isinstance(file_ids, str):
            file_ids = [file_ids]
        # check that the file is attached to the item
        item = self.get_item(item_id=item_id, load_blocks=False)
        attached_file_ids = set(item["file_ObjectIds"])
        if not set(file_ids).issubset(attached_file_ids):
            raise RuntimeError(
                f"Not all IDs {file_ids=} are attached to item {item_id=}: {attached_file_ids=}"
            )

    block_resp = self.session.post(blocks_url, json=payload, follow_redirects=True)
    if block_resp.status_code != 200:
        raise RuntimeError(
            f"Failed to create block {block_type=} for item {item_id=}:\n{block_resp.text}"
        )
    block_data = block_resp.json()["new_block_obj"]

    if file_ids:
        block_data = self._update_data_block(
            block_type=block_type, block_data=block_data, file_ids=file_ids
        )

    return block_data

update_data_block(self, item_id: str, block_id: str, block_type: str, block_data: dict) -> dict[str, Any]

Attempts to update a block with the given payload of data, returning the updated block.

Parameters:

Name Type Description Default
item_id str

The ID of the item that the block is attached to.

required
block_id str

The ID of existing block.

required
block_type str

The type of block to update.

required
block_data dict

The payload of block data to update; will be used to selectively update any fields present, with validation performed by the remote block code itself.

required

Returns:

Type Description
dict[str, Any]

If successful, the updated block data.

Exceptions:

Type Description
RuntimeError

If the block update fails.

Source code in datalab_api/__init__.py
def update_data_block(
    self, item_id: str, block_id: str, block_type: str, block_data: dict
) -> dict[str, Any]:
    """Attempts to update a block with the given payload of data,
    returning the updated block.

    Parameters:
        item_id: The ID of the item that the block is attached to.
        block_id: The ID of existing block.
        block_type: The type of block to update.
        block_data: The payload of block data to update; will be used to selectively update
            any fields present, with validation performed by the remote block code itself.

    Returns:
        If successful, the updated block data.

    Raises:
        RuntimeError: If the block update fails.

    """
    payload = {k: v for k, v in block_data.items()}
    payload["block_id"] = block_id
    payload["blocktype"] = block_type
    payload["item_id"] = item_id

    return self._update_data_block(block_type=block_type, block_data=payload)

get_collection(self, collection_id: str) -> dict[str, Any]

Get a collection with a given ID.

Parameters:

Name Type Description Default
collection_id str

The ID of the collection to search for.

required

Returns:

Type Description
dict[str, Any]

A dictionary of collection data for the collection with the given ID.

Source code in datalab_api/__init__.py
def get_collection(self, collection_id: str) -> dict[str, Any]:
    """Get a collection with a given ID.

    Parameters:
        collection_id: The ID of the collection to search for.

    Returns:
        A dictionary of collection data for the collection with the given ID.

    """
    collection_url = f"{self.datalab_api_url}/collections/{collection_id}"
    collection_resp = self.session.get(collection_url, follow_redirects=True)
    if collection_resp.status_code != 200:
        raise RuntimeError(
            f"Failed to find collection {collection_id=} at {collection_url}: {collection_resp.status_code=}. Check the collection ID is correct."
        )
    collection = collection_resp.json()
    if collection["status"] != "success":
        raise RuntimeError(
            f"Failed to get collection at {collection_url}: {collection['status']!r}."
        )
    return collection["data"]

create_collection(self, collection_id: str, collection_data: dict | None = None) -> dict[str, Any]

Create a collection with a given ID and collection data.

Parameters:

Name Type Description Default
collection_id str

The ID of the collection to create.

required
collection_data dict | None

The data for the collection.

None
Source code in datalab_api/__init__.py
def create_collection(
    self, collection_id: str, collection_data: dict | None = None
) -> dict[str, Any]:
    """Create a collection with a given ID and collection data.

    Parameters:
        collection_id: The ID of the collection to create.
        collection_data: The data for the collection.

    """
    collection_url = f"{self.datalab_api_url}/collections"

    new_collection = {}
    if collection_data is not None:
        new_collection = collection_data
    new_collection.update({"collection_id": collection_id, "type": "collections"})

    collection_resp = self.session.put(
        collection_url,
        json={"data": new_collection},
        follow_redirects=True,
    )

    if collection_resp.status_code != 201 or collection_resp.json()["status"] != "success":
        raise RuntimeError(
            f"Failed to create collection {collection_id=} at {collection_url}: {collection_resp.status_code=}. Check the collection information is correct: {collection_resp.content}"
        )

    created_collection = collection_resp.json()["data"]
    return created_collection

cli

app

console

launch(ctx: Context, instance_url: Annotated[Optional[str], <typer.models.ArgumentInfo object at 0x7f710b5d9e40>] = None, animate_intro: bool = True)

Makes an interactive REPL-style interface using the subcommands below.

Source code in datalab_api/cli.py
@app.callback(invoke_without_command=True)
def launch(
    ctx: typer.Context,
    instance_url: Annotated[Optional[str], typer.Argument()] = None,
    animate_intro: bool = True,
):
    """Makes an interactive REPL-style interface using the subcommands below."""
    shell = make_click_shell(
        ctx,
        prompt="datalab > ",
    )

    animation_intro = _make_fancy_intro()
    if animate_intro:
        with Live(console=console, auto_refresh=False) as live:
            for frame in animation_intro:
                live._live_render.set_renderable(
                    Panel(
                        frame,
                        subtitle=app.info.epilog,
                        width=len(app.info.epilog) + 6,
                        highlight=False,
                    )
                )  # type: ignore
                console.print(live._live_render.position_cursor())
                time.sleep(0.05)

    console.print(
        Panel(animation_intro[-1], subtitle=app.info.epilog, width=len(app.info.epilog) + 6),
        highlight=False,
    )  # type: ignore
    console.print()
    console.print(
        Panel(
            "This CLI is an experimental work in progress and does not expose the full functionality of the underlying DatalabClient.",
            title="[red]WARNING![/red]",
            width=len(app.info.epilog) + 6,
        )
    )
    console.print()

    if instance_url:
        authenticate(ctx, instance_url)
    shell.cmdloop()

authenticate(ctx: Context, instance_url: Annotated[Optional[str], <typer.models.ArgumentInfo object at 0x7f710b5d9db0>] = None, log_level: str = 'WARNING')

Source code in datalab_api/cli.py
@app.command()
def authenticate(
    ctx: typer.Context,
    instance_url: Annotated[Optional[str], typer.Argument()] = None,
    log_level: str = "WARNING",
):
    client = _get_client(ctx, instance_url, log_level)
    user = client.authenticate()
    console.print(
        f"Welcome [red]{user['display_name']}[/red]!\nSuccessfully authenticated at [blue]{client.datalab_api_url}[/blue]."
    )

get(ctx: Context, item_type: str, instance_url: Annotated[Optional[str], <typer.models.ArgumentInfo object at 0x7f710b5d9f30>] = None, page_limit: int = 10, log_level: str = 'WARNING')

Get a table of items of the given type.

Source code in datalab_api/cli.py
@app.command()
def get(
    ctx: typer.Context,
    item_type: str,
    instance_url: Annotated[Optional[str], typer.Argument()] = None,
    page_limit: int = 10,
    log_level: str = "WARNING",
):
    """Get a table of items of the given type."""
    client = _get_client(ctx, instance_url, log_level)
    items = client.get_items(item_type)
    table = Table(title=f"/{item_type}/", show_lines=True)
    table.add_column("type", overflow="crop", justify="center")
    table.add_column("ID", style="cyan", no_wrap=True)
    table.add_column("refcode", style="magenta", no_wrap=True)
    table.add_column("name", width=30, overflow="ellipsis", no_wrap=True)
    table.add_column("nblocks", justify="right")
    table.add_column("collections")
    table.add_column("creators")
    for item in items[:page_limit]:
        table.add_row(
            item["type"][0].upper(),
            item["item_id"],
            item["refcode"],
            item["name"],
            str(item["nblocks"]),
            ", ".join(d["collection_id"] for d in item["collections"]),
            ", ".join(d["display_name"] for d in item["creators"]),
            end_section=True,
        )
    console.print(table)

info(ctx: Context, instance_url: str, log_level: str = 'WARNING')

Print the server info.

Source code in datalab_api/cli.py
@app.command()
def info(ctx: typer.Context, instance_url: str, log_level: str = "WARNING"):
    """Print the server info."""
    client = _get_client(ctx, instance_url, log_level)
    pprint(client.get_info())

helpers

import_cheminventory(filename: Path, client: DatalabClient)

Helper function to import a ChemInventory xlsx export into datalab, using the 'starting_materials' item type.

Migrated from the original version in the main datalab repository: https://github.com/the-grey-group/datalab/blob/43764fb494c2cc1bf9f7dc90c25594aeb79d5767/pydatalab/tasks.py#L350-L413

Source code in datalab_api/helpers.py
def import_cheminventory(filename: Path, client: DatalabClient):
    """Helper function to import a ChemInventory xlsx export into datalab, using the
    'starting_materials' item type.

    Migrated from the original version in the main datalab repository:
        https://github.com/the-grey-group/datalab/blob/43764fb494c2cc1bf9f7dc90c25594aeb79d5767/pydatalab/tasks.py#L350-L413

    """

    def _generate_random_startingmaterial_id():
        """Generate 'XX' + a random 15-length string for use as an id for starting materials
        that don't have a barcode.
        """
        yield "".join(["XX"] + random.choices("abcdefghijklmnopqrstuvwxyz0123456789", k=15))

    try:
        import pandas as pd

        inventory_df = pd.read_excel(filename)
    except ImportError as exc:
        raise ImportError(
            "Please install pandas + openpyxl to use this helper function, via `pip install datalab-api[cheminventory-helper]`"
        ) from exc
    inventory_df["type"] = "starting_materials"
    inventory_df["item_id"] = inventory_df["Barcode"]
    # Fill missing barcodes
    inventory_df["item_id"] = inventory_df["item_id"].fillna(
        inventory_df["item_id"].apply(lambda _: next(_generate_random_startingmaterial_id()))
    )
    inventory_df["Molecular Weight"] = inventory_df["Molecular Weight"].replace(" ", float("nan"))

    counts = {"success": 0, "duplicate": 0, "failed": 0}

    for item in inventory_df.to_dict(orient="records"):
        try:
            client.create_item(
                item["item_id"], item["type"], item, collection_id="jul24-inventory-import"
            )
            counts["success"] += 1
        except DuplicateItemError:
            counts["duplicate"] += 1
        except Exception as exc:
            counts["failed"] += 1
            print(f"Failed to import item: {item}. Error: {exc}")
            continue

    print(f"Done: {counts=}")