Taps

​

Tap Types

• Structured/Semi-Structured — the script returns records that flow into a structured pipeline destination.
• Document Ingestion — the script 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 directly. The script still runs through the same test / auto-fix / post-run review / auto-optimize flow and can use any of the pre-installed packages or bring its own.

​

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. Datris caches installed packages in Docker volumes so subsequent runs reuse them instantly.

• “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

​

Auto-optimize after a successful test

• One pass per successful test. The optimizer is instructed to preserve fetch(), DATRIS_TAP_TEST_LIMIT handling, error handling, and Vault env-var reads — it only changes how the script fetches, not what
• If the optimized re-test runs ≥20% slower than the original, or fails, the platform auto-reverts to the original script with no user action
• On success, an amber banner shows "Auto-optimized: 12.4s → 1.1s (11.3×)" with a list of the changes the AI made (e.g. Parallelized ticker fetches with ThreadPoolExecutor(10), Removed per-item sleep) and a Revert to original link
• If you revert or edit the script, the auto-optimize pass won’t rerun until you click Generate again

​

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.

​

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