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 |
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 |
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 |
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=}")