Quick start: Python SDK

This quickstart walks through creating a Tinybird project with the Python SDK. You define Data Sources and Endpoints in Python, use the CLI to build and deploy, and call your Endpoint from application code.

Looking for other ways to start? See Quick starts.

Before you begin

To get started, you need:

Python 3.14 is not supported yet for this workflow. The Python SDK relies on the Tinybird CLI under the hood, and current CLI support covers Python 3.11 to 3.13.

Initialize a Python project and add the SDK

Create a new directory and initialize a Python SDK project:

mkdir tinybird-python-quickstart
cd tinybird-python-quickstart

uv init --name tinybird-sdk-python-sample --python 3.13
uv add tinybird-sdk python-dotenv

git checkout -b tinybird_intro

uv run tinybird init --type python --folder src/tinybird/

This creates:

  • tinybird.config.json with an include path for Python resources
  • .env.local for your Tinybird credentials

Define resources in Python

Create src/tinybird/tinybird_resources.py:

from tinybird_sdk import define_datasource, define_endpoint, define_token, engine, node, t

app_token = define_token("app_token")

sample_events = define_datasource(
    "sample_events",
    {
        "description": "Events ingested by the Tinybird Python SDK sample",
        "schema": {
            "event_time": t.date_time64(3),
            "event_name": t.string().low_cardinality(),
            "user_id": t.string(),
            "amount": t.float64().default(0),
        },
        "engine": engine.merge_tree(
            {
                "sorting_key": ["event_time", "user_id", "event_name"],
            }
        ),
        "tokens": [
            {"token": app_token, "scope": "APPEND"},
        ],
    },
)

sample_event_totals = define_endpoint(
    "sample_event_totals",
    {
        "description": "Aggregated totals for SDK-ingested demo events",
        "nodes": [
            node(
                {
                    "name": "totals",
                    "sql": """
                        SELECT
                            event_name,
                            count() AS total_events,
                            uniqExact(user_id) AS unique_users,
                            round(sum(amount), 2) AS total_amount
                        FROM sample_events
                        GROUP BY event_name
                        ORDER BY total_events DESC
                    """,
                }
            )
        ],
        "output": {
            "event_name": t.string(),
            "total_events": t.uint64(),
            "unique_users": t.uint64(),
            "total_amount": t.float64(),
        },
        "tokens": [
            {"token": app_token, "scope": "READ"}
        ],
    },
)

Create src/tinybird/client.py:

import os

from tinybird_sdk import Tinybird
from src.tinybird.tinybird_resources import sample_event_totals, sample_events

tinybird = Tinybird(
    {
        "datasources": {"sample_events": sample_events},
        "pipes": {"sample_event_totals": sample_event_totals},
        "base_url": os.getenv("TINYBIRD_API_URL", "https://api.tinybird.co"),
        "token": os.getenv("TINYBIRD_TOKEN"),
    }
)

Build and deploy

Build your Python definitions into Tinybird resources against a branch:

uv run tinybird build

Deploy your resources to Tinybird Cloud:

uv run tinybird deploy

Ingest and query from your app

Use the runtime client in application code. Edit main.py file so that it looks like this:

from datetime import datetime, timezone

from dotenv import load_dotenv


def main():
    load_dotenv(".env.local")

    from src.tinybird.client import tinybird

    now = datetime.now(timezone.utc).isoformat(timespec="milliseconds")

    tinybird.sample_events.ingest(
        {
            "event_time": now,
            "event_name": "page_view",
            "user_id": "abc123",
            "amount": 1.0,
        }
    )

    result = tinybird.sample_event_totals.query({})

    for row in result["data"]:
        print(
            row["event_name"],
            row["total_events"],
            row["unique_users"],
            row["total_amount"],
        )


if __name__ == "__main__":
    main()

Get your host and app_token for Tinybird Cloud and set them in .env.local

uv run tinybird --cloud token copy app_token
echo "TINYBIRD_TOKEN=$(pbpaste)" >> .env.local
echo "TINYBIRD_API_URL=$(uv run tinybird --output=json info | jq -r '.cloud.api')" >> .env.local

Run main.py:

uv run python main.py

Check your Workspace in the Tinybird UI

uv run tinybird --cloud open

Iterate in a branch

Add a materialized view in another branch, test the app against that branch, and deploy to cloud after finishing.

git checkout -b materialization

Edit tinybird_resources.py to add a new Data Source, a new materialization, and update the Endpoint to read from the MV.

from tinybird_sdk import (
    define_datasource,
    define_endpoint,
    define_materialized_view,
    define_token,
    engine,
    node,
    t,
)

app_token = define_token("app_token")

sample_events = define_datasource(
    "sample_events",
    {
        "description": "Events ingested by the Tinybird Python SDK sample",
        "schema": {
            "event_time": t.date_time64(3),
            "event_name": t.string().low_cardinality(),
            "user_id": t.string(),
            "amount": t.float64().default(0),
        },
        "engine": engine.merge_tree(
            {
                "sorting_key": ["event_time", "user_id", "event_name"],
            }
        ),
        "tokens": [
            {"token": app_token, "scope": "APPEND"},
        ],
    },
)

sample_event_totals_mv_ds = define_datasource(
    "sample_event_totals_mv_ds",
    {
        "description": "Aggregated event totals populated by a materialized view",
        "schema": {
            "event_name": t.string().low_cardinality(),
            "total_events": t.aggregate_function("sum", t.uint64()),
            "unique_users": t.aggregate_function("uniq", t.string()),
            "total_amount": t.aggregate_function("sum", t.float64()),
        },
        "engine": engine.aggregating_merge_tree(
            {
                "sorting_key": ["event_name"],
            }
        ),
        "json_paths": False,
    },
)

sample_event_totals_mv = define_materialized_view(
    "sample_event_totals_mv",
    {
        "description": "Materialized rollup of sample events by event name",
        "datasource": sample_event_totals_mv_ds,
        "nodes": [
            node(
                {
                    "name": "aggregate_totals",
                    "sql": """
                        SELECT
                            event_name,
                            sumState(toUInt64(1)) AS total_events,
                            uniqState(user_id) AS unique_users,
                            sumState(amount) AS total_amount
                        FROM sample_events
                        GROUP BY event_name
                    """,
                }
            )
        ],
    },
)

sample_event_totals = define_endpoint(
    "sample_event_totals",
    {
        "description": "Aggregated totals for SDK-ingested demo events from MV-backed datasource",
        "nodes": [
            node(
                {
                    "name": "totals",
                    "sql": """
                        SELECT
                            event_name,
                            sumMerge(total_events) AS total_events,
                            uniqMerge(unique_users) AS unique_users,
                            round(sumMerge(total_amount), 2) AS total_amount
                        FROM sample_event_totals_mv_ds
                        GROUP BY event_name
                        ORDER BY total_events DESC
                    """,
                }
            )
        ],
        "output": {
            "event_name": t.string(),
            "total_events": t.uint64(),
            "unique_users": t.uint64(),
            "total_amount": t.float64(),
        },
        "tokens": [
            {"token": app_token, "scope": "READ"}
        ],
    },
)

Build against the branch:

uv run tinybird build

Adapt your .env.local to that branch:

sed -i '' '/^TINYBIRD_TOKEN=/d' .env.local
uv run tinybird --branch=mv token copy app_token
echo "TINYBIRD_TOKEN=$(pbpaste)" >> .env.local

Run main.py:

uv run python main.py

Once validated, you're good to deploy to Tinybird Cloud with uv run tinybird deploy

Next steps

Updated