telmai

Python SDK for gating data pipelines on Telmai data quality scans.

Pre-release. Not published to PyPI yet, so pip install telmai does not work. Install from source — see Getting set up. The first published version will be 0.2.0.

Why this exists

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.

from telmai import CircuitBreakerTripped, Severity, Telmai

tm = Telmai.from_env()  # TELMAI_HOST / TELMAI_TENANT / TELMAI_API_KEY
tm.circuit_breaker("a1b2c3d4e5f6")  # raises if the batch is not clean

Getting set up

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.

Three environment variables, and nothing else:

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.

Python 3.9 or newer. CI runs 3.9 through 3.13. The floor is 3.9 rather than something more modern because Databricks and Airflow runtimes lag, and those are the two places this code actually runs.

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.

Three ways to act on a scan

One scan underneath, three responses to its result.

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)
tm.circuit_breaker([ORDERS, CLAIMS], min_severity=Severity.MEDIUM)

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

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.

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

Everything else

Nine namespaces, mirroring what you are working on:

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

tm.assets.create(project_id, {...})
tm.assets.detect_columns(ORDERS)  # async; poll with tm.jobs.wait()
tm.assets.set_monitored_columns(ORDERS, {...})

tm.monitors.create(ORDERS, {...})
tm.monitors.set_enabled(ORDERS, monitor_id, False)
tm.monitors.export(ORDERS)  # monitors as code, with import_()

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

tm.alerts.for_scan(ORDERS, job_id)
tm.jobs.wait(ORDERS, job_id)
tm.incidents.list()
tm.dq_score.get(ORDERS)
tm.binning.set(ORDERS, {...})  # where quarantine routes bad rows

Below those, tm._api carries the generated operations — every customer-facing route in Telmai's OpenAPI spec, so a missing wrapper is never a dead end. It is not the whole platform: tools/generate.py filters out /internal and /admin routes plus the tag areas in its EXCLUDE_TAGS set, which serve the web app or are configured in the product, auth among them. The names there come from the platform's own operationIds, so they are faithful rather than pretty — creating a connection is tm._api.get_connection_4(). See docs/versioning.md for why it is underscored and what that means for upgrades.

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.

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

Pre-release. 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
• recipes/databricks/ — notebook, job wiring, and where the halt comes from
• recipes/airflow/ — operator
• CONTRIBUTING.md — the one rule, and what not to tidy