Common SDK Interactions and Examples

Common behaviors you’ll use across most SDK interactions. This builds on the Getting Started with the V3 Python SDK quickstart.

Pagination

List endpoints return results one page at a time. Iterate the returned pager to get every item. The SDK fetches each page as you go:

for dna_sequence in benchling.dna_sequence.list():
    print(dna_sequence.id, dna_sequence.name)

To work with a page at a time, use iter_pages():

for page in benchling.dna_sequence.list().iter_pages():
    print(len(page.items))

Some objects have sub-resources you can list. Many return the full set at once, so you access items directly:

for option in benchling.dropdown.list_options(dropdown_id="<dropdown-id>").items:
    print(option.id, option.name)

Others are paginated like list(), so iterate them instead. Each method’s return type says which: a SyncPager when it’s paginated, and a <Type>UnpaginatedList when it isn’t.

Batch operations

Most resources have a batch_ method for each write they support. Each takes its items inline and returns them in items:

from benchling_v3_sdk import CreateDnaSequenceInput
response = benchling.dna_sequence.batch_create(
    items=[
        CreateDnaSequenceInput(name="seq-1", bases="ATGCATGCAT", folder_id="<folder-id>"),
        CreateDnaSequenceInput(name="seq-2", bases="GGGCCCAATT", folder_id="<folder-id>"),
    ],
)
for dna_sequence in response.items:
    print(dna_sequence.id, dna_sequence.name)

Each request runs in one transaction and is capped, so either every item is written or none is. A rejected item fails the whole request.

batch_update and batch_delete carry each object’s ID inside the item:

from benchling_v3_sdk import UpdateDnaSequenceInputWithPathParams
benchling.dna_sequence.batch_update(
    items=[UpdateDnaSequenceInputWithPathParams(id="<dna-sequence-id>", name="seq-1-renamed")],
)

For more items than a request allows, use the resource’s bulk method, which runs asynchronously as a task.

Async tasks

Long-running operations don’t return a result directly. They return an AsyncTaskLink, which carries the ID of the task Benchling started.

For a function’s specific behavior, check the resource’s methods in the API reference.

Polling a task

To check a task’s status, pass the task ID to the getter for the operation you started. Each reports status, plus the operation’s output in result once the task finishes:

export_link = benchling.audit_log.export(format="CSV", object_id="<object-id>")
task = benchling.audit_log.get_task_for_export(task_id=export_link.task_id)
print(task.status)
if task.status == "COMPLETED":
    print(task.result.download_url)

More commonly, you’ll want to wait until the task finishes.

There are two ways to wait. wait_for_task_response blocks until the task finishes and returns the final poll payload, the same response the task getters return. It raises TaskFailureException unless the task fully succeeds:

task = benchling.wait_for_task_response(export_link)

Catch it to handle a task that didn’t succeed:

from benchling_v3_sdk import TaskFailureException
try:
    task = benchling.wait_for_task_response(export_link)
except TaskFailureException as error:
    print("The task didn't succeed:", error)

wait_for_task_completion waits the same way but doesn’t raise. It returns a TaskCompletion describing how the task ended. Finishing isn’t the same as succeeding, so check success:

completion = benchling.wait_for_task_completion(export_link)
if completion.success:
    print("The task succeeded")
else:
    print("The task failed:", completion.message)

message describes the outcome. The task’s payload is on completion.response.

By default the SDK polls every second for up to 600 seconds. You can override both the polling interval and the total wait with interval_wait_seconds and max_wait_seconds:

task = benchling.wait_for_task_response(export_link, interval_wait_seconds=2, max_wait_seconds=60)

If the task is still running when the time runs out, wait_for_task_response and wait_for_task_completion both raise WaitForTaskTimeoutError:

from benchling_v3_sdk import WaitForTaskTimeoutError
try:
    task = benchling.wait_for_task_response(export_link, max_wait_seconds=60)
except WaitForTaskTimeoutError:
    print("The task is still running")

To run your own code on every poll, pass on_poll. It receives the same response that task’s getter returns:

def report(task):
    print(task.status)
task = benchling.wait_for_task_response(export_link, on_poll=report)

An exception from the callback or a failed poll ends the wait while the task keeps running.

Bulk operations

Bulk methods are the asynchronous form of a resource’s batch_ methods. They’re all polled with bulk_task.get, and a finished task can report item-level failures.

The input is a scratch file with one item per line:

import json
rows = [
    {"request_id": "1", "input_object": {"name": "seq-1", "bases": "ATGCATGCAT", "folderId": "<folder-id>"}},
    {"request_id": "2", "input_object": {"name": "seq-2", "bases": "GGGCCCAATT", "folderId": "<folder-id>"}},
]
scratch_file = benchling.scratch_file.create_and_upload(
    contents="\n".join(json.dumps(row) for row in rows) + "\n",
    filename="bulk.jsonl",
)

Pass the file’s ID to the bulk method, then poll the returned task:

link = benchling.dna_sequence.bulk_create(file_id=scratch_file.id)
task = benchling.bulk_task.get(task_id=link.task_id)

The response reports the task’s status and how far it has progressed:

print(task.status)  # NOT_STARTED, IN_PROGRESS, COMPLETED, or FAILED
print(task.progress.number_processed, "of", task.progress.total_records)
print(task.result.success_count, "succeeded,", task.result.error_count, "failed")

A COMPLETED bulk task is not necessarily one in which every item succeeded. Item failures appear in result.error_count.

wait_for_task_response treats a bulk task as successful only when every item succeeded, so a task that finished with any errors raises rather than returns. wait_for_task_completion reports the same case as partial:

completion = benchling.wait_for_task_completion(link)
if completion.success:
    print("Every item succeeded")
elif completion.partial:
    print("Some items failed:", completion.message)
else:
    print("The task failed:", completion.message)

For a partial bulk task, message includes the success and error counts.

Error handling

A failed request raises ApiError:

from benchling_v3_sdk.core.api_error import ApiError
from benchling_v3_sdk.errors import NotFoundError
try:
    benchling.dna_sequence.get(dna_sequence_id="<id>")
except NotFoundError:
    print("No sequence with that ID")
except ApiError as error:
    print(error.status_code, error.body)

Every ApiError exposes status_code, headers, and body. Some statuses also have a subclass in benchling_v3_sdk.errors, such as NotFoundError for 404 and TooManyRequestsError for 429. Not all do. A 401 arrives as ApiError.

If a response isn’t returned to the SDK, an exception other than ApiError could be raised.

Retries

The SDK automatically retries failed requests with exponential backoff and jitter. It retries HTTP 408, 409, 429, 502, 503, and 504, plus failed and dropped connections. Nothing else is retried, including timeouts: a request that exceeds the timeout, 60 seconds by default, raises instead.

If a retried response carries a Retry-After header, the SDK waits for that instead, up to 60 seconds.

The default is 2 retries. Set max_retries on the client to change it:

benchling = Benchling(
    base_url="https://yourcompany.benchling.com",
    api_key="<your-api-key>",
    early_access="true",
    max_retries=5,
)

To override for a single call, pass max_retries in request_options:

benchling.dna_sequence.list(request_options={"max_retries": 0})

A value of 0 disables retries.

Rate limiting

When Benchling rate-limits a request, it responds with HTTP 429. The SDK retries 429 up to max_retries, waiting longer after each attempt with exponential backoff and jitter, up to about a minute between attempts. If the retries are exhausted, it raises TooManyRequestsError.