TraceBi

What is TraceBi?

TraceBi is a Python library for building BI and analytics workflows entirely in code — no GUI builders, no config files. Every data operation from raw ingest to formatted report is tracked in a full, immutable lineage chain. You always know exactly where a number came from and how it was calculated.

The same DataModel and lineage chain powers every output — whether that's an Excel file, an HTML report, a live Dash dashboard, or a scheduled pipeline writing back to a database.

Phases

Connectors

Connectors are the entry point to any data source. Every connector implements connect(), load(source), and write(df, table). They are registered in a DataModel by name.

from tracebi import SQLConnector, MemoryConnector, DataModel

db = SQLConnector("sales_db", url="sqlite:///data/tracebi.db")

model = DataModel("SalesModel")
model.add_connector(db)
model.add_table("orders",    connector="sales_db", source="orders")
model.add_table("customers", connector="sales_db", source="customers")
model.add_relationship("orders_customers", "orders", "customers",
                        left_key="customer_id", how="left")

DataSet & Lineage

DataSet is an immutable pandas DataFrame wrapper. Every fluent operation returns a new DataSet — the original is never mutated. Each operation appends a LineageNode to the chain.

orders = (
    model.load("orders")
    .filter("status == 'shipped'", description="Shipped orders only")
    .transform(
        lambda df: df.assign(margin=df["revenue"] - df["cost"]),
        description="margin = revenue - cost",
    )
    .sort("margin", ascending=False)
)

orders.print_lineage()
# ============================================================
#   Lineage for DataSet: 'orders'   Shape: 198 rows × 7 cols
# ============================================================
#   Step 1: [LOAD]       Loaded 'orders' from connector 'sales_db'
#   Step 2: [FILTER]     Shipped orders only  (250 → 198 rows)
#   Step 3: [TRANSFORM]  margin = revenue - cost
#   Step 4: [SORT]       Sorted by margin (desc)
# ============================================================

LineageNode fields

Lineage node colors

Used in LineageDiagram to color-code each step:

Report Engine (Phase 2)

Reports are built with a fluent builder and rendered to Excel or HTML. Every render saves a .manifest.json alongside the output file containing the full lineage for each section's dataset.

from tracebi.reports.report import Report, TextSection, TableSection, ChartSection
from tracebi.reports.excel_renderer import ExcelRenderer
from tracebi.reports.html_renderer import HTMLRenderer

report = (
    Report("Q2 Sales Report")
    .author("Data Team")
    .parameter("period", "Q2 2024")
    .add(TextSection(title="Summary", content="Summary", style="heading1"))
    .add(TextSection(content="Revenue up 12% vs Q1.", style="normal"))
    .add(ChartSection(title="Revenue Trend", dataset=trend_ds,
                      chart_type="line", x="month", y="revenue"))
    .add(TableSection(title="Orders", dataset=orders_ds,
                      columns=["region", "product", "revenue"],
                      totals=["revenue"],
                      number_formats={"revenue": "{:,.2f}"}))
)

ExcelRenderer().render(report, "output/q2_sales.xlsx")
HTMLRenderer().serve(report, port=8080)   # opens browser

Medallion Architecture (Phase 2.5)

Three distinct layer classes enforce the Bronze → Silver → Gold contract. Each layer can run standalone or as part of a PipelineRunner that coordinates scheduling and writes results back to a database.

StarSchema

Dimension references use dot notation: "dim_customer.region". Measures are a dict: {"revenue": "sum", "order_id": "count"}. Supported aggregations: sum, count, mean, min, max, nunique.

from tracebi.model.star_schema import StarSchema

schema = StarSchema("Sales", model=model)
schema.add_dimension("dim_customer", table_name="customers_silver",
                     key_col="customer_id", attributes=["region", "segment"])
schema.add_fact("fact_orders", table_name="orders_silver",
                measures=["revenue", "qty"],
                foreign_keys={"dim_customer": "customer_id"})

ds = schema.query(
    fact="fact_orders",
    measures={"revenue": "sum", "qty": "sum"},
    dimensions=["dim_customer.region"],
    filters={"status": "shipped"},
    aggregate=True,
)

Lineage Diagram

from tracebi.lineage.diagram import LineageDiagram

diag = LineageDiagram(ds)        # or LineageDiagram(report)
diag.show()                       # matplotlib window / Jupyter inline
diag.to_html("lineage.html")      # standalone HTML with embedded SVG
print(diag.to_mermaid())          # paste into GitHub markdown

Pipeline Runner (Phase 4)

PipelineRunner registers each layer with its own independent cron schedule and persists every run to a SQLite database. Layers are linked via depends_on, enabling refresh=True to walk the full chain.

from tracebi.pipeline.runner import PipelineRunner

runner = PipelineRunner(db_url="sqlite:///data/tracebi.db")

# Each layer registered with its own schedule
runner.register(orders_bronze,   name="orders_bronze",     schedule="0 * * * *")
runner.register(orders_silver,   name="orders_silver",     schedule="15 * * * *",
                depends_on="orders_bronze")
runner.register(revenue_by_region, name="revenue_by_region", schedule="30 6 * * *",
                depends_on="orders_silver")

# On-demand: single layer only
runner.run("orders_silver")

# On-demand: full chain refresh (bronze → silver → gold)
runner.run("revenue_by_region", refresh=True)

# Start all scheduled jobs (blocking — Ctrl+C to stop)
runner.start()

Cross-layer lineage

Each run record stores an upstream_run_id pointing to the most recent successful run of its upstream layer. This means you can always trace a gold table row back through silver → bronze to the exact source ingestion.

runner.lineage("revenue_by_region")
# ============================================================
#   Pipeline Lineage — 'revenue_by_region'
#   Chain: orders_bronze → orders_silver → revenue_by_region
# ============================================================
#   [orders_bronze]  schedule=0 * * * *  runs=3
#     ✓ run_id=  1  2026-05-12T02:27:22  rows 11 → 11  [success]
#
#   [orders_silver]  schedule=15 * * * * ← orders_bronze  runs=2
#     ✓ run_id=  3  2026-05-12T02:27:25  rows 11 → 10  [success]  upstream_run=1
#
#   [revenue_by_region]  schedule=30 6 * * * ← orders_silver  runs=1
#     ✓ run_id=  5  2026-05-12T02:27:27  rows 10 → 4   [success]  upstream_run=3

One-command database setup

This script: creates data/tracebi.db, seeds source tables (orders_raw, customers_raw), registers all six layers with the runner, and runs the initial Bronze load so the DB is ready immediately.

Dashboard (Phase 3)

A live Dash 2.x application driven by the same DataModel. Filters are associative — selecting a value in any FilterPanel automatically filters every panel whose dataset contains that column.

from tracebi.dashboard import Dashboard, DashboardServer
from tracebi.dashboard import FilterPanel, MetricPanel, ChartPanel, TablePanel

dashboard = (
    Dashboard("Q2 Sales Dashboard")
    .columns(2)
    .add_filter(FilterPanel("region-filter", label="Region",
                            column="region", table_name="orders"))
    .add_panel(MetricPanel("total-revenue", title="Total Revenue",
                           table_name="orders", column="revenue",
                           aggregation="sum", prefix="$"))
    .add_panel(ChartPanel("by-region", title="Revenue by Region",
                          table_name="orders", chart_type="bar",
                          x="region", y="revenue"))
    .add_panel(TablePanel("orders-table", title="Orders",
                          table_name="orders",
                          columns=["order_id", "region", "revenue"],
                          width=2))   # span both columns
)

DashboardServer(dashboard, model=model).run(port=8050)
# Open http://localhost:8050/

Web UI (Phase 5)

A FastAPI + Jinja2 browser interface over your TraceBi registry. Connectors, models, reports, pipelines, and live Dash dashboards — all accessible from one server on a single port.

# Install web dependencies
pip install -r web/requirements.txt

# Start the dev server (hot-reload on)
python web/run.py
# → http://localhost:8000

# Point at your own app module
TRACEBI_APP=mypackage.tracebi_config python web/run.py

Registry pattern

Everything flows through a single Registry singleton. Your app module (default: web/demo_app.py) registers objects at startup; every API route reads from the registry at request time.

from web.api.registry import registry

registry.add_connector(connector)
registry.add_model(model)

@registry.report("my_report", description="...")
def my_report():
    ds = model.load("orders")
    return Report("My Report").add(TableSection(dataset=ds, title="Orders"))

registry.add_dashboard("sales", DashboardServer(dashboard, model=model))
registry.add_pipeline("main", runner)

Medallion-aware demo

demo_app.py detects data/tracebi.db at startup and adapts:

• Silver tables present → reports and dashboard read from orders_silver ⋈ customers_silver
• Gold tables present → additional gold_revenue_by_region report registered
• DB absent or empty → falls back to in-memory MemoryConnector data with a warning
• Pipeline always registered if DB file exists → Pipelines page shows layer status regardless

Project Structure

Install groups

Quick Start

Run the examples

Run tests

New ad hoc report

Run the web UI

Pull latest (after a session)