Taps

​

Tap Types

• I Have My Own Code — paste your own Python fetch() script directly. The script runs through the same test / auto-fix / post-run review flow and can use any of the pre-installed packages or bring its own. (AI-generated scripts are still available through the Assistant and MCP create_tap with an instruction. Generation reports the CodeGen provider and model that wrote the script (the primary model when no CodeGen model is set): POST /api/v1/tap/generate and its /status poll return provider and model, and create_tap returns them as generatedWith, so the Assistant can tell you which model generated the tap.)
• HTTP Endpoint — you host the tap as an HTTP service, in any language; no code runs on the platform. See HTTP taps below.

​

HTTP taps: bring your own service, in any language

​

Creating a Tap

​

Step 1: Describe

• Structured/Semi-Structured (default) — AI generates a script that returns records flowing into a structured pipeline destination.
• Document Ingestion — AI generates a script that returns {uri, filename, content} dicts for each file; the pipeline handles text extraction, chunking, and embedding.
• I Have My Own Code — skip the AI and paste your own Python fetch() script. The AI brainstorm, Instruction textarea, and Generate Script button are all hidden; you see a full-width code textarea and a Use My Code button that uploads the pasted script to MinIO. Once uploaded, Next is enabled and the rest of the wizard (Edit & Test, Schedule, Review & Save) behaves identically to an AI-authored tap. Editing the textarea after upload invalidates the upload — you’ll see the button flip to Re-upload My Code and Next will be disabled until you re-click it.

1. Ask one focused clarifying question at a time
2. Suggest specific data sources (e.g., yfinance for stocks, Open-Meteo for weather, Alpha Vantage for fundamentals)
3. Recognize when you reference a Datris table (it knows the platform’s metadata and query endpoints)
4. Auto-update the Instruction box below on every turn as the conversation progresses
5. Suggest the environment variable names you’ll need (e.g., ALPHA_VANTAGE_API_KEY) when an external API requires authentication

• requests, beautifulsoup4, pandas, lxml, feedparser
• boto3 (AWS S3), google-cloud-storage (GCS), azure-storage-blob (Azure Blob)
• openpyxl (Excel), pyyaml (YAML), python-dateutil, pytz
• Additional packages can be specified and are installed at runtime via pip into a throwaway per-run environment. In in-process mode the pip download cache lives in a Docker volume so reinstalls skip the download; in the isolated runner (compose default) packages are fetched fresh on every run.

• “Fetch current weather data for New York from the Open-Meteo API”
• “Retrieve daily stock prices for all S&P 500 companies from yfinance”
• “Get the latest news headlines from the BBC RSS feed”
• “Query the consumer_discretionary_earnings table on Datris to get tickers, then fetch historical earnings from Alpha Vantage”

​

Step 2: Edit & Test

• Modify the Python code directly
• Add or remove pip packages
• Edit the instruction and click Regenerate to produce a new script
• Copy the script to your clipboard with the Copy button next to the editor
• Click Test Script to execute and preview results. Preview renders as JSON (up to 100 records) in a scrollable pane
• If the script fails, an AI Diagnosis explains what went wrong and is auto-applied — the platform fixes the script and retests, up to 3 attempts. If all retries fail, the diagnosis stays visible for you to review and apply manually
• Click Stop Test to cancel an in-flight test (also halts the auto-fix chain)

​

Post-run script review

• Rate-limit / throttle / burst warnings — add time.sleep, lower max_workers, add adaptive backoff on HTTP 429
• Deprecation / migration hints — switch to the recommended replacement endpoint, method, or field
• Pagination / partial-response hints (“truncated”, “page 1 of 5”, “next_cursor=…”) — add pagination so the tap keeps fetching until the source says no more
• Schema-drift / auth warnings — update parsing for renamed fields; for auth errors, add a clear stderr note naming the suspect env var

​

Step 3: Schedule (Optional)

• Presets: Every Hour, Daily (Midnight), Weekdays (Midnight), Weekly (Monday)
• Custom: Describe your schedule in plain English (e.g., “Every weekday at 4pm ET”) and click Generate to create the CRON expression

​

Step 4: Review & Save

• Attach to Pipeline — pick an existing pipeline whose source columns match the tap’s test output exactly. If columns don’t match, you’ll see exactly which columns are missing or extra so you can fix the mismatch before attaching.
• Generate Pipeline — create a new pipeline pre-wired to the tap’s output. The new pipeline’s destination is Postgres (public.<tap_name>) for tabular data, or MongoDB for nested/document data. All columns are created as string for safe ingestion — tap output shape can vary across runs, and string columns ingest reliably regardless of what the script returns. You can promote individual columns to richer types in the pipeline editor once you’re confident about the data shape.

• Attach to Pipeline — the picker is filtered to only compatible pipelines: those whose source is unstructuredAttributes and whose destination is a vector store (Qdrant, pgvector, Weaviate, Milvus, Chroma). Column matching doesn’t apply — every document has the same {uri, filename, content} shape. The selection card shows the destination (e.g. pgvector → public.contracts).
• Generate Pipeline — creates an unstructured → vector-store pipeline. You pick:
  • Vector store — only stores whose secret is configured and the service is reachable are listed (server probes each on modal open via /api/v1/vector-stores/available). If exactly one is available it’s preselected; if none are, the modal blocks with a link to Configuration.
  • Destination name — collection name (Qdrant / Milvus / Chroma), class name (Weaviate), or table + schema (pgvector).
  • Chunking strategy — recursive (default), fixed, sentence, paragraph, or none. Plus chunk size (default 500 chars) and overlap (default 50). These live on the pipeline, not the tap.
  • Embedding is inherited from the environment’s default embedding secret — no picker this pass.

​

Step 5: Run the Tap (Optional)

• Run the Tap Now — runs the tap script immediately and pushes the records it returns into the linked pipeline. The pipeline applies any data quality and transformation rules and writes to its destination. After the run completes you’re returned to the Taps list.
• Done — skip the run and return to the Taps list. You can run the tap later from the Taps page or via the MCP run_tap tool.

​

Querying Datris Data

import os, requests
host = os.environ['DATRIS_PLATFORM_HOST']
port = os.environ['DATRIS_PLATFORM_PORT']
pg_db = os.environ['DATRIS_POSTGRES_DATABASE']
base_url = f'http://{host}:{port}/api/v1'

# List all tables
response = requests.get(f'{base_url}/metadata/postgres/tables',
                        params={'database': pg_db, 'schema': 'public'})
tables = response.json()

response = requests.post(f'{base_url}/query/postgres', json={
    'sql': 'SELECT * FROM public.my_table',
    'database': pg_db
})
results = response.json()['results']

mongo_db = os.environ['DATRIS_MONGODB_DATABASE']

# List collections
response = requests.get(f'{base_url}/metadata/mongodb/collections',
                        params={'database': mongo_db})
collections = response.json()

# Query a collection
response = requests.post(f'{base_url}/query/mongodb', json={
    'database': mongo_db,
    'collection': 'my_collection',
    'query': '{}'
})
results = response.json()['results']

​

Dependencies between taps

• Inferred. When a tap’s script is saved, the platform reads the table names its queries use (FROM / JOIN in SQL, collection in a Mongo query) and matches them against the destination tables of the pipelines in the same catalog. The taps that feed those pipelines are stored on the tap as inferredDependsOn. Common table expressions, tables of other catalogs and the tap itself are ignored. A tap saved before dependency tracking existed is analysed the first time its catalog’s enable order (or the tap itself) is read.
• Declared. Set dependsOn to a list of tap names in the same catalog to state the dependencies yourself (for example for a read inference cannot see, such as a warehouse or object-store read). A declared list replaces the inferred one; an empty list [] declares that the tap has none. Blank names are refused and repeated names are stored once. Every name must be a tap of the same catalog, not the tap itself, and the list may not close a cycle: a save that breaks either rule is refused (400 naming the tap, or 409 with the cycle). Leaving dependsOn out of an update keeps the stored list.
• Order only. Dependencies are never enforced: saving, testing, running and scheduling a tap behave exactly as without them.

​

Credentials / Secrets

• Select existing: Choose from previously created tap secrets
• Create new: Define key-value pairs inline (e.g., API_KEY=your-key)
• Edit existing: Modify an existing secret’s fields
• Use suggested keys: When the brainstorm AI suggests environment variables, click + Create tap secret with these keys to jump straight into the create form with the keys pre-filled

​

Taps can only use tap secrets

Tap 'my-tap' uses secret 'ai-primary', which is a platform secret. Taps may only use tap secrets (Configuration → Secrets → Tap). Create a tap secret with the fields this tap needs and select it, or set DATRIS_TAP_SECRET_SCOPE=any to allow platform secrets.

• Create a tap secret under Configuration → Secrets → Tap with the fields the script reads, and select it on the tap.
• Tag a hand-made tap secret. A secret you created for a tap outside the wizard, without _type=tap, counts as a platform secret. Add the field _type with the value tap to it in the Secrets tab. Tagging a secret as a tap secret makes its values readable by any tap, so only an admin should do it, and only for credentials a tap is meant to use.

​

Running Taps

​

Test Run (from UI)

• Shows the tap instruction
• Send to pipeline checkbox (only if a target pipeline is configured)
• Displays script output, results table, and errors
• Run Again button for re-execution

​

CRON Schedule

​

Sending Data to Pipelines

1. The tap executes the script
2. Records are converted to the pipeline’s expected format (CSV or JSON)
3. Data is fed through the pipeline’s processing chain (data quality, transformation, destinations)

​

Watching a Run

• persisted: true/false with persistedReason when false (no_target_pipeline, test_mode, no_records, run_error, debounced) so callers know whether the data actually landed. debounced means another run for this same tap was triggered within the last 5 seconds — that earlier run is still executing; do not retry, just track it via get_tap_logs / get_pipeline_status.
• publisherToken — a single ID covering every ingestion job this run submitted. Document taps fan out to many jobs per run; one publisherToken covers them all.
• pipelineTokens — the per-job IDs (length 1 for structured, N for document).
• recordCount — how many records the script produced.

​

Per-run Parameters

import os, json

start_date = os.environ.get("DATRIS_TAP_PARAM_start_date", "2026-01-01")   # cron-safe default
end_date   = os.environ.get("DATRIS_TAP_PARAM_end_date")
tickers    = json.loads(os.environ.get("DATRIS_TAP_PARAM_tickers", "[]"))  # JSON-encoded list

• Key constraints: must match [A-Za-z_][A-Za-z0-9_]* so the keys map cleanly onto env var names. Anything else is rejected with an actionable error.
• Value handling: strings pass through; numbers/booleans are stringified; nested objects/arrays are JSON-encoded so the script can json.loads() them back.
• Scheduled runs supply no params — cron-triggered runs have an empty params bag. Scripts MUST apply sensible defaults when the env var is absent.

​

Incremental Sync (State)

1. Before each run, the last committed state is injected into the script as the DATRIS_TAP_STATE environment variable (absent on the very first run).
2. The script saves new state by assigning a dict to the module-global DATRIS_STATE inside fetch().
3. The platform commits the new state only after a successful run. A failed run leaves the old bookmark in place, so the automatic retry re-fetches the same window — no data holes. With keyFields configured on the destination, overlapping re-fetches upsert safely.

import os, json

def fetch():
    global DATRIS_STATE
    state = json.loads(os.environ.get("DATRIS_TAP_STATE") or "{}")
    since = state.get("updated_after", "1970-01-01T00:00:00Z")

    rows = get_rows_updated_after(since)   # fetch only what changed

    if rows:
        newest = max(r["updated_at"] for r in rows)
        DATRIS_STATE = {"updated_after": max(since, newest)}
    return rows

• State is a cursor, not a data store — a timestamp, an ID, a token, or a hash. The platform rejects state over 64 KB.
• Cursors must be monotonic: advance with max(previous, newest_seen) so re-processing old data never moves the bookmark backwards.
• Backfills don’t move the bookmark: when an explicit window is passed via run_tap params, the script should leave DATRIS_STATE unset for that run.
• Never put credentials in state — it is stored and displayed unmasked.
• Test runs read state but never commit it.

​

Run Status Outcomes

​

Output Size Guard

​

Data Types

​

Pipelines and Taps

​

Fetch-only taps

​

Run History

• Status (success/failure)
• Timestamp and duration (formatted using the configured dateFormat and dateTimezone — see Configuration Reference)
• Record count
• Whether data was sent to a pipeline
• Expandable script output logs and error messages

​

MCP Tools

create_tap_secret(name="stripe-key", fields={"apiKey": "sk_..."})
create_tap(name="stripe-charges", script="...", secret_name="stripe-key")

​

CLI Commands

# List all taps
datris taps

# Create a tap with AI-generated script
datris tap create "Fetch weather data from Open-Meteo API" --pipeline weather --cron "0 0 * * * ?"

# Create a tap with your own script
datris tap create --script ./my_script.py --name my-tap --pipeline weather

# Create an HTTP tap — a service you host speaks the tap contract (see HTTP Tap Contract doc)
datris tap create --name my-tap --kind http --endpoint-url https://taps.example.com/my-tap --pipeline weather

# Create a tap config only (add script later)
datris tap create --name my-tap --pipeline weather

# Show tap details including script
datris tap show my-tap

# Test-run a tap (no data push)
datris tap test my-tap

# Run a tap manually
datris tap run my-tap

# Read the rows a fetch-only (Live Read) tap run produced
datris pipeline result <pipeline-token> --out rows.jsonl

# View run history
datris tap logs my-tap

# Update a tap's schedule or config
datris tap update my-tap --cron "0 0 * * * ?" --enabled

# Disable a tap
datris tap update my-tap --disabled

# Delete a tap
datris tap delete my-tap

​

Configuration

​

Document Taps

​

Return shape

• Return one entry per source file. Never chunk, split, or segment. Never decode the bytes.
• Don’t generate embeddings, pick embedding models, choose a vector store, or create tables — those are pipeline concerns.
• Don’t invent metadata fields that describe pipeline behavior (chunk_size, embedding_model, target_table); those belong on the pipeline.

​

The Document Ledger

1. Loads uri → contentHash for every existing ledger entry owned by this tap.
2. For each document returned by fetch(), computes (or reads) its content hash.
3. Skips the document if the ledger already has an entry with the same hash (just refreshes lastSeenAt for operator visibility).
4. Otherwise, stages the bytes to MinIO, writes a ledger entry (status: staged), submits to the pipeline, and marks the entry processed (or failed on error).

• UI — the document-ledger button (page icon) next to a document tap in the Taps list opens a modal listing every entry. Actions: Clear All (force full re-scan on next run), or delete a single entry (force that one file to re-process).
• REST — GET /api/v1/tap/ledger?name={tap}, DELETE /api/v1/tap/ledger?name={tap}&uri={uri} for a single entry, DELETE /api/v1/tap/ledger?name={tap} to clear all.
• MCP — get_tap_ledger with optional clear_uri / clear_all.

• Deleting a tap clears its ledger entries and deletes all staged MinIO objects.
• Deleting a pipeline’s data (from the Pipelines page → Delete Data Only, or DELETE /api/v1/pipeline?pipeline=X&deleteData=true) clears the ledger for every document tap pointed at that pipeline — so the next run re-ingests every file into the now-empty destination. Without this, taps would skip docs they already “processed” and the pipeline would stay empty.

​

Pipeline compatibility

1. On save (POST /tap) — if tapType == "document" and targetPipeline is set, it checks the pipeline’s shape. Rejects with HTTP 400 if the source isn’t unstructuredAttributes or the destination isn’t a vector store.
2. On feed (TapRunner.feedDocumentPipeline) — re-checks before submitting bytes, so a pipeline that was reshaped after the tap was saved fails the run with a clear error instead of silently corrupting a destination.

​

Concurrency

• pgvector — transactional advisory lock plus CREATE TABLE IF NOT EXISTS; serializes concurrent sessions at the schema-qualified table name.
• Qdrant / Milvus / Weaviate — try-create; on error, re-check existence and swallow if a concurrent runner won the race.
• Chroma — sends get_or_create: true plus a re-GET fallback for older servers.

​

Script Requirements

• Take no arguments
• Return a list of dictionaries (records), a list of lists (rows, header first), or a string (JSON/XML/text) — or yield the records / rows one at a time (any generator or iterator works)
• For large sources, yield. The platform streams whatever fetch() produces to disk one record at a time, so a run is never limited by memory — but only if the script does not build the whole result first. A fetch() that appends millions of rows to a list (or loads a whole file into pandas and calls to_dict("records")) is killed for memory, and the run fails with an error naming memory and yield. Read the source in chunks / pages / batches (pd.read_csv(..., chunksize=50000), pyarrow.parquet.ParquetFile(...).iter_batches(), one API page at a time) and yield each row before fetching the next chunk. Never cap the rows or split the source into several runs for size — the one exception is when the estimate below exceeds a budget and the user has chosen to chunk the range.
• Yield batches for a large columnar or tabular source. For a large columnar or tabular source, yield each batch (a pyarrow RecordBatch, a pandas DataFrame, or a list of dicts) instead of each row — the platform serialises a batch natively, several times faster — and read only the columns the pipeline needs (columns=[...]). Yielding one row at a time remains correct for small or API-paged sources. Each batch is written as its rows, so nothing downstream changes: the record count is rows, the columns are the union of the batches’ column names, and the staged records are identical to yielding the batch’s own rows (df.to_dict("records") / batch.to_pylist()) one at a time. Floats inside a batch are written with up to 15 significant digits rather than Python’s shortest round-trip form — the type and the destination schema are unchanged. Two other cosmetic differences in a batch: -0.0 lands as 0.0, and a / inside a string is escaped as \/; both parse to the same value.
• Progress is reported for you. A fetch() that yields records makes the platform print [wrapper] streamed N records (M s) to the run log every 100,000 records or every 30 seconds, so a long or timed-out run shows how far it got. No script change is needed, and a fetch() that returns a list prints none.
• Stream the source, never download it. The isolated tap runner has no disk of its own: its only writable area is a 512 MB in-memory scratch space shared with the per-run venv and any pip-installed packages (see Tap execution isolation). A fetch() that saves a large file locally before reading it — urlretrieve, open(path, "wb"), requests with stream=True written to a file, a library that spools under TMPDIR — fails with No space left on device. Read remote files directly: pyarrow.parquet.ParquetFile(fsspec.open(url).open()) (or an io.BytesIO of a single small file), pd.read_csv(url, chunksize=...), requests.get(url, stream=True).iter_lines(), and yield records as they arrive.
• Check the time budget too. A test is killed after TAP_SCRIPT_TIMEOUT_SECONDS (default 300 s) and a real or scheduled run after TAP_RUN_TIMEOUT_SECONDS (default 3600 s); both ceilings in force are reported by get_version / GET /api/v1/version. A test is also capped at 20 records, so a passing test says nothing about how long a real run takes — only an estimate does. A timed-out run says which mode it ran in and which variable to raise.
• Check both budgets before you build. Estimate the run from two measured numbers, and show them before creating anything:
  • Disk — records are staged as JSON with their keys repeated on every row, so a staged record costs roughly 26 bytes × column count for numeric / timestamp / short-string data (long text or nested fields add their own length). rows × that is what the run stages, capped by PIPELINE_MAX_PAYLOAD_MB (default 4096 MB, 0 = unlimited). Read the value in force from get_version / GET /api/v1/version (pipelineMaxPayloadMB) — an install that raised the variable runs with a different ceiling, and the estimate is compared with that, not with the default.
  • Time — a fetch() that yields one record at a time moves about 15,000 rows per second; one that yields batches moves about 90,000 rows per second. Say which rate the estimate used, and prefer batches for any source over about a million rows. Compare the minutes with TAP_RUN_TIMEOUT_SECONDS (default 3600 s) — again, the value in force is tapRunTimeoutSeconds from get_version / GET /api/v1/version.
  • Show the estimate as a four-line table — rows, bytes/record, GB staged, minutes — so the row-count assumption can be corrected. For example, 20,000,000 rows × 25 columns → ~650 bytes/record → ~13 GB staged → ~4 minutes at the batch rate.
  • If either estimate is over its budget, choose before building: raise the variable (the staging volume needs matching free space; a longer timeout holds a runner slot that long), narrow the window, or chunk the range via run_tap params — several runs land in the same pipeline. Reading only the columns the destination needs cuts both numbers linearly. Finding out after several minutes of streaming is the expensive way.
• Let exceptions propagate (the wrapper captures the traceback for diagnosis); only catch what you can actually recover from
• Include timeouts on network requests (30 seconds recommended)
• Use os.environ.get() for any credentials
• Read DATRIS_POSTGRES_DATABASE, DATRIS_MONGODB_DATABASE, DATRIS_PLATFORM_HOST, and DATRIS_PLATFORM_PORT directly without fallback defaults — they are always injected by the platform
• Column names are auto-normalized for tabular results (list of dicts). The platform converts each key to lowercase snake_case using only [a-z0-9_] so that downstream pipeline registration succeeds and SQL queries don’t need quoting. Examples: EPS Estimate → eps_estimate, Surprise(%) → surprise_percent. You can return raw source column names — the platform will clean them — but for clarity in the test preview, prefer to emit clean keys directly. JSON/XML results destined for MongoDB are not normalized (they go through as raw blobs in the _json field). See Schema Definition → Column Naming Rules for the underlying validator rule.