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.)
• 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']

​

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

​

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

​

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

# 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) or a string (JSON/XML/text)
• Handle errors gracefully with try/except
• Include timeouts on network requests (30 seconds recommended)
• Return an empty list on failure rather than raising exceptions
• 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.