Common SDK Interactions and Examples
This document covers high level interactions with code samples that are commonly used in the SDK, as well as known areas where it can be confusing to use. Please use this as a reference for understanding patterns in the SDK.
Examples in this document assume you have created a Benchling client
The examples assume that your Benchling client is named
benchling
. See Getting Started with the SDK.
Entity Interactions
Creation
One of the most basic interactions with Benchling is creating an unregistered DNA sequence entity. This requires only that you provide the folder in which you want the entity to reside. Unlike registering entities directly, this will not enforce any required fields, as we are not required to pre-select a specific schema at this time. This is ideal when more details will be added to the entities at a later date. A single schema field, species
, is included on this entity, using the serialization helper method fields()
, which is covered in more detail below.
from benchling_sdk.benchling import Benchling
from benchling_sdk.models import DnaSequenceCreate
from benchling_sdk.helpers.serialization_helpers import fields
# Here we instantiate the object to be created
entityToCreate = DnaSequenceCreate(
folder_id="lib_xyz",
name="test123",
is_circular=False,
fields=fields({
"species": {"value": "mouse"}
}),
bases="cgatctgagtctaaaaacgtactgagtcgcg"
)
# Next we use the create function to serialize the data and call the API
benchling.dna_sequences.create(entityToCreate)
Updates
A common case for updating an existing entity is setting one or more fields on the entity. When performing such an update, we should only specify fields which we want to replace. Other fields on the existing entity will be preserved.
from benchling_sdk.benchling import Benchling
from benchling_sdk.models import DnaSequenceUpdate
from benchling_sdk.helpers.serialization_helpers import fields
# The ID of an existing entity we want to update
existing_sequence_id = "seq_SmiFihRA"
# Specify the attributes of the entity to update
# Unspecified fields will not be altered
entityToUpdate = DnaSequenceUpdate(
fields=fields({
# Set this field to a new value
"species": {"value": "horse"},
# Blank out the value for this field
"comment": {"value": None}
})
)
# Next we use the update function to serialize the data and call the API
benchling.dna_sequences.update(dna_sequence_id=existing_sequence_id, dna_sequence=entityToUpdate)
Registration
At times, you may want to create an entity directly into the registry instead of calling create and register separately. The process for doing this is defined in the documentation for Create endpoints. You will need to provide the registry_id
and entity_registry_id
OR the registry_id
and a naming_strategy
.
You may not specify both an entity_registry_id
and a naming_strategy
when creating into the registry.
The following code snippet creates an entity into the registry by providing the registry_id
and the entity_registry_id
:
from benchling_sdk.benchling import Benchling
from benchling_sdk.models import CustomEntityCreate
from benchling_sdk.helpers.serialization_helpers import fields
from benchling_api_client.models.naming_strategy import NamingStrategy
entity = CustomEntityCreate(
fields=fields({
"Wavelength (nm)": {"value": "280"}
}),
folder_id="folder_id",
name="entity_name",
registry_id="registry_id",
entity_registry_id="entity_registry_id",
aliases=['testAlias'],
schema_id="schema_id"
)
benchling.custom_entities.create(entity)
OR
The following code snippet creates an entity into the registry by providing the registry_id
and the naming_strategy
:
from benchling_sdk.benchling import Benchling
from benchling_sdk.models import CustomEntityCreate
from benchling_sdk.helpers.serialization_helpers import fields
from benchling_api_client.models.naming_strategy import NamingStrategy
entity = CustomEntityCreate(
fields=fields({
"Wavelength (nm)": {"value": "280"}
}),
folder_id="folder_id",
name="entity_name",
registry_id="registry_id",
naming_strategy=NamingStrategy.NEW_IDS,
aliases=['testAlias'],
entities.schema_id="schema_id"
)
benchling.custom_entities.create(entity)
Iterating through Pages
Many list()
endpoints in the Benchling SDK do not return the entire response all at once, but instead implement a python generator, which returns a paginated list of objects instead. This decreases memory consumption in the case of large queries, and reduces latency for the initial response. In the example below, that generator is passed to an iterating function that loops through those pages and then loops through those individual entities in turn.
from benchling_sdk.benchling import Benchling
from benchling_sdk.auth.api_key_auth import ApiKeyAuth
benchling = Benchling(url="https://my.benchling.com", auth_method=ApiKeyAuth("api_key"))
def print_sequences(seq_list):
for page in seq_list:
for sequence in page:
print(f"name: {sequence.name}\nid:{sequence.id}\n")
dna_sequence_generator = benchling.dna_sequences.list()
print_sequences(dna_sequence_generator)
The estimated count of entities or entries will also be returned as a header on the API response, which can be surfaced using the estimated_count()
method on the generator, such as shown here:
dna_sequence_generator = benchling.dna_sequences.list()
print(dna_sequence_generator.estimated_count)
Working with Schema Fields
Many objects in Benchling have the concept of fields. They are represented in the SDK via the benchling_sdk.models.Fields
class.
To conveniently construct Fields
from a dictionary, we have provided a fields method in the serialization_helper
module:
from benchling_sdk.helpers.serialization_helpers import fields
from benchling_sdk.models import CustomEntityCreate
entity = CustomEntityCreate(
name="My Entity",
fields=fields({
"a_field": {"value": "First value"},
"second_field": {"value": "Second value"},
})
)
Async Tasks
Many Benchling endpoints that perform expensive operations launch Tasks. These are modeled in the SDK as benchling_sdk.models.AsyncTask
.
To simply retrieve the status of a task given its id:
async_task = benchling.tasks.get_by_id("task_id")
This will return the AsyncTask
object, which may still be in progress. More commonly, it may be desirable to delay further execution until a task is completed.
In this case, you may block further processing until the task is no longer RUNNING
:
completed_task = benchling.tasks.wait_for_task("task_id")
The wait_for_task
method will return the task once its status is no longer RUNNING
. This does not guarantee the task executed successfully (see benchling_sdk.models.AsyncTaskStatus
), only that Benchling considers it complete.
wait_for_task
can be configured by optionally specifying interval_wait_seconds
for the time to wait between calls and max_wait_seconds
which is the maximum number of seconds before wait_for_task
will give up and raise benchling_sdk.errors.WaitForTaskExpiredError
.
Example: Check the task status once every 2 seconds for up to 60 seconds
completed_task = benchling.tasks.wait_for_task(task_id="task_id", interval_wait_seconds=2, max_wait_seconds=60)
Retries
The SDK will automatically retry certain HTTP calls when the calls fail and certain conditions are met.
The default strategy is to retry calls failing with HTTP status codes 429
, 502
, 503
, and 504
. The rationale for these status codes being retried is that many times they are indicative of a temporary network failure or exceeding the rate limit and may be successful upon retry.
Retries will be attempted up to 5 times, with an exponential time delay backoff between calls.
To disable retries, specify None
for retry_strategy
when constructing Benchling:
benchling = Benchling(url="https://my.benchling.com", auth_method=ApiKeyAuth("api_key"), retry_strategy=None)
Alternatively, instantiate your own benchling_sdk.helpers.retry_helpers.RetryStrategy
to further customize retry behavior.
Adjusting Retry Parameters
The default parameters of the SDK retry logic is meant to accommodate the normal request limits, but if you're doing a bulk data ingestion, you might run into the throughput limits that can require more aggressive backoff. If you're looking to "fire and forget" a larger ingestion, you can adjust the retry strategy parameters to back off more quickly.
Default parameters are:
max_tries: typing.Optional[int] = 5, backoff_factor: float = 1.0
Increasing max_tries
determines how many times the client will retry before raising the 429 as a blocking error.
backoff_factor
is a constant scaling factor used to determine how long to wait between subsequent retries. It is c
in the formula c * 2 ^ n
(n
= what number retry attempt).
Here is an example for how to adjust those parameters:
from benchling_sdk.benchling import Benchling
from benchling_sdk.auth.client_credentials_oauth2 import ClientCredentialsOAuth2
from benchling_sdk.helpers.retry_helpers import RetryStrategy
client_id="id_here"
client_secret="secret_here"
auth_method = ClientCredentialsOAuth2(client_id=client_id, client_secret=client_secret)
benchling = Benchling(
url="https://{tenant_name}.benchling.com",
auth_method=auth_method,
retry_strategy=RetryStrategy(max_tries=20, backoff_factor=2.0)
)
Error Handling
Failed API interactions will generally return a BenchlingError
, which will contain some underlying information on the HTTP response such as the status. Example:
from benchling_sdk.errors import BenchlingError
try:
requests = benchling.requests.get_by_id("request_id")
except BenchlingError as error:
print(error.status_code)
If an HTTP error code is not returned to the SDK or deserialization fails, an unbounded Exception could be raised instead.
Using the returning
parameter
returning
parameterFor some API endpoints, the returning
parameter enables finer control over the data returned by the Benchling API. For these endpoints, the corresponding SDK methods also support the returning parameter; using the returning parameter in the SDK involves passing an iterable of strings, where each string corresponds to the fields being requested.
For example, the following code would produce an iterable of plate objects with only the name
and id
fields for each plate:
plates = benchling.plates.list(returning=["plates.name","plates.id"]):
Fields that are not requested using the returning
parameter are set to UNSET
; for more information, see the section below on the unset
type.
Interacting with API calls not yet supported in the SDK
For making customized API calls to Benchling, the SDK supports an open-ended Benchling.api
namespace that exposes varying levels of interaction for HTTP GET
, POST
, PATCH
, and DELETE
methods.
This is useful for API endpoints which the SDK may not support yet or for more granular control at the HTTP layer.
For each verb, there are two related methods. Using GET as an example:
get_response()
- Returns abenchling_api_client.types.Response
which has been parsed to a JSON dict and is slightly more structured.get_modeled()
- Returns any custom model which extendsbenchling_sdk.helpers.serialization_helpers.DeserializableModel
and must be a Python@dataclass
.
Both will automatically retry failures according to RetryStrategy
and will marshall errors to BenchlingError
.
When calling any of the methods in Benchling.api
, specify the full path to the URL except for the scheme and server. This differs from other API services, which will prepend the URL with a base_path
.
For example, if wishing to call an endpoint https://my.benchling.com/api/v2/custom-entities?some=param
, pass api/v2/custom-entities?some=param
for the url.
Here's an example of making a custom call with post_modeled()
:
from dataclasses import dataclass, field
from typing import Any, Dict
from dataclasses_json import config
from benchling_sdk.helpers.serialization_helpers import DeserializableModel, SerializableModel
@dataclass
class ModeledCustomEntityPost(SerializableModel):
name: str
fields: Dict[str, Any]
# If the property name in the API JSON payload does not match the Python attribute, use
# field and config to specify the appropriate name for serializing/deserializing
folder_id: str = field(metadata=config(field_name="folderId"))
schema_id: str = field(metadata=config(field_name="schemaId"))
@dataclass
class ModeledCustomEntityGet(DeserializableModel):
id: str
name: str
fields: Dict[str, Any]
folder_id: str = field(metadata=config(field_name="folderId"))
# Assumes `benchling` is already configured and instantiated as `Benchling`
body = ModeledCustomEntityPost(
name="My Custom Entity Model",
folder_id="folder_id",
schema_id="schema_id",
fields={"My Field": {"value": "Modeled Entity"}},
)
created_entity = benchling.api.post_modeled(
url="api/v2/custom-entities", body=body, target_type=ModeledCustomEntityGet
)
The returned created_entity
will be of type ModeledCustomEntityGet
. Classes extending SerializableModel
and DeserializableModel
will inherit serialize()
and deserialize()
methods respectively which will act on Python class attributes by default. These can be overridden for more customized serialization needs.
Customization of the API Client
While the SDK abstracts most of the HTTP transport layer, access can still be granted via the BenchlingApiClient
. A common use case might be extending HTTP timeouts for all calls.
This can be achieved by specifying a function which accepts a default configured instance of BenchlingApiClient
and returns a mutated instance with the desired changes.
For example, to set the HTTP timeout to 180 seconds:
from benchling_api_client.benchling_client import BenchlingApiClient
def higher_timeout_client(client: BenchlingApiClient) -> BenchlingApiClient:
return client.with_timeout(180)
benchling = Benchling(
url="https://my.benchling.com",
auth_method=ApiKeyAuth("api_key"),
client_decorator=higher_timeout_client,
)
To fully customize the BenchlingApiClient
and ignore default settings, construct your own instance in lieu of modifying the client argument.
Using self-signed certificates
When the SDK is used behind a proxy with a self-signed certificate, you may receive an error :self-signed certificate in certificate chain
. To resolve this, you can modify the httpx
client to include the trusted root certificate:
from httpx import Client
from benchling_sdk.benchling import Benchling
from benchling_sdk.auth.client_credentials_oauth2 import ClientCredentialsOAuth2
httpx_client = Client(
verify="/path/to/root-cert.pem"
)
auth_method = ClientCredentialsOAuth2(
client_id = "client-id",
client_secret = "client-secret",
httpx_client = httpx_client,
)
benchling = Benchling(
url="https://my.benchling.com",
auth_method=auth_method,
httpx_client=httpx_client,
)
The Unset type
The Benchling SDK uses the type benchling_api_client.types.Unset
and the constant value benchling_api_client.types.UNSET
to represent values that were not present in an interaction with the API. This is to distinguish from values that were explicitly set to None
from those that were simply unspecified.
A common example might be updating only specific properties of an object:
from benchling_sdk.models import CustomEntityUpdate
update = CustomEntityUpdate(name="New name")
updated_entity = benchling.custom_entities.update(entity_id="entity_id", entity=update)
All other properties of CustomEntityUpdate
besides name will default to UNSET
and not be sent with the update. Setting any optional property to None
will send a null JSON value. In general, you should not need to set UNSET
directly.
When receiving objects from the API, some of their fields may be Unset. The SDK will raise a benchling_api_client.extensions.NotPresentError
if a field which is Unset is accessed, so that type hinting always reflects the type of the field without needing to Union with the Unset type. When constructing objects, if you need to set a field to Unset after its construction, properties which are optional support the python del
keyword, e.g.:
from benchling_sdk.models import CustomEntityUpdate
update = CustomEntityUpdate(name="New name", folder_id="folder_id")
del update.folder_id
If the property cannot be Unset but del is used on it, Python will raise with AttributeError: can't delete attribute.
If you happen to have an instance of Unset that you'd like to treat equivalent to Optional[T], you can use the convenience function unset_as_none()
:
from typing import Union
from benchling_sdk.helpers.serialization_helpers import unset_as_none
sample_value: Union[Unset, None, int] = UNSET
optional_value = unset_as_none(sample_value)# optional_value will be None
Forward-Compatibility in Enums
When new functionality or types are added to the API, that may come along with a new enum value in a type that already existed. For instance, if a new type of Blob is added to the API, the GET /blobs/<blob_id>
endpoint would have an additional possible value under type.
In the general case, Python would raise ValueError
when trying to construct an Enum type with a value which is not valid for that type. If this happens during deserialization of a response, it would both prevent deserialization of the response for other fields, and prevent normal program behavior which doesn't care about the value.
Enums in the SDK are implemented in a way that will be forwards-compatible. When deserializing responses, if a value being received is not known for its destination Enum type, an ad-hoc class with the same name as the enum will be created with a singular member _UNKNOWN
, and that _UNKNOWN
member will be returned.
- Multiple deserializations of this value will always result in the same instance being returned.
- Enums in the model package are defined with mixin classes, which define the
known()
member. It returns bool and isTrue
if the value is part of the defined values in thatEnum
, andFalse
if it isn't, as in the case of deserializing it in the above case. - Accessing the raw value of the
_UNKNOWN
value is still possible via the.value
member.
Forward-Compatibility for Polymorphic Types
Oftentimes, for example in listing endpoints, many different types of an entity can be returned. As new features are added, new types can be returned from that same endpoint, but given that the installed version of the SDK may not have been updated to know how to deserialize them, this would normally cause an error during deserialization. This prevents normal program flow in cases which could be otherwise written defensively: none of the fields are programmatically accessible if deserialization fails, but programs may opt to handle errors or use the information they do know how to work with if serialization is made to be more 'nice.'
To remedy this, for objects with fields that can be one of any given number of types, the UnknownType is also a possibility. If a new type is added to an endpoint that could return one of many other types, but the installed version of the SDK has not been updated to recognize it, the type will be returned as an instance of UnknownType.
If the value returned from a field is an instance of UnknownType, the raw value received is accessible via its value
member.
Updated 3 months ago