Skip to content

Python API Quick Start

Install the package, build a first pipeline, and tour the operations you'll reach for most.

Installation

pip install flowfile

Your first pipeline

This pipeline reads a CSV, derives a column, filters, and aggregates. It runs against committed sample data and is executed by the docs test suite:

import flowfile as ff

SALES = "https://raw.githubusercontent.com/edwardvaneechoud/flowfile/main/data/templates/supermarket_sales.csv"

revenue_by_line = (
    ff.read_csv(SALES)
    .with_columns((ff.col("unit_price") * ff.col("quantity")).alias("revenue"))
    .filter(ff.col("quantity") > 5)
    .group_by("product_line")
    .agg(ff.col("revenue").sum().alias("total_revenue"))
    .collect()
)

collect() runs the plan and returns a Polars DataFrame.

Key ideas

FlowFrame

Your data container — like a Polars LazyFrame, but every operation is also recorded as a graph node:

import flowfile as ff

df = ff.FlowFrame({"col1": [1, 2, 3]})  # from a dict
df = ff.read_csv("file.csv")            # from CSV
df = ff.read_parquet("file.parquet")    # from Parquet

Always lazy

Operations don't execute until you call .collect():

# These calls just build the plan
df = ff.read_csv("huge_file.csv")
df = df.filter(ff.col("status") == "active")
df = df.select(["id", "name", "amount"])

# Now it executes, reading only what's needed
result = df.collect()

Check df.schema to see column types without running anything.

Descriptions

Any operation accepts a description that shows up on the node in the visual editor:

df = (
    ff.read_csv("input.csv", description="Raw customer data")
    .filter(ff.col("active") == True, description="Keep active only")
    .unique(description="Remove duplicates")
)

Common operations

Filtering

# Polars expression predicate
df.filter(ff.col("age") > 21)

# Flowfile formula (renders as an editable Filter node)
df.filter(flowfile_formula="[age] > 21 and [status] = 'active'")

Adding columns

# Expression form
df.with_columns([
    (ff.col("price") * ff.col("quantity")).alias("total")
])

# Formula form
df.with_columns(
    flowfile_formulas=["[price] * [quantity]"],
    output_column_names=["total"],
)

Grouping and aggregation

df.group_by("category").agg([
    ff.col("sales").sum().alias("total_sales"),
    ff.col("sales").mean().alias("avg_sales"),
    ff.col("id").count().alias("count"),
])

Joining

customers = ff.read_csv("customers.csv")
orders = ff.read_csv("orders.csv")

result = customers.join(
    orders,
    left_on="customer_id",
    right_on="customer_id",
    how="left",
)

See the Joins reference for the full set of strategies.

Cloud storage

Store an S3 connection once, then read and write with it by name. This tested example round-trips a Parquet aggregate through S3:

income_by_city = (
    ff.read_csv("data/templates/supermarket_sales.csv")
    .group_by("city")
    .agg(ff.col("gross_income").sum().alias("total_income"))
)

income_by_city.write_parquet_to_cloud_storage(s3_path, connection_name="analytics-s3")

reloaded = ff.scan_parquet_from_cloud_storage(s3_path, connection_name="analytics-s3").collect()

Local S3-compatible stacks vs plain S3

That example sets endpoint_url, aws_allow_unsafe_html=True, and inline keys to reach a local MinIO stack. Against real AWS S3, the connection needs only its connection_name, storage_type, auth_method, aws_region, and credentials — no endpoint URL and no unsafe-HTML flag. Once the connection exists, reads and writes just reference it by name. See Cloud Connection Management.

Visual integration

Open in the editor

pipeline = ff.read_csv("data.csv").filter(ff.col("value") > 100)
ff.open_graph_in_editor(pipeline.flow_graph)

Start the web UI

ff.start_web_ui()  # opens a browser tab

Next steps