Python SDK
dalea by example — sync + async across all four domains.
dalea is the official public Python SDK — a backend/agent-focused client
with a pooled httpx transport and sync + async on one resource tree,
covering the data, documents, inventory, and storage domains. It is
contract-aligned with the TypeScript @dalea/sdk:
both are generated from the same public OpenAPI subset. This page is the
by-example tour; shared concepts live in the
Dalea SDK overview, and every operation in the
API reference shows a ready-to-copy dalea snippet.
Install
pip install dalea
Python 3.11+. Ships type hints (PEP 561), so editors and type checkers see the full surface.
Quickstart
Set DALEA_API_KEY (a workspace-scoped key) and construct a client. Use it as a
context manager so the HTTP pool is cleaned up.
from dalea import DaleaClient
with DaleaClient(
base_url="https://dalea.app",
api_key="dalea_...", # or omit → read from DALEA_API_KEY
) as dalea:
env = dalea.data.environments.list()["environments"][0]
print(env["name"])
The API key already pins the workspace, so that's all you need.
Every resource method returns the parsed JSON response (a dict / list),
exactly like the TS SDK's runtime return.
Async
The same API, awaited — AsyncDaleaClient shares the resource tree:
import asyncio
from dalea import AsyncDaleaClient
async def main():
async with AsyncDaleaClient(
base_url="https://dalea.app", api_key="dalea_..."
) as dalea:
table = await dalea.data.tables.get("table-uuid")
asyncio.run(main())
Credentials
Auth resolves in this precedence: explicit credential → api_key → the
DALEA_API_KEY environment variable. There is no session-cookie option — that
is a browser concern, so it exists only in the TypeScript SDK. See the
overview for when to use each
credential, and Authentication for obtaining them.
import os
from dalea import DaleaClient, OAuthPkceCredential, OAuthClientCredentialsCredential
# API key — the default for scripts and backends. With DALEA_API_KEY set:
dalea = DaleaClient(base_url="https://dalea.app")
# OAuth 2.1 + PKCE — acting on behalf of a user (auto-refresh on 401):
cred = OAuthPkceCredential.exchange_code(
token_endpoint="https://dalea.app/api/v1/auth/oauth2/token",
client_id="my-app",
code=code, # from the ?code=… redirect
code_verifier=code_verifier,
redirect_uri="https://myapp.example/callback",
on_tokens=lambda tokens: persist(tokens), # called on every rotation
)
dalea = DaleaClient(base_url="https://dalea.app", credential=cred)
# OAuth 2.1 client-credentials — machine-to-machine:
m2m = DaleaClient(
base_url="https://dalea.app",
credential=OAuthClientCredentialsCredential(
token_endpoint="https://dalea.app/api/v1/auth/oauth2/token",
client_id="my-service",
client_secret=os.environ["OAUTH_CLIENT_SECRET"],
),
)
Workspace context
API keys and OAuth credentials are bound to a workspace when created, so you
normally never set one. For credentials that don't pin a workspace, the client
sends an X-Workspace header: set a default (workspace=... or
dalea.set_workspace(id)), or override per call — the per-call value wins:
dalea.data.tables.get("t1", workspace="other-workspace-id")
Data
Walk your workspace's data: an environment → its first table → that table's size.
env = dalea.data.environments.list()["environments"][0]
table = dalea.data.environments.list_tables(env["id"])["tables"][0]
page = dalea.data.tables.list_objects(table["id"], {"limit": 50})
print(f'{env["name"]} → {table["name"]}: {page["total"]} objects')
Documents
Create a document, write its body as markdown, and read the whole document back as markdown:
doc = dalea.documents.create({
"workspaceId": "ws_123",
"title": "Assay run notes",
})
dalea.documents.markdown.append(doc["id"], {
"markdown": "## Results\n\nAll 96 wells within expected range.",
})
outline = dalea.documents.markdown.outline(doc["id"])
print(f'{outline["blockCount"]} blocks:\n{outline["markdown"]}')
The markdown sub-resource also supports insert, update_block,
replace_section, and parse — see the API reference for each.
Inventory
Read the item-type catalogue, find a container by barcode, and log consumption from an item:
item_types = dalea.inventory.item_types.list()["itemTypes"]
container = dalea.inventory.containers.get_by_barcode("FRZ-A-001")
dalea.inventory.consumption.consume("item-uuid", {
"quantity": 2.5,
"reason": "Assay run 42",
})
Placements handle physical movement — check_in, batch_check_out, move,
and place_object under dalea.inventory.placements.
Search
Unified search across documents, files, data objects, and result schemas:
page = dalea.search.search({"query": "aspirin", "types": "document,data_object", "limit": 25})
# Scope data-object results to one or more tables (e.g. the same table
# name across several environments) — `table_ids` joins into the wire param:
scoped = dalea.search.search({"query": "batch-42", "table_ids": [table_a["id"], table_b["id"]]})
# Auto-paged: walk every result across pages.
for hit in dalea.search.search_all({"query": "aspirin"}):
print(hit["type"], hit["name"])
Storage
Upload a file and hand out a short-lived presigned download URL:
# Multipart from raw bytes:
with open("plate-map.csv", "rb") as fh:
uploaded = dalea.storage.upload.file(fh, filename="plate-map.csv")
# Or JSON with base64 content:
dalea.storage.upload.base64({
"filename": "report.pdf",
"mimeType": "application/pdf",
"data": base64_payload,
})
url = dalea.storage.files.get_url(uploaded["file"]["id"])["url"]
Presigned URLs are short-lived — fetch a fresh one per download rather than storing it.
Pagination
List methods return one page. The data domain's iterate_* companions walk
every item across pages — a sync generator on DaleaClient, an async
generator on AsyncDaleaClient:
# sync — on a DaleaClient
for obj in dalea.data.tables.iterate_objects(table_id, {"limit": 200}):
... # every object, across all pages
# async — the same method on an AsyncDaleaClient
async for obj in dalea.data.tables.iterate_objects(table_id, {"limit": 200}):
...
Also available: dalea.data.saved_queries.iterate(). Cursor-paged lists
(documents, files) return a nextCursor you pass back as cursor.
Errors
Non-2xx responses map to a typed error hierarchy. Throttling (429) and transient
503s are retried automatically with exponential backoff + jitter, honoring
Retry-After — writes included (a throttle is rejected before the handler runs,
so replaying is safe). A DaleaRateLimitError surfaces only once retries are
exhausted and carries retry_after_seconds. Tune with max_retries (default 2),
retry_writes (default True — opt out for strict callers), retry_base_delay,
and max_retry_delay.
from dalea import DaleaNotFoundError, DaleaRateLimitError
try:
dalea.data.tables.get(table_id)
except DaleaNotFoundError:
... # 404
except DaleaRateLimitError as err:
print(err.retry_after_seconds) # 429
The public SDK never deletes. Lifecycle is reversible — archive /
restore across the data domain, and documents/inventory/storage
follow the same no-hard-delete posture.
Resources
client.data.*—environments(incl. data import/export, staging views),tables,columns,objects,query,saved_queries,results,schema,naming_schemes,import_mappings.client.documents.*—list/create/get/update, plusmarkdown.*for the block round-trip.client.inventory.*—item_types,lots,containers,placements,consumption,activity.client.search.*—search, plus the auto-pagedsearch_all.client.storage.*—files,upload,usage.
The client covers exactly the public operations in the contract — the API reference is the authoritative catalogue.