updo-sdk

Python client for the Updo360 public API (Qlaris ERP) — synchronous and asynchronous.

pip install updo-sdk

from updo import UpdoClient

with UpdoClient(token="sk_live_…") as client:
    print(client.me().tenant_slug)

    for product in client.entity("product").iterate(where={"status": "active"}):
        print(product["sku"], product["sale_price"])

---

What you need to know before you start

The public API lives under https://api.updo.pro/api/public/v1/ and accepts exactly one thing: a personal access token (sk_live_… in production, sk_test_… for the sandbox twin of that workspace).

• One token = one workspace. The binding is inside the token, so there is no tenant header to send (the SDK never sends one: an X-Tenant-ID can only cause a 401).
• The data model is defined per tenant. Entities and their fields are created in the Atelier (labelled Workshop in the English UI), so no static class can describe them. You discover them at runtime (client.entities) or generate typed models (updo codegen).
• A token can only call /api/public/v1/. The internal /api/v1/ routes track whatever the product's own UI needs and carry no stability contract; the server deliberately rejects them when the caller is a token.

Create a token in Updo (Settings → API tokens). The secret is shown only once.

---

Installation

pip install updo-sdk          # library
pip install "updo-sdk[cli]"   # + a TOML parser for `updo --profile` on 3.10

Python ≥ 3.10. Only dependency: httpx.

The updo command is installed either way — the [cli] extra only adds tomli, so that updo --profile can read a TOML config file on Python 3.10 where tomllib is not yet in the standard library. Importing the package never touches the CLI: import updo pulls in httpx and nothing else.

A runnable end-to-end example ships in the source distribution as examples/quickstart.py.

---

Configuration

client = UpdoClient(token="sk_live_…")  # api.updo.pro
client = UpdoClient(token="sk_test_…", base_url="http://localhost:8000")
client = UpdoClient()  # $UPDO_API_TOKEN / $UPDO_BASE_URL

Parameter	Default	Role
token	$UPDO_API_TOKEN	sk_… token
base_url	$UPDO_BASE_URL, otherwise https://api.updo.pro	API root
timeout	30.0	seconds (or an httpx.Timeout)
max_retries	3	number of retries
on_approval	"raise"	"return" to receive an object instead of an exception on a 202
site_id	$UPDO_SITE_ID	the site every request is scoped to — see Multi-site
http_client	None	your own httpx.Client (pool, proxy, mTLS…)

The base URL is normalised: https://api.updo.pro, https://updo.pro/, http://localhost:8000 and even …/api/v1 (pasted from the frontend config) all lead to …/api/public/v1.

---

Discovering the model

for entity in client.entities.list():
    print(entity.slug, entity.display("fr"))  # 'product', 'Produit'

schema = client.entities.schema("product")  # cached
for field in schema.fields:
    print(field.slug, field.field_type, "writable" if field.writable else "read-only")

schema.fields contains only the fields this token is allowed to read, and writable is read from the server's serialiser: a form built from it cannot offer a field the server would refuse.

Why the entity list can come back empty

Authorization is deny-by-default, and the index deliberately omits entities the token cannot read rather than listing them as forbidden — so a token with no grants sees a 200 and an empty workspace, not an error. That looks like a broken API and is really a missing grant. What each case looks like:

What you see	What it means	Fix
200, results: []	no grant on any entity	grant the token's role read on the entities, in the access manager
403 ACCESS_DENIED on schema/	same, for that entity	idem
index lists an entity, its schema/ returns 404	the entity belongs to a module whose subscription has lapsed — the index does not check subscriptions, the schema route does	reactivate the module, or pass --entity explicitly
403 with policy_id: token:scope_ceiling	the token's own scopes are narrower than its roles	reissue the token with scopes: [], or add data:<slug>:read
fewer fields than expected	per-field read permissions	expected: two tokens legitimately see two different shapes

updo codegen without --entity skips a lapsed-module entity and reports it on stderr rather than aborting the run. Three curl calls settle where you stand:

curl -H "Authorization: Bearer $UPDO_API_TOKEN" https://api.updo.pro/api/public/v1/me/

curl -H "Authorization: Bearer $UPDO_API_TOKEN" https://api.updo.pro/api/public/v1/data/

---

Multi-site

An Updo organisation can hold several sites, and a request resolves to exactly one — that is what decides which rows you see and, on a write, where the record lands. The server picks in this order: the X-Site-ID header, then the member's default site, then the organisation's main site.

client = UpdoClient(token="sk_live_…", site_id="0f4d2b5e-1c3a-4e6f-9a8b-7c6d5e4f3a2b")

# One credential, several sites, one connection pool:
for site in (montreal, quebec):
    for row in client.with_site(site).entity("stock").iterate():
        ...

site_id also reads $UPDO_SITE_ID. with_site(None) clears the site rather than falling back to that variable.

⚠️ A service-account token ignores the site completely. Measured against a real Qlaris: for a token whose user is None — the canonical machine mode — the server never populates its site context, so the site filter is not applied at all. Such a token reads every site's rows, as though all_sites=True were permanently on, and records it creates are stamped with no site, making them visible from every site. The header is honoured only by a token bound to a user, and then only if the token's own roles allow that site — the token's roles override the membership's for that check. Until the server changes, use a user-bound token for anything site-scoped.

A wrong site id does not fail. An unknown, malformed, foreign, inactive or forbidden site makes the server fall back to the main site and answer 200. Nothing in the status, the headers or the record envelope reveals it — the envelope never carries site_id. The SDK therefore validates the UUID before sending, which is the only moment the mistake can still be an error.

There is no way to list sites through the public API: the whole sites surface is internal, and an API token is refused there. Take the site UUIDs from the web interface.

all_sites=True on list() / iterate() wins over a configured site — the server short-circuits its site filter before it looks at the active site.

---

Reading records

A record is flat: {id, created_at, updated_at, <field>: value…}. Business fields are read like a dictionary; technical columns remain attributes (so an entity that happens to define a field named id does not shadow the record's own identifier).

products = client.entity("product")

page = products.list(page_size=50, ordering="-sale_price")
print(page.count, len(page.results))

product = products.get("0f4d2b5e-1c3a-4e6f-9a8b-7c6d5e4f3a2b")
product.id, product.created_at  # metadata
product["sku"], product.get("name")  # business fields

for p in products.iterate(where={"status": "active"}):  # walks every page, ordered
    ...

products.count(where={"status": "draft"})
products.first(where={"sku": "ABC-123"})

Filters

products.list(
    where={
        "sku": "ABC-123",  # equality
        "name__icontains": "croquette",  # substring, case-insensitive
        "sale_price__gte": 10,  # ≥
        "status__in": ["active", "draft"],  # list
        "barcode__isnull": True,  # empty field
    }
)

Suffixes: gte lte gt lt contains icontains in isnull. Conditions are combined with AND. Expression variant:

from updo import F

products.list(where=F(status="active") & F(sale_price__gte=10))

Two server pitfalls worth knowing. A filter on an unknown field returns 400. A filter on a field the token is not allowed to read is silently ignored — this is intentional (otherwise ?data__salaire__gte= would become an oracle for guessing a masked value), but it means a result set can be wider than expected with no error to signal it. When in doubt, check the field with client.entities.schema(...).

contains is not a substring test. On the server it is PostgreSQL's JSONB containment operator @>, so name__contains="crok" matches nothing at all against "croquette" — and returns a cheerful 200 while doing it. Use icontains for substrings. What contains IS good for is membership in a list-valued field: tags__contains="promo" finds records whose multi_select includes that choice.

An __in list cannot carry a value containing a comma: the server splits on commas and strips each part, so ["Dupont, Jean", "Tremblay"] would arrive as three wrong terms. The SDK refuses such a value rather than sending it wrong.

Dates and times need an explicit offset. A naive datetime is refused by the SDK rather than sent, because the server reads a naive value in the tenant's timezone — the same code would then mean different instants in different workspaces:

from datetime import datetime, timezone

products.list(where={"released_on__gte": datetime(2026, 1, 1, tzinfo=timezone.utc)})

Search, ordering, relations

products.list(search="chien")  # free text over text fields
products.list(ordering=["-sale_price", "name"])
products.list(expand="supplier")  # populates record.expanded

Automatic value typing

By default, values come back raw — the JSON exactly as sent. With the schema loaded, dates and decimals are promoted to Python objects:

products = client.entity("product", coerce=True)  # loads the schema once
p = products.get(record_id)
p["sale_price"]  # Decimal('12.50') — exact, not a float
p["released_on"]  # datetime.date(2026, 1, 15)

---

Writing

Values this SDK hands you can be sent straight back: Decimal and date are rendered the way the server reads them (money as a string, so a cent cannot be lost to a float). record.to_json_dict() gives a payload ready for update(), with the server-owned id / created_at / updated_at left out.

p = products.create({"sku": "ABC-123", "name": "Croquettes", "sale_price": 12.50})
products.update(p.id, {"sale_price": 13.90})  # PATCH — recommended
products.replace(p.id, {...})  # PUT
products.delete(p.id)

Fields of type password read back masked (••••••••). Sending a freshly-read record back through replace() would therefore overwrite the real secret. record.is_masked("field") detects it; prefer update().

Writes subject to approval

A write can be put on hold for approval (HTTP 202) instead of being applied. By default the SDK raises ApprovalRequired — returning an object with no id would let the calling code carry on as if the write had happened.

from updo import ApprovalRequired

try:
    products.create({...})
except ApprovalRequired as pending:
    print(pending.approval_request_id)

# or, to handle the case without an exception:
client = UpdoClient(token="sk_live_…", on_approval="return")
result = client.entity("invoice").create({...})  # Record or ApprovalPending

---

Analytics and export

agg = products.aggregate(group_by="status", metrics=["count", "sum:sale_price"])
for row in agg.results:
    print(row["group"], row["count"])

agg = products.aggregate(group_by="created_at", bucket="month", metrics=["count"])

pv = products.pivot(rows="status", cols="category", metric="count")
pv.cell("active", "chien")
pv.to_rows()  # ready for csv.DictWriter or pandas

od = products.query(
    select=["sku", "sale_price"], orderby=[("sale_price", "desc")], top=100, count=True
)
od.value, od.count  # $top is capped at 500 server-side

products.export("csv", dest="products.csv", where={"status": "active"})
products.export("xlsx", dest="products.xlsx")
data = products.export("csv")  # without dest: the bytes

The export honours filters, search and ordering, and ignores pagination.

The server truncates an export at 10 000 rows — no header, no marker, no error: a 40 000-record entity yields a perfectly well-formed CSV holding the first 10 000. The SDK therefore counts first and raises rather than hand you a file that is quietly a quarter of your data. Pass check_complete=False to skip the count, and use iterate() — which has no cap — for anything larger.

---

Payment links

Turn an invoice into a payer-facing URL, server to server. The payer needs no account.

link = client.payments.create_link(invoice_id, expires_days=30)
send_to_customer(link.url)

client.payments.revoke_links(invoice_id)  # returns how many were cancelled

⚠️ link.url is a bearer credential. Whoever holds it can open the payment form for that invoice. PaymentLink masks it in repr() for the same reason the client masks your API token — read .url explicitly when you mean to hand it to the payer, and treat it like a password everywhere else. If one leaks: revoke_links(), then create_link() again, which mints a genuinely new secret.

You cannot set a price. There is no amount, no currency and no return_url — the amount comes from the invoice and is recomputed on every payer visit, and the currency is the workspace's payment_currency setting. Those parameters are absent from the signature rather than accepted and ignored.

Retrying is safe. The server keeps at most one active link per invoice and returns the existing one instead of minting a second secret, so a duplicated request cannot produce two collectable URLs. A new secret appears only when the previous link expired, was revoked, or was settled — and when two mints for the same invoice race, since the server has no unique constraint there. Mint one invoice at a time rather than in parallel.

expires_days (1–365, server default 60) applies only to a fresh mint; on a reuse it is silently ignored. This endpoint cannot extend an existing link.

link.amount_due is a Decimal, but indicative: the server sends it as a JSON double and recomputes the real amount when the payer arrives. Do not reconcile against it.

The invoice must be payable — a draft, cancelled, voided or already-paid invoice, or one with a zero balance, raises ValidationError.

---

Record images

The one surface here that publishes to the open internet. Upload and publish are a single call: a 201 means the image is attached and its URL is live.

images = client.entity("product").images(product_id)

image = images.upload(
    "photos/moulin.png",
    category="photo",
    alt_text={"fr": "Le moulin en hiver", "en": "The mill in winter"},
)
print(image.public_url)  # paste straight into <img src="...">

for image in images.list():
    print(image.position, image.category, image.public_url)

images.delete(image.id)

upload() takes a path, raw bytes, or a handle opened in binary mode. The file is read into memory — the server caps uploads at 10 MiB — so that a throttled request can be replayed without uploading an empty body.

category is required, whatever the schema says

The server's OpenAPI marks it optional. It is not: publication checks the category against the entity's public_media whitelist, and that whitelist is built by a normaliser that strips blank entries — so "" can never be a member, and omitting the category is a guaranteed 409, not a default. The SDK requires it so the refusal happens at the call site instead of a round trip later.

The whitelist is not readable through the public API. The schema endpoint returns fields, and this declaration is a property of the entity. Today the only ways to learn which categories publish are to open the entity in the Workshop, or to try and catch PublicationRefused.

What the SDK refuses before sending

Each of these also comes back from the server — but only after the whole file has gone up, and only as a bare code that names the lock rather than the fix:

Refused locally	Because
an SVG, PDF, BMP, TIFF, ICO, HEIC or ZIP	the server publishes only PNG, JPEG, GIF and WebP, and it decodes the bytes rather than trusting the file name
over 10 MiB	the server's ceiling
position outside 0–32767	it lands in a smallint
alt text over 125 characters, or over 24 languages	the domain's bounds
a non-UUID record id	the server's 404 would not say which id was wrong

SVG is excluded deliberately, not by oversight: it is a document format that executes script when served inline, and a published blob is served with no origin of its own. Rasterise it first.

Publication refusals

from updo import PublicationRefused

try:
    images.upload(photo, category="contract")
except PublicationRefused as exc:
    if exc.is_configuration:
        alert_ops(exc.code)  # nothing the caller can resend will fix it
    else:
        convert_and_retry()

PublicationRefused subclasses ConflictError, so an existing except ConflictError keeps working. Its .code is one of five stable, machine-readable values the server documents for branching:

exc.code	What to do
public_media_not_declared	the entity publishes nothing at all; add a public_media declaration in the Workshop
public_media_category_not_publishable	name the category in that whitelist
public_media_not_an_image	send different bytes — the only refusal the caller can fix
public_media_unavailable	the deployment has no public container; an operator matter
public_media_sandbox_refused	a sandbox twin may not publish — its URLs would go dead on a real site

Every refusal is decided before anything is stored, so a 409 has written no blob and attached no row. Retrying after fixing the cause needs no cleanup.

Two warnings

Uploading is not idempotent. Unlike a payment link, nothing de-duplicates here: two calls with the same file attach two images. Only the transport's 429 replay is safe, because a throttle refuses before the server has done anything.

Un-publishing is not retroactive. delete() stops new public reads, but a CDN or a browser that already fetched the blob may keep serving it. Treat a published image as published for good — which is exactly why the server refuses to publish anything but decoded raster images.

image.public_url is not masked in repr(), unlike PaymentLink.url. It carries no authority beyond "show these bytes" and exists to be embedded publicly; hiding it would obstruct the only thing anyone does with it.

---

Webhooks

Subscribing

hook = client.webhooks.create(
    url="https://my-service.example.com/updo",  # https required
    event_pattern="invoice.*",  # or "invoice.paid", "*.created"
    slug="invoice-paid",
    secret="whsec_…",  # signs every delivery
)

client.webhooks.update(hook.id, is_active=False)
client.webhooks.delete(hook.id)

Patterns accept a wildcard per segment: invoice.*, *.created. Events are <entity>.created / .updated / .deleted, plus the platform events (document.signed, workflow.approval_requested…).

Verifying a received delivery

from updo.webhooks import parse_event, verify_signature


@app.post("/updo")
def receive(request):
    raw = request.body  # the BYTES, before any JSON parsing
    if not verify_signature(
        SECRET,
        raw,
        request.headers.get("X-Qlaris-Signature"),
        timestamp=request.headers.get("X-Qlaris-Timestamp"),
    ):
        return 401

    event = parse_event(raw, headers=request.headers)
    if already_processed(event.delivery_id):  # stable across retries
        return 200
    process(event.event, event.data)
    return 200

⚠️ Sign the raw bytes. Updo signs exactly what it puts on the wire (json.dumps(sort_keys=True, separators=(",",":"))). Re-parsing then re-serialising the JSON changes key order and whitespace: the signature will no longer match — and you will reject a payload that was perfectly genuine.

Reply 2xx quickly: a response ≥ 400 is retried up to 3 times with an increasing delay, and the codes 400/401/403/404/405/410/422 are treated as final (the delivery is abandoned).

Delivery log

for delivery in client.webhooks.iterate_deliveries(status="failed"):
    print(delivery.event_name, delivery.response_code, delivery.error)

---

Errors

from updo import (
    UpdoError,
    UpdoAPIError,
    AuthenticationError,
    PermissionDenied,
    NotFoundError,
    ValidationError,
    ConflictError,
    PublicationRefused,
    PlanLimitExceeded,
    RateLimitError,
    ServerError,
    ApprovalRequired,
)

try:
    products.create({"sku": ""})
except ValidationError as exc:
    print(exc.code)  # 'BUSINESS_RULE_VIOLATION'
    print(exc.field_errors)  # {'sku': ['This field is required.']}
except PermissionDenied as exc:
    print(exc.policy_id)  # e.g. 'token:scope_ceiling'

The server speaks two error dialects (the platform envelope {code, detail, field_errors} and DRF's raw form); the SDK normalises them, so exc.code / exc.detail / exc.field_errors are always readable.

Status	Exception	Typical case
401	AuthenticationError	unknown or expired token, IP refused, call outside /api/public/
402	PlanLimitExceeded	subscription quota reached
403	PermissionDenied	ABAC refusal, token scope ceiling, module disabled
404	NotFoundError	entity or record does not exist
400 / 422	ValidationError	business rule, invalid value, unknown parameter
409	ConflictError	protected deletion, separation-of-duties conflict
409	PublicationRefused	a public_media declaration forbids publishing an image — a ConflictError subclass carrying a stable .code
429	RateLimitError	throttling — exc.retry_after
5xx	ServerError	platform-side outage

Retries

The SDK retries automatically, with exponential backoff and jitter:

• 429 on every method (Retry-After honoured) — the request was rejected before any side effect, so replaying it is safe;
• 5xx and network drops only on idempotent methods (GET/PUT/DELETE). A 500 after a POST can mean the write succeeded and only the response was lost: replaying it would create a duplicate.

Four throttle buckets apply to a token at once: 1000/h per token, 10000/h per workspace, 3000/h per user, and a 120/min burst limit.

---

Asynchronous client

Same surface, same guarantees:

import asyncio
from updo import AsyncUpdoClient


async def main():
    async with AsyncUpdoClient(token="sk_live_…") as client:
        products = await client.entity("product")
        async for p in products.iterate(where={"status": "active"}):
            print(p["sku"])


asyncio.run(main())

Only difference: client.entity() is a coroutine (it may have to load the schema).

---

Typed models (codegen)

The OpenAPI document cannot describe a record's fields — they are defined per tenant. So they are generated from the real workspace:

updo codegen --entity product --entity invoice --out my_app/updo_models.py

from my_app.updo_models import Product

p = Product.from_record(client.entity("product").get(id))
p.sale_price  # Decimal | None, with IDE autocompletion
p.status  # Literal["active", "draft", "archived"] | None

new_product = Product(sku="ABC-123", name="Croquettes")
client.entity("product").create(new_product.to_payload())  # writable fields only

Regenerate after any change to the model in the Atelier. --no-timestamp makes the output stable byte for byte, useful if the file is version-controlled.

The codegen needs nothing but a token that can read the entity — no builder rights, no Atelier licence. What it cannot do for you is the bootstrap: minting the token and defining the entities are admin gestures on the internal surface, so they happen in the UI. See Why the entity list can come back empty.

What the generated types promise, and what they don't

Field type	Generated as	Note
select radio segmented chips	Literal[...]	narrowed to the declared choices
multi_select checkboxes	list[Literal[...]]
decimal currency	Decimal	exact, never a float
date datetime	date / datetime
relation member	str	the wire carries a string, so this is what you get
computed json	Any	the server does not publish a result type

Two caveats the public schema itself cannot resolve, because it does not publish the discriminant:

• Relation cardinality. A to-many relation carries a list of ids, but the schema publishes no cardinality, so every relation is annotated str and the field's docstring says so. Check the entity in the Atelier before writing to a relation.
• Translatable fields read back as a locale map ({"fr": ..., "en": ...}) rather than a string, and nothing in the schema marks them.

Use FieldSpec.choice_labels(lang) when you need the human labels behind a choice field rather than its raw keys.

---

Command line

The short version is below; docs/cli.md in the source distribution is the full guide, with output for every command and a troubleshooting section.

export UPDO_API_TOKEN=sk_live_…

updo whoami
updo entities
updo schema product
updo get product --where status=active --where sale_price__gte=10 --limit 20
updo get product --table --columns sku,name,sale_price
updo count product --where status=draft
updo create product --data '{"sku":"ABC-123","name":"Croquettes"}'
updo update product <id> --file patch.json
updo delete product <id>
updo export product --format xlsx --out products.xlsx
updo aggregate product --group-by status --metrics count,sum:sale_price
updo pivot product --rows status --cols category
updo query product --select sku,sale_price --orderby 'sale_price desc' --top 50
updo webhooks list
updo webhooks create --url https://my-service.example.com/updo --event 'invoice.*'
updo webhooks deliveries --status failed
updo codegen --out models.py
updo openapi --out openapi.json

Output is JSON when stdout is redirected and a readable table when it is a terminal; override with --json / --table. Exit codes: 0 success, 2 API error, 3 write pending approval.

Profiles in ~/.config/updo/config.toml (or %APPDATA%\updo\config.toml):

[default]
token = "sk_live_…"

[sandbox]
token = "sk_test_…"
base_url = "http://localhost:8000"

updo --profile sandbox entities

The token is never displayed again: every output passes it through a mask (sk_live_ab…yz).

---

Covered surface

Every documented public route — a contract test checks this against the OpenAPI document served by the platform, and fails the build on any operation the SDK does not reach.

Route	SDK method
GET /me/	client.me()
GET /data/	client.entities.list()
GET /data/{slug}/schema/	client.entities.schema(slug)
GET · POST /data/{slug}/	.list() .iterate() .create()
GET · PUT · PATCH · DELETE /data/{slug}/{id}/	.get() .replace() .update() .delete()
GET /data/{slug}/aggregate/	.aggregate()
GET /data/{slug}/pivot/	.pivot()
GET /data/{slug}/query/	.query()
GET · POST /webhooks/	client.webhooks.list() .create()
GET · PUT · PATCH · DELETE /webhooks/{id}/	.get() .replace() .update() .delete()
GET /webhooks/deliveries/[{id}/]	.deliveries() .delivery()
POST · DELETE /payment-links/	client.payments.create_link() .revoke_links()
GET · POST /data/{slug}/{id}/images/	.images(id).list() .upload()
DELETE /data/{slug}/{id}/images/{image_id}/	.images(id).delete()
GET /schema/	client.openapi()

For any route that is not modelled, the escape hatch keeps authentication and retries:

client.request("GET", "some/future/route", params={"x": 1})

---

Development

python -m venv .venv && .venv/Scripts/pip install -e ".[dev,cli]"
pytest                       # offline, simulated transport
ruff check . && ruff format --check .
mypy

Integration tests against a real instance (optional):

UPDO_BASE_URL=http://localhost:8000 UPDO_API_TOKEN=sk_test_… pytest -m integration

Use a mode: "test" token: it is bound to a disposable twin of the workspace, so write tests do not touch production.

Refresh the contract test fixture after an API change:

updo openapi --out tests/data/openapi.json

---

License

MIT. The full text ships as LICENSE in the source distribution.