Metadata-Version: 2.5
Name: telmai
Version: 0.3.4
Summary: Python client for the Telmai data quality platform: provision connections, assets and monitors, run scans, and gate pipelines on the results
Project-URL: Homepage, https://docs.telm.ai/telmai
Author-email: Josh Finlayson <joshua.finlayson@telm.ai>
License-Expression: Apache-2.0
License-File: LICENSE
License-File: NOTICE
Keywords: circuit breaker,data observability,data quality,databricks,provisioning,telmai
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Database
Classifier: Topic :: Software Development :: Quality Assurance
Classifier: Typing :: Typed
Requires-Python: >=3.9
Requires-Dist: requests>=2.32
Provides-Extra: dev
Requires-Dist: build; extra == 'dev'
Requires-Dist: mypy; extra == 'dev'
Requires-Dist: pytest-cov; extra == 'dev'
Requires-Dist: pytest>=7.0; extra == 'dev'
Requires-Dist: pyyaml; extra == 'dev'
Requires-Dist: ruff; extra == 'dev'
Requires-Dist: twine; extra == 'dev'
Description-Content-Type: text/markdown

# 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](#the-namespaces) — and the gate is a feature built on top of them,
not the whole product.

```python
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](#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:

```bash
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:

```bash
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`](.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:

```bash
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:

```bash
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.

```python
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`](#tmbinning) and the
[binning example](#set-up-data-binning-so-quarantine-can-route-rows).

```python
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.

```python
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`.

```python
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:

```python
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 `operationId`s, 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](docs/versioning.md) for the full stability policy.

## Usage examples

Each of these is a working shape you can copy. The
[`examples/`](examples/README.md) directory has the same things as runnable
scripts.

### Gate a pipeline

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

```python
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.

```python
# 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:

```python
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:

```python
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.

```python
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.

```python
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:

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

### Onboard a table end to end

Connection, asset, schema discovery, monitored columns.

```python
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.

```python
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:

```python
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

```python
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.

```python
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

```python
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.

```bash
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/`](examples/README.md), 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:

```bash
./.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/`](recipes/databricks/) (notebook, job wiring, and where
the halt comes from) and [`recipes/airflow/`](recipes/airflow/) (an operator).

## Developing

```bash
./.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:

```bash
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](TODO.md). If you are picking this
up, start with [HANDOFF.md](HANDOFF.md).

## More

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