telmai

Python client for the Telmai data quality platform, covering both halves of the job. Configuration: connections, projects, assets and their attributes, monitors, scans and jobs, alerts, incidents, DQ score and data binning — enough to provision a tenant end to end and keep it in sync, from code. Gating: three ready-made ways to turn a scan result into a decision an Airflow or Databricks job can act on.

So if you are here to find out whether this can configure a tenant: yes. Ten namespaces cover the platform's configuration surface — see the namespaces — and the gate is a feature built on top of them, not the whole product.

from telmai import Telmai

tm = Telmai.from_env()  # TELMAI_HOST / TELMAI_TENANT / TELMAI_API_KEY

tm.connections.create({...})  # connect a warehouse
tm.assets.create(project_id, {...})  # onboard a table
tm.monitors.create(ORDERS, {...})  # say what good looks like
tm.scans.run(ORDERS, wait=True)  # and go and check

Published. telmai is on PyPI — pip install telmai works. Current version is 0.3.4. Developing the SDK itself, rather than just using it? See Getting set up for installing from source.

Why the gate is here too

Telmai finds quality problems. It cannot act on them, because the moment that matters is inside someone else's pipeline: the step between writing a batch and publishing it. No API call can fail a customer's Databricks task or skip their Airflow step. That has to be code running in their job.

Customers who want that today write their own HTTP client. Asking "did this batch pass?" is four calls, a state machine, and five failure modes that all have to resolve the same way. Getting it wrong is easy and the consequences are asymmetric, so this ships it once, correctly.

Getting set up

Just using the SDK in a pipeline? pip install telmai is all you need — into a virtual environment, because macOS system Python (and Debian's) refuses to install into itself and fails with externally-managed-environment, PEP 668:

python3 -m venv .venv
./.venv/bin/pip install telmai

The steps below are for developing the SDK itself, so they install from source in editable mode with the dev extras:

git clone https://github.com/Telmai/telmai-python.git && cd telmai-python
python3 -m venv .venv && ./.venv/bin/pip install -e ".[dev]"

Every command below uses ./.venv/bin/... explicitly, so nothing depends on whether the venv is activated.

Configuration

Three environment variables, and nothing else:

Variable	What it is
TELMAI_HOST	bare hostname, no scheme — your-tenant.telm.ai
TELMAI_TENANT	the tenant id, the opaque string in your Telmai URL
TELMAI_API_KEY	from your secret store, never from a file in a repo

.env.example documents these three plus TELMAI_LIVE, and is the fastest local setup. Nothing auto-loads it — the SDK reads the environment, not a file — so source it yourself:

cp .env.example .env        # then fill in TELMAI_API_KEY
set -a && source .env && set +a

.env is gitignored. .env.example deliberately leaves TELMAI_API_KEY commented out rather than empty, so sourcing a half-filled file cannot export an empty string over a key that was already working in your shell.

Constructing a client

Call	What it does
Telmai.from_env()	Read the three variables above. Names the missing one rather than failing generically.
Telmai(host=..., tenant=..., api_key=...)	Explicit construction, for a job that gets its key from a secret manager rather than the environment.
Telmai(..., poll_interval_s=10.0)	Maximum delay between job polls. The first poll is about a second in and the delay ramps to this cap.
Telmai(..., transport=...)	Inject a transport. Used by the test suite; you should not normally need it.
tm.close()	Release pooled HTTP connections.
with Telmai.from_env() as tm:	Same, as a context manager.

close() is optional in a short script — the process exiting closes everything. It matters in a long-lived host that builds a client per unit of work, an Airflow worker being the usual case, where sockets otherwise sit until garbage collection reaches the connection pools.

Assets are addressed by id

Always ids, never table names. Two tables in one project can both be called orders; only the id is unique, and it survives a rename in Telmai while your pipeline code stays put. Passing a name where an id belongs is rejected before any scan is triggered.

Look the id up once, out of band, and put it in your job config:

python -m telmai resolve gold.orders    # prints the id, then exits

There is a tm.resolve(name) for scripts that must do it at runtime, but a pipeline that runs every day should not be paying for a lookup — or risking an ambiguous one — on every run. resolve() raises TelmaiAssetError if the name is unknown or if more than one asset shares it, rather than picking whichever came back first.

Three ways to act on a scan

One scan underneath, three responses to its result. All three take the same keyword arguments:

Argument	Default	What it does
asset_ids	—	One asset id for one result, a list for a list of results.
min_severity	Severity.HIGH	Alerts at or above this rank block. HIGH, MEDIUM, LOW — there is no CRITICAL, see below.
tags	None	Monitor tags that block regardless of severity, e.g. {"block-writes"}.
max_concurrency	20	How many assets are scanned at once. Every concurrent scan is warehouse load.
timeout_s	3600	Deadline for the whole batch. A scan that exceeds it becomes a blocked verdict, not an exception.

Circuit Breaker. Stop the pipeline. Wire it between the write step and the promote step, so a failure skips everything downstream.

tm.circuit_breaker(ORDERS)                                     # -> ScanVerdict
tm.circuit_breaker([ORDERS, CLAIMS], min_severity=Severity.MEDIUM)  # -> list

Quarantine. Keep running, route the bad records aside. Needs Data Binning configured on the asset — see tm.binning and the binning example.

for r in tm.quarantine([ORDERS, CLAIMS]):
    if not r.clean:
        if r.fully_isolated:
            move_flagged_rows(r.location.incorrect_data_path)
        else:
            hold_entire_batch(r.label)  # binning covers only some monitors

Live Pass Through. Get told, without the gate deciding for you. For pipelines that must run regardless of what the scan finds.

tm.live_pass_through(ORDERS, on_result=lambda v: notify(v) if not v.passed else None)

It waits for the scan to finish, same as the other two, then calls back with the verdict — it does not run alongside your pipeline. What it gives you over circuit_breaker is that it never raises for a quality result, so nothing downstream is skipped. A callback that raises is logged and swallowed, because a broken notification must never turn "never blocks" into "sometimes blocks".

Fail closed, by contract

Every path that cannot confirm a clean result blocks: a scan that fails or times out, a severity we cannot read, an alert type meaning the scan never read the table, an asset with no monitors. A gate that reports success on bad data is worse than no gate, because the pipeline then certifies the batch.

Four consequences worth knowing before reading the code:

• A typo raises, it does not become a verdict. An unresolvable or ambiguous asset id fails the whole batch before anything is scanned. A typo should not look like a data problem.
• CircuitBreakerTripped is not a TelmaiError. A blocked batch is a correct decision, not an API failure, so except TelmaiError: retry cannot silently retry past it.
• A severity we cannot read blocks. If Telmai adds a tier above HIGH tomorrow, an SDK that predates it treats that alert as unrankable and stops the pipeline, rather than reading an unfamiliar value as harmless.
• "Blocked" and "bad data" are different things. A scan that could not read the table blocks too, and says so separately. passed decides whether the pipeline continues; blocked_by_quality decides what you tell someone. The CLI splits them as exit 2 versus exit 1.

try:
    tm.circuit_breaker([ORDERS, CLAIMS])
except CircuitBreakerTripped as e:
    for v in e.failed:  # every failure, not just the first
        log.error("%s: %s", v.label, [a.policy_name for a in v.blocking_alerts])
    raise

The namespaces

Ten of them, mirroring what you are working on. This is the configuration surface, and it is most of the SDK:

tm.projects.list()  # project ids, the first argument to most writes

tm.connections.create({...})  # connect a warehouse
tm.connections.test(conn_id, {...})  # check it can still reach the source

### Client

| Method | What it does |
|---|---|
| `tm.circuit_breaker(ids, **kw)` | Scan, and raise `CircuitBreakerTripped` if anything is not clean. Returns `ScanVerdict` (or a list). |
| `tm.quarantine(ids, **kw)` | Scan, never raise, report where flagged rows landed. Returns `QuarantineResult` (or a list). |
| `tm.live_pass_through(ids, on_result=fn, **kw)` | Scan, never raise, never block. Returns `ScanVerdict` (or a list). |
| `tm.resolve(name)` | Asset id for a table name. Raises if unknown or ambiguous. |
| `tm.close()` | Release pooled HTTP connections. |

run = tm.scans.run(ORDERS, wait=True)  # run.job_id, run.status
tm.scans.cancel(ORDERS, run.job_id)  # abandoning a scan does not stop it

Connect Telmai to a warehouse. Every method takes an optional `project_id=` to
use the project-scoped route instead of the tenant-wide one.

| Method | What it does |
|---|---|
| `create(body)` | Add a connection. `body` is a `ConnectionRequest` — see the trap below. |
| `get(connection_id)` | Read one connection. |
| `list()` | Connections you can edit. |
| `list_all()` | Also those you can see but not edit. |
| `update(connection_id, body)` | Replace a connection's configuration. |
| `delete(connection_id)` | Remove a connection. |
| `move(connection_id, body)` | Move it to a different project. |
| `test(connection_id, body)` | Can a *saved* connection still reach its source? First thing to reach for when onboarding fails. |
| `test_config(body)` | Check a payload *before* saving it. Credentials and reachability only — **it cannot tell you whether the connection can read a table**. |
| `browse(body)` | List what a connection can see, for picking tables to onboard. |
| `assets(connection_id)` | Every asset reading through this connection. |
| `set_credentials(connection_id, body)` | Rotate the stored credentials. |
| `delete_credentials(connection_id)` | Remove them. |

Credentials are a separate call from `update()` on purpose: editing a
connection's name or project should not mean re-sending its secrets, and a
caller cannot then send credentials by accident while editing something else.

**Three things about a DELTALAKE body, each of which has cost someone hours.**

```python
tm.connections.create({
    "name": "warehouse",
    "type": "DELTALAKE",
    "payload": {"host": "...", "httppath": "...", "use_catalog": False},
    "credential": {"type": "TOKEN", "token": "<pat>"},   # top level, not in payload
})

1. Credentials go in a top-level credential object. A token nested inside payload is accepted and silently ignored.
2. credential.type is "TOKEN", not "SIMPLE_TOKEN" — despite the platform's own class being named SimpleTokenCredentials.
3. use_catalog should be False unless you specifically need Unity Catalog. It is a boolean, not a catalog name. Setting it True against a workspace where Unity Catalog is not set up for the token produces a connection that creates cleanly, passes test_config(), onboards assets — and then fails every table read with a 90-second cluster timeout whose message names no cause.

That third one is why test_config() is not the last word. It opens a session; it does not run a query, and it returns a byte-identical success either way. Onboard one trivial asset and scan it before onboarding the rest — a SELECT 1 query asset is enough. One working baseline tells you far more than a pile of failing ones.

tm.projects

How a tenant is partitioned. Assets, connections, dashboards and permissions are all project scoped, and a project id is the first argument to most writes.

Method	What it does
list()	Every project on the tenant, each with its assets under sources.
get(project_id)	One project.
create(body)	Add a project. name at minimum.
update(project_id, body)	Replace its fields.
delete(project_id)	Remove it. The platform decides what happens to assets still inside.

id comes back as an integer, and every route that takes a project id takes it as a path segment. Stringify it rather than relying on the URL builder.

tm.assets

A table Telmai knows about. create, update, delete and get_in_project take project_id because their underlying routes are project scoped; the rest are addressed by asset id alone.

Method	What it does
list()	Every asset on the tenant. Always a real list, even on an empty tenant.
list_detailed(**kw)	The v2 list, one page, carrying last_scan_at, last_scan_status, monitors_count and incidents_count.
iter_detailed(page_size=100, **kw)	Every asset, following pagination to the end. Use this when "all assets" has to mean all of them.
get(asset_id)	One asset, from just an id copied out of a Telmai URL.
get_in_project(project_id, asset_id)	One asset, the lighter request, if you know the project.
list_by_project(project_id, **kw)	Assets in one project.
list_by_connection(connection_id)	Assets behind one connection.
create(project_id, body)	Add a table. body is an AssetRequest.
create_many(body)	Add many in one request. Use this over looping create() for bulk onboarding.
update(project_id, asset_id, body)	Full replacement — there is no partial-update route for assets.
delete(project_id, asset_id)	Remove an asset.
detect_columns(asset_id, monitor_all=False)	Start schema analysis. Asynchronous — poll with tm.jobs.wait(), do not sleep.
columns(asset_id)	The attributes Telmai discovered. Each has the id that set_monitored_columns needs.
set_monitored_columns(asset_id, body)	Choose which columns are monitored, in one bulk request, by column id.
set_column_description(asset_id, column_name, description)	Describe one column, by name, without touching anything else.
set_column_monitored(asset_id, column_name, monitored)	Monitor one column on or off, without clearing its description.
update_column(asset_id, attribute_id, body)	Full replacement. Prefer the two setters above.
set_parents(project_id, asset_id, parents)	Declare lineage. Takes a plain list of parent asset ids.

list_detailed() applies a default limit and silently returns a prefix past it, which is why iter_detailed() exists. Schema analysis is one-shot per asset: calling detect_columns() again on an asset that already has attributes configured returns an error from the platform.

tm.monitors

Two kinds, which the API keeps separate and so does this. Custom monitors get full CRUD and are addressed by id. Prebuilt monitors already exist on every asset, are addressed by name, and support read and update only.

Method	What it does
list(asset_id)	Custom monitors on an asset.
get(asset_id, monitor_id)	One monitor.
create(asset_id, body)	Add a monitor. body is a CreateMonitorRequestDTO.
update(asset_id, monitor_id, body)	Full replacement, not a patch.
delete(asset_id, monitor_id)	Remove a monitor.
set_enabled(asset_id, monitor_id, enabled)	Turn one on or off, leaving everything else alone.
set_tags(asset_id, monitor_id, tags)	Replace a monitor's tags.
add_tag(asset_id, monitor_id, tag)	Add one tag, keeping what is there.
list_prebuilt(asset_id)	Prebuilt monitors on an asset.
get_prebuilt(asset_id, monitor_name)	One, by name.
update_prebuilt(asset_id, monitor_name, body)	Replace it.
set_prebuilt_enabled(asset_id, monitor_name, enabled)	Turn one on or off.
set_prebuilt_attributes(asset_id, monitor_name, columns)	Scope one to a subset of columns. Check is_attributes_supported first — only some accept a scope.
export(asset_id)	Every monitor on an asset, as a portable definition.
import_(asset_id, body)	Apply an exported monitor set. Trailing underscore because import is a keyword.

Enable and disable are not endpoints — enabled is a field on the update request, so set_enabled() is a read-modify-write rather than a route of its own. That matters: the update route replaces, so a hand-built body silently drops whatever you forgot, typically a threshold or a notification block. The set_* and add_tag helpers project the read onto the update contract and raise rather than send an update that would clear a required field.

import_() replaces rather than merges — monitors absent from the payload are deleted. Export first and diff before running it against production.

tm.scans

Starting scans, and configuring what happens when they end.

Method	What it does
run(asset_id, wait=False, ...)	Scan an asset using its stored connection config. Returns job_id, or (job_id, status) with wait=True.
run_group(source_group_id, **options)	Scan every asset in a source group, in one request.
run_batch(asset_id, **options)	Scan with batch options: delta_only, from_time, to_time, limit, sample_fraction, train_model, id_attributes.
replay(asset_id, body)	Re-process a batch Telmai has already seen.
history(asset_id, all_history=False)	Past scans. Same as tm.jobs.list().
cancel(asset_id, job_id)	Stop a running scan. Same as tm.jobs.cancel().
get_callback(asset_id)	The webhook URL Telmai calls on job state changes.
set_callback(asset_id, url)	Set it, or pass None to stop. Read the warning below.

run() also takes timeout_s, poll_interval_s and on_poll when wait=True; they behave exactly as on tm.jobs.wait().

Callbacks are unauthenticated. The platform sends an unsigned POST with no retry and no delivery guarantee. There is no signature to verify, so a receiver cannot tell a real callback from a forged one, and anyone who learns the URL can fake a job completion. Treat a callback as a hint to go and check, never as evidence: on receipt, call tm.jobs.get(asset_id, job_id) and trust that instead.

tm.jobs

A scan is something you cause; a job is the record you read.

Method	What it does
list(asset_id, all_history=False)	Recent jobs, newest first. all_history=True reaches older runs.
get(asset_id, job_id)	One job, including its status.
wait(asset_id, job_id, ...)	Block until terminal. Returns "FINISHED", "FAILED" or "TIMEOUT".
cancel(asset_id, job_id)	Stop a running scan.

wait() does not raise on a failed job — a failure is an answer, and what it means is your decision. Only a transport problem raises. It takes timeout_s, poll_interval_s, and on_poll, a callback invoked with each observed status (pass print from a notebook, where log output is invisible).

Cancelling is worth knowing about: abandoning a scan does not stop it, so a pipeline that gives up on a slow scan keeps paying for compute until it finishes on its own.

tm.alerts

One monitor firing on one scan.

Method	What it does
for_scan(asset_id, job_id)	Every alert from one scan, as parsed Alert objects.
for_scan_raw(asset_id, job_id)	The same, untouched.
search(body)	Alerts across the tenant, filtered.
counts(sources=[...]) or counts(body={...})	Alert counts. One or the other is required by the platform.
counts_by_time(body=None)	Counts bucketed over time. A bare call is valid here.
tags()	Every monitor tag in use on the tenant.

Call tags() before relying on tag-based gating: a tag that does not exist blocks nothing, silently.

tm.incidents

Related alerts, grouped over time.

Method	What it does
list(**kw)	Open incidents.
get(incident_id)	One incident.
summary(**kw)	Aggregate view.
per_day(**kw)	Counts bucketed by day, for a trend.

tm.dq_score

The whole asset as one number.

Method	What it does
get(asset_id)	Current score for one asset.
all(**kw)	Scores for every asset, for a scorecard.
history(asset_id, body)	Score at specific past scans.
all_history(body=None)	The same, tenant-wide.
get_config(asset_id)	How the score is weighted.
set_config(asset_id, body)	Change the weighting.

history() is the awkward one. body is a DQScoreHistoryRequestDTO — {"scan_dates": ["2026-08-25T14:14:27.786Z", ...]}, ISO 8601 timestamps matching a job's start_time. It is not a day count or a range: the endpoint looks up specific scans by exact timestamp and returns 400 on anything else, including an empty list. The body is required, not optional.

tm.binning

Data Binning is what makes quarantine more than a blunt block: Telmai routes failing records to a cloud storage path, so a pipeline can exclude those and promote the rest.

Method	What it does
get(asset_id)	Raw binning config, as the platform stores it.
location(asset_id)	Parsed BinningLocation, or None if binning is not usable.
set(asset_id, body)	Replace the whole config.
patch(asset_id, body)	Change part of it, leaving the rest.
disable(asset_id)	Turn binning off without discarding the config.

Three things to know before writing config here.

Telmai's own documentation describes this as a UI-only feature. It is not — the GET, PUT and PATCH routes all exist, they are simply undocumented. That matters for anyone automating onboarding, because otherwise every new asset needs a human in the console before quarantine can do anything useful on it.

The payload carries storage credentials. Enabling binning means giving Telmai write access to a bucket, so set() is the one method in this namespace that handles secrets. Keep them out of source and out of logs, the same way you would a warehouse password.

Binning is scoped to specific monitors, by monitor_ids. An alert from a monitor outside that list was never written to the bad-data path, which is exactly why quarantine computes fully_isolated rather than assuming a configured bucket holds everything. Adding a monitor later does not extend binning to it.

location() returns None for three different reasons — binning genuinely disabled, no permission, or not found — and does not distinguish them. That fails safe, since quarantine treats an unknown location as "not covered", but it means a permissions problem looks like a config choice. Use get() if you need the error.

What is not here

Those namespaces are the surface. There is a generated layer underneath, one method per route in Telmai's OpenAPI spec, but it is internal and the client does not expose it: which endpoint the SDK calls for you is ours to change, and its method names come from the platform's operationIds, which are not names anyone should be asked to call (get_connection_4 is the POST that creates a connection). If something you need is not covered above, ask us for a wrapper — that is the supported path, and it lands on the stable surface. See docs/versioning.md for the full stability policy.

Usage examples

Each of these is a working shape you can copy. The examples/ directory has the same things as runnable scripts.

Gate a pipeline

The whole point. Put this between the write step and the promote step.

from telmai import CircuitBreakerTripped, Severity, Telmai

ORDERS = "a1b2c3d4e5f6"   # from `python -m telmai resolve gold.orders`

with Telmai.from_env() as tm:
    write_batch()
    try:
        tm.circuit_breaker(ORDERS, min_severity=Severity.MEDIUM)
    except CircuitBreakerTripped as tripped:
        for v in tripped.failed:
            print(f"{v.label}: {v.reason}")
        raise                       # fail the task; downstream is skipped
    promote_batch()

Set up Data Binning, so quarantine can route rows

Binning has to exist on the asset before quarantine() can tell you anything more useful than "blocked". Configure it once, per asset, as part of onboarding.

# Which monitors binning covers. An alert from a monitor outside this list is
# never written to the bad-data path, so quarantine will report the batch as
# not fully isolated and you should hold all of it.
monitor_ids = [int(m["id"]) for m in tm.monitors.list(ORDERS)]

tm.binning.set(ORDERS, {
    "type": "S3",                                            # S3, GCS, or AZURE
    "bucket": "acme-telmai",
    "correct_data_path": "s3://acme-telmai/clean/orders/",
    "incorrect_data_path": "s3://acme-telmai/quarantine/orders/",
    "monitor_ids": monitor_ids,
    "enabled": True,
    "output_format": "PARQUET",
    "credentials": {...},   # from your secret store; see the note below
})

The request is a DataBinningV2DTO:

Field		What it is
type	required	S3, GCS, or AZURE.
bucket	required	The bucket name.
correct_data_path	required	Where passing records go.
incorrect_data_path	required	Where failing records go. This is what quarantine reports back to you.
monitor_ids	required	Which monitors binning covers.
enabled	optional	Whether binning is on.
output_format	optional	CSV, JSON, PARQUET, AVRO, FLAT, DELTA, ICEBERG, XML, or PDF.
credentials	optional	What Telmai writes to the bucket with.

credentials is a polymorphic object whose per-provider shape the platform's OpenAPI spec does not declare, so the SDK passes it through untouched rather than guessing at field names. Copy the shape from the Data Binning page in Telmai's API reference for your storage type, and source the values from your secret store — never a literal in code, and never logged.

set() is a full replacement, not a merge — omit monitor_ids and you disable binning for the monitors you left out. To change one field, patch:

tm.binning.patch(ORDERS, {"monitor_ids": monitor_ids + [new_monitor_id]})
tm.binning.disable(ORDERS)                    # off, config retained

Read it back as a parsed object, which is what quarantine uses internally:

loc = tm.binning.location(ORDERS)
if loc is None:
    print("binning is not usable on this asset")
else:
    print(loc.storage_type, loc.bucket, loc.incorrect_data_path)
    print(loc.output_format, loc.monitor_ids)

Quarantine a batch and route around the bad rows

With binning configured, this is the payoff: keep the pipeline running, and promote everything except what Telmai isolated.

result = tm.quarantine(ORDERS)

if result.clean:
    promote_everything()

elif result.fully_isolated and result.location:
    # Every failing row came from a monitor binning covers, so the bad-data
    # path holds all of them and the rest of the batch is safe.
    loc = result.location
    exclude_rows_at(f"{loc.bucket}/{loc.incorrect_data_path}")
    promote_the_rest()

else:
    # Binning does not cover every failing monitor. Some bad rows are still in
    # the batch, so routing around the bin path would promote them.
    hold_entire_batch(result.label)
    for alert in result.blocking_alerts:
        print(f"  {alert.policy_name}: {alert.description}")

fully_isolated is the field that matters, and the reason this is not just "read the bin path".

Block on one specific check, without lowering the threshold

Tag-based gating. A monitor tagged block-writes blocks regardless of its severity, which lets one check be fatal while everything else stays advisory.

tm.monitors.set_tags(ORDERS, monitor_id, ["block-writes"])
# or, keeping whatever tags are already there:
tm.monitors.add_tag(ORDERS, monitor_id, "block-writes")

tm.circuit_breaker(ORDERS, min_severity=Severity.HIGH, tags={"block-writes"})

Check the tag actually exists first — one that does not blocks nothing, silently:

assert "block-writes" in tm.alerts.tags()

Onboard a table end to end

Connection, asset, schema discovery, monitored columns.

conn = tm.connections.create({...})              # a ConnectionRequest
tm.connections.test(conn["id"], {...})           # does it reach the source?

asset = tm.assets.create(PROJECT_ID, {
    "name": "gold.orders",
    "type": "DELTALAKE",
    "payload": {"catalog": "main", "schema": "gold", "table": "orders"},
    "connection_id": conn["id"],
})
asset_id = str(asset["id"])

# Asynchronous. Poll the job it starts rather than sleeping.
tm.assets.detect_columns(asset_id)
job_id = tm.jobs.list(asset_id)[0]["id"]         # newest first
tm.jobs.wait(asset_id, job_id)

columns = tm.assets.columns(asset_id)
tm.assets.set_monitored_columns(asset_id, {
    "attributes": [
        {"id": c["id"], "monitored": c["name"] in {"order_id", "amount"}}
        for c in columns
    ],
})

Pass monitor_all=True to detect_columns() to mark every discovered column monitored in the same call, and skip the last step.

Run a scan and watch it

When you want control over the scan rather than a gate decision.

job_id, status = tm.scans.run(ORDERS, wait=True, on_poll=print)
print(status)                                    # FINISHED / FAILED / TIMEOUT

# Or trigger and poll separately:
job_id = tm.scans.run(ORDERS)
status = tm.jobs.wait(ORDERS, job_id, timeout_s=1800, poll_interval_s=15)

if status != "FINISHED":
    tm.scans.cancel(ORDERS, job_id)   # abandoning it does not stop it

For a delta-only or sampled scan, use the batch route:

tm.scans.run_batch(ORDERS, delta_only=True, sample_fraction=0.1)
tm.scans.run_group(SOURCE_GROUP_ID, delta_only=True)   # a whole group at once

Read what a scan found

alerts = tm.alerts.for_scan(ORDERS, job_id)      # parsed Alert objects

for a in alerts:
    if a.is_process_failure:
        # The scan could not read the table. Not a data defect — telling an
        # operator "your data is bad" sends them hunting the wrong problem.
        print(f"SCAN FAILED: {a.description}")
    else:
        print(f"{a.policy_name} [{a.impact}] on {a.attribute}: {a.description}")

print(tm.dq_score.get(ORDERS))
print(tm.incidents.list())
print(tm.alerts.counts(sources=[ORDERS]))

alert.raw holds the unparsed payload. In production it contains violation_data, a sample of the offending rows — real customer data. It is kept out of repr() and equality so it cannot leak by accident, but do not log it, serialise it, or pickle a verdict into a store you would not put customer rows in.

Keep monitors in git

Export from one environment, review the diff, import to another.

import json, pathlib

definition = tm.monitors.export(STAGING_ORDERS)
pathlib.Path("monitors/orders.json").write_text(json.dumps(definition, indent=2))

# ... review the diff in a pull request, then:
body = json.loads(pathlib.Path("monitors/orders.json").read_text())
tm.monitors.import_(PROD_ORDERS, body)

import_() replaces rather than merges: monitors absent from the payload are deleted. The response reports created, updated and deleted counts.

React to results without blocking

def announce(verdict):
    if not verdict.passed:
        slack(f"{verdict.label} failed: {verdict.reason}")

tm.live_pass_through([ORDERS, CLAIMS], on_result=announce)
# Pipeline continues either way. Nothing was blocked.

Command line

For orchestrators that shell out. Anything that can run a process and read an exit status can use the gate, no Python required.

python -m telmai resolve gold.orders
python -m telmai gate --assets a1b2c3d4e5f6,f6e5d4c3b2a10 --min-severity HIGH

Flag	Default	What it does
--assets	required	Comma-separated asset ids.
--mode	block	block exits 2 on bad data; notify always exits 0.
--min-severity	HIGH	HIGH, MEDIUM, or LOW.
--tags	none	Comma-separated monitor tags that also block.
--timeout-s	3600	Deadline for the batch.
--max-concurrency	20	Assets scanned at once.
--json	off	Machine-readable output.
--verbose / -v	off	Debug logging.

Exit codes are the contract:

Code	Means
0	Clean. Every asset passed.
1	Error. Could not determine an answer — auth, config, bad asset id, or a scan that could not read the table.
2	Blocked. At least one asset failed a real quality check.

1 and 2 are deliberately different. A tool that cannot tell whether the data is good is not the same as one reporting that the data is bad, and an orchestrator often wants to alert differently on each. Both are non-zero, so the default behaviour is still to stop.

Types and exceptions

Everything below is importable from telmai directly.

ScanVerdict

Returned by circuit_breaker() and live_pass_through().

Attribute	What it is
asset_id	The id that was scanned.
asset_name	Display name, read off the alerts response. None when the scan produced no alerts to read it from.
label	asset_name if known, else asset_id. Use this for anything a human sees.
job_id	The scan job.
passed	Whether the pipeline may continue.
blocking_alerts	Alerts at or above the threshold, plus tagged and unreadable ones.
advisory_alerts	Everything else the scan found.
quality_blocks	Blocking alerts genuinely about the data.
process_failures	Blocking alerts meaning the scan itself could not run.
blocked_by_quality	True only when real data problems were found. Use this to decide what to tell someone.
scan_status	FINISHED, FAILED, or TIMEOUT.
reason	One line explaining the verdict.
detail	Why it failed, when the reason was not a quality alert.

QuarantineResult

Returned by quarantine().

Attribute	What it is
asset_id, asset_name, label, job_id, scan_status, detail	As above.
clean	Nothing blocked.
fully_isolated	Every blocking alert came from a monitor binning covers, so the bad-data path holds all the failing rows.
location	BinningLocation, or None.
blocking_alerts	What blocked.

Alert

Attribute	What it is
policy_name, policy_id	Which monitor fired.
impact	Severity, or None when absent or unreadable.
priority	AlertPriority, the fallback signal when impact is absent.
severity_rank	The rank used against the threshold. Unreadable ranks high, so it blocks.
alert_type	AlertType, or None.
is_process_failure	The scan could not read the table, rather than the data being bad.
is_unreadable_type	The wire carried a type this SDK does not recognise. Blocks.
tags	Monitor tags, for tag-based gating.
attribute	The column, when the alert is about one.
description	Flattened human-readable text.
raw	Unparsed payload. Contains customer rows in production — see above.

BinningLocation

storage_type (GCS, S3, or AZURE), bucket, incorrect_data_path, output_format, monitor_ids.

Enums

Enum	Values	Note
Severity	HIGH, MEDIUM, LOW	No CRITICAL. The platform never emits one, so a gate configured for it would set a threshold nothing could reach and pass every batch silently. Severity.parse("CRITICAL") raises instead.
AlertPriority	P1, P2, P3, NIL	NIL is a real value meaning "no priority set", not an absence. It carries no severity information, so it ranks unknown and blocks.
JobStatus	PREPARING, IN_PROGRESS, BATCH_WAIT, FINISHED, FAILED	Only the last two are terminal. Treating PREPARING or BATCH_WAIT as terminal makes a healthy in-flight scan look like a failure.
ScanStatus	FINISHED, FAILED, TIMEOUT	How a scan concluded, from the gate's point of view.
AlertType	POLICY, PROCESS_FAILURE, and the drift types	PROCESS_FAILURE is the one that matters: the platform raising an alert about itself, not about the data.

Exceptions

Exception	Raised when
TelmaiError	Base for transport, auth, config and API failures.
TelmaiConfigError	Client constructed without the configuration it needs.
TelmaiAuthError	401 or 403.
TelmaiAPIError	Non-2xx, or a body that could not be parsed. Carries status and body.
TelmaiTimeout	A single HTTP request exceeded its transport timeout. Not a scan exceeding timeout_s — that becomes a blocked verdict.
TelmaiAssetError	An asset name did not resolve, or resolved ambiguously. Raised for the whole batch before any scan is triggered.
CircuitBreakerTripped	At least one asset did not pass. Carries .verdicts (all of them) and .failed.

CircuitBreakerTripped deliberately does not inherit from TelmaiError, so except TelmaiError: retry() cannot silently retry past a real quality failure.

Runnable examples

Nine scripts under examples/, covering all twenty features, with a table of which ones cost compute. Start with the read-only one, which prints the asset ids the rest need:

./.venv/bin/python examples/01_find_your_assets.py

Anything that costs compute asks first. Anything that writes configuration is a dry run until you pass --commit.

For orchestrator wiring rather than scripts, see recipes/databricks/ (notebook, job wiring, and where the halt comes from) and recipes/airflow/ (an operator).

Developing

./.venv/bin/pytest -q                          # offline, no network, under a second
./.venv/bin/ruff check . && ./.venv/bin/ruff format --check .
./.venv/bin/mypy                               # strict
./.venv/bin/python tools/generate.py --check   # generated layer is current

mypy takes no argument on purpose: it is configured to check the package. mypy . would also walk the tests and the Airflow recipe, which imports a package the SDK does not depend on, and it reports several hundred errors that are not defects. Widening strict typing to the test suite is real work, not a config flag.

Never hand-edit telmai/_generated/. Fix tools/generate.py and regenerate. Any tool that generates code here runs the real linter and formatter on its own output rather than reimplementing their rules — that bit us twice.

The live suite

tests/live/ creates and deletes real objects on a real tenant. It is excluded from the default run and opts in by naming the tenant twice, which is what stops a stray environment variable pointing destructive tests somewhere unintended:

export TELMAI_LIVE=$TELMAI_TENANT
./.venv/bin/pytest tests/live -m "not costly"   # free
./.venv/bin/pytest tests/live -m costly         # triggers real scans, spends compute

Nothing pre-existing is ever deleted: cleanup only removes ids the run itself recorded. Read tests/live/conftest.py before changing it.

It earns its keep. Two bugs no offline test could have found: iter_detailed looped forever on a real first page, and alerts.counts() could never have worked because a required query parameter was generated as optional.

Where it stands

Published on PyPI as 0.3.4. All three gate modes are built and validated end to end against a live tenant, not just against fakes. Real alert payloads captured from that tenant are checked against our enums in tests/test_contract.py, which found defects no offline test could — including a severity tier the platform does not emit, and an alert type we were reporting as bad data when it actually meant the scan could not run.

The offline suite runs in under a second and is the specification, not a safety net: tests/test_fail_closed.py encodes decisions that look like over-engineering and are not. mypy --strict and ruff are clean, and the wheel ships py.typed, verified reaching a consumer's own type checker.

Known gaps and open questions are in TODO.md. If you are picking this up, start with HANDOFF.md.

More

• HANDOFF.md — where the project is and what to do next
• ARCHITECTURE.md — how it fits together, and the platform contract read off platform source because the public docs disagree in places
• docs/versioning.md — what you can rely on across versions, and the release process
• docs/phase1-features.md — the twenty features and the endpoint behind each
• docs/gitbook/ — the customer-facing pages, including a fuller SDK reference with per-route field lists
• CONTRIBUTING.md — the one rule, and what not to tidy