Taps

​

Creating a Tap

​

Step 1: Describe

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
• 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
• Click Test Script to execute and preview results
• If errors occur, an AI Diagnosis explains what went wrong
• Click Apply Diagnosis to Script to auto-fix the issue

​

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.

​

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)

​

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

​

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

​

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.