Python SDK

pip install feldera

uv pip install feldera

from feldera import FelderaClient
 
# Connect to local Feldera instance (default)
client = FelderaClient()
 
# Connect to a remote instance with authentication
client = FelderaClient(
    url="https://try.feldera.com",
    api_key="apikey:your_key_here"
)
 
# Connect with custom TLS certificate
client = FelderaClient(
    url="https://localhost:8080",
    api_key="apikey:your_key_here",
    requests_verify="/path/to/tls.crt"
)

export FELDERA_HOST="https://localhost:8080"
export FELDERA_API_KEY="apikey:your_key_here"
export FELDERA_HTTPS_TLS_CERT="/path/to/tls.crt"

from feldera import PipelineBuilder
 
sql = """
CREATE TABLE events (
    id BIGINT NOT NULL PRIMARY KEY,
    message VARCHAR
) WITH ('materialized' = 'true');
 
CREATE MATERIALIZED VIEW event_count AS
SELECT COUNT(*) AS total FROM events;
"""
 
# Create a new pipeline
pipeline = PipelineBuilder(client, "my_pipeline", sql).create_or_replace()

from feldera import Pipeline
 
# Get an existing pipeline
pipeline = Pipeline.get("my_pipeline", client)
 
# Or get all pipelines
pipelines = Pipeline.all(client)

from feldera import FelderaClient, PipelineBuilder
 
client = FelderaClient()
 
sql = """
CREATE TABLE orders (
    order_id BIGINT NOT NULL PRIMARY KEY,
    customer_id BIGINT,
    amount DECIMAL(10, 2)
) WITH ('materialized' = 'true');
 
CREATE MATERIALIZED VIEW total_revenue AS
SELECT SUM(amount) AS revenue FROM orders;
"""
 
# Create and start the pipeline
pipeline = PipelineBuilder(client, "revenue_pipeline", sql).create_or_replace()
pipeline.start()
 
# Wait for the pipeline to be running
pipeline.wait_for_status(PipelineStatus.RUNNING, timeout=60)

# Start the pipeline
pipeline.start()
 
# Start in paused state
pipeline.start_paused()
 
# Pause a running pipeline
pipeline.pause()
 
# Resume a paused pipeline
pipeline.resume()
 
# Stop the pipeline (with checkpoint)
pipeline.stop(force=False)
 
# Stop immediately (no checkpoint)
pipeline.stop(force=True)
 
# Restart the pipeline
pipeline.restart()

from feldera.enums import PipelineStatus
 
# Get current status
status = pipeline.status()
 
if status == PipelineStatus.RUNNING:
    print("Pipeline is running")
elif status == PipelineStatus.PAUSED:
    print("Pipeline is paused")
 
# Wait for a specific status
pipeline.wait_for_status(PipelineStatus.RUNNING, timeout=60)

# Push a single record
pipeline.input_json("orders", {
    "order_id": 1,
    "customer_id": 100,
    "amount": 99.99
})
 
# Push multiple records
pipeline.input_json("orders", [
    {"order_id": 2, "customer_id": 101, "amount": 149.99},
    {"order_id": 3, "customer_id": 102, "amount": 79.99}
])
 
# Push with insert/delete format
pipeline.input_json("orders", {
    "insert": {"order_id": 4, "customer_id": 103, "amount": 199.99}
}, update_format="insert_delete")

import pandas as pd
 
# Create a DataFrame
df = pd.DataFrame({
    "order_id": [5, 6, 7],
    "customer_id": [104, 105, 106],
    "amount": [299.99, 399.99, 499.99]
})
 
# Push to the pipeline
pipeline.input_pandas("orders", df)

# Execute a query and iterate over results
for row in pipeline.query("SELECT * FROM total_revenue"):
    print(f"Total revenue: {row['revenue']}")
 
# Query returns a generator, so you can process large results efficiently
results = list(pipeline.query("SELECT * FROM orders WHERE amount > 100"))

# Save query results to a Parquet file
pipeline.query_parquet("SELECT * FROM orders", "orders.parquet")

# Get results as formatted text
for line in pipeline.query_tabular("SELECT * FROM total_revenue"):
    print(line)

# Execute INSERT or DELETE statements
pipeline.execute("INSERT INTO orders VALUES (8, 107, 599.99)")
pipeline.execute("DELETE FROM orders WHERE order_id = 1")

# Listen to changes in a view
handler = pipeline.listen("total_revenue")
 
# Process changes as they arrive
for chunk in handler:
    print(f"Received {len(chunk)} changes")
    for change in chunk:
        print(change)
 
# Stop listening
handler.stop()

import pandas as pd
 
def process_chunk(chunk: pd.DataFrame, seq_no: int):
    print(f"Processing chunk {seq_no} with {len(chunk)} rows")
    print(chunk)
 
# Process each chunk with a callback
pipeline.foreach_chunk("total_revenue", process_chunk)

# Pause an input connector
pipeline.pause_connector("orders", "kafka_source")
 
# Resume an input connector
pipeline.resume_connector("orders", "kafka_source")
 
# Get connector statistics
stats = pipeline.input_connector_stats("orders", "kafka_source")
print(f"Records processed: {stats.total_records}")

stats = pipeline.output_connector_stats("total_revenue", "kafka_sink")
print(f"Records transmitted: {stats.transmitted_records}")

stats = pipeline.stats()
 
# Global metrics
print(f"Total input records: {stats.global_metrics.total_input_records}")
print(f"Total processed records: {stats.global_metrics.total_processed_records}")
print(f"Pipeline complete: {stats.global_metrics.pipeline_complete}")
 
# Check if pipeline is idle
if pipeline.is_complete():
    print("Pipeline has processed all inputs")

# Wait for all inputs to be processed
pipeline.wait_for_completion(timeout_s=300)
 
# Wait for pipeline to become idle
pipeline.wait_for_idle(idle_interval_s=5.0, timeout_s=60.0)

# Stream pipeline logs
for log_line in pipeline.logs():
    print(log_line)

# Start a transaction
txn_id = pipeline.start_transaction()
 
# Push data within the transaction
pipeline.input_json("orders", {"order_id": 9, "customer_id": 108, "amount": 699.99})
pipeline.input_json("orders", {"order_id": 10, "customer_id": 109, "amount": 799.99})
 
# Commit the transaction
pipeline.commit_transaction(txn_id, wait=True)
 
# Check transaction status
status = pipeline.transaction_status()

# Create a checkpoint
seq = pipeline.checkpoint(wait=True, timeout_s=60)
 
# Check checkpoint status
status = pipeline.checkpoint_status(seq)
 
# Sync checkpoint to object store
uuid = pipeline.sync_checkpoint(wait=True, timeout_s=120)
 
# Get checkpoint metadata
checkpoints = pipeline.checkpoints()
for cp in checkpoints:
    print(f"Checkpoint {cp.sequence_number} at {cp.timestamp}")

# Modify pipeline SQL
new_sql = """
CREATE TABLE orders (
    order_id BIGINT NOT NULL PRIMARY KEY,
    customer_id BIGINT,
    amount DECIMAL(10, 2),
    order_date TIMESTAMP
) WITH ('materialized' = 'true');
"""
 
pipeline.modify(sql=new_sql)
 
# Modify runtime configuration
from feldera.runtime_config import RuntimeConfig
 
runtime_config = pipeline.runtime_config()
runtime_config.workers = 8
pipeline.set_runtime_config(runtime_config)

# Stop and delete the pipeline
pipeline.stop(force=True)
pipeline.delete()
 
# Or stop, clear storage, and delete
pipeline.stop(force=True)
pipeline.clear_storage(wait=True)
pipeline.delete()

from feldera.rest.errors import FelderaAPIError
 
try:
    pipeline = Pipeline.get("nonexistent_pipeline", client)
except FelderaAPIError as e:
    print(f"Error: {e.message}")
    print(f"Status code: {e.status_code}")

from feldera import FelderaClient, PipelineBuilder
from feldera.enums import PipelineStatus
import time
 
# Initialize client
client = FelderaClient()
 
# Define SQL program
sql = """
CREATE TABLE sensor_data (
    sensor_id BIGINT NOT NULL,
    temperature DECIMAL(5, 2),
    timestamp TIMESTAMP
) WITH ('materialized' = 'true');
 
CREATE MATERIALIZED VIEW avg_temperature AS
SELECT sensor_id, AVG(temperature) AS avg_temp
FROM sensor_data
GROUP BY sensor_id;
"""
 
# Create and start pipeline
pipeline = PipelineBuilder(client, "sensor_pipeline", sql).create_or_replace()
pipeline.start()
pipeline.wait_for_status(PipelineStatus.RUNNING)
 
# Push some data
pipeline.input_json("sensor_data", [
    {"sensor_id": 1, "temperature": 22.5, "timestamp": "2024-01-01T10:00:00"},
    {"sensor_id": 2, "temperature": 23.1, "timestamp": "2024-01-01T10:00:00"},
    {"sensor_id": 1, "temperature": 22.8, "timestamp": "2024-01-01T10:05:00"}
])
 
# Wait for processing
time.sleep(2)
 
# Query results
for row in pipeline.query("SELECT * FROM avg_temperature ORDER BY sensor_id"):
    print(f"Sensor {row['sensor_id']}: {row['avg_temp']}°C")
 
# Clean up
pipeline.stop(force=True)
pipeline.delete()