flowsurgeon 0.5.1

FlowSurgeon — framework-agnostic profiling middleware for Python (WSGI & ASGI).

FlowSurgeon

Framework-agnostic profiling middleware for Python — drop-in debug UI for Flask and FastAPI.

---

FlowSurgeon wraps your existing WSGI or ASGI app with a single line. It injects a collapsible debug panel into every HTML response and stores a full request history — timing, headers, SQL queries, response bodies — in a local SQLite database, with a built-in dark-themed UI at /flowsurgeon.

Features

• Zero application changes — wraps any WSGI or ASGI callable
• Auto-detect WSGI vs ASGI via the FlowSurgeon() factory
• Inline debug panel injected before </body> in every HTML response
• Built-in history UI at /flowsurgeon — no extra server needed
• Request grid view — browse captured requests sorted by query count, duration, or path
• SQL query tracking via SQLAlchemy and DB-API 2.0 (sqlite3, psycopg2, …)
• Call-stack profiling — cProfile-based per-request profiling with a sortable stats table and callers drilldown (opt-in via enable_profiling=True)
• Route auto-discovery from Flask (url_map) and FastAPI/Starlette (app.routes)
• Response body capture — stores up to 128 KB for text/JSON/XML responses
• SQLite persistence with auto-pruning (configurable max records)
• Sensitive header redaction — Authorization, Cookie, Set-Cookie stripped by default
• FLOWSURGEON_ENABLED env var — safe to ship in codebase; disabled by default

Installation

# Recommended
uv add flowsurgeon

# pip
pip install flowsurgeon

Requires Python 3.12+. The only runtime dependency is jinja2.

[!WARNING] FlowSurgeon is a development tool. Do not enable in production. See Security for details.

Quick start

FastAPI / Starlette (ASGI)

from fastapi import FastAPI
from flowsurgeon import FlowSurgeon, Config

_app = FastAPI()

app = FlowSurgeon(
    _app,
    config=Config(enabled=True),
)

@_app.get("/books")
async def books():
    return {"books": ["Clean Code", "Refactoring"]}

uvicorn myapp:app --reload
# Debug UI → http://127.0.0.1:8000/flowsurgeon

Flask (WSGI)

from flask import Flask
from flowsurgeon import FlowSurgeon, Config

app = Flask(__name__)

app.wsgi_app = FlowSurgeon(
    app.wsgi_app,
    config=Config(enabled=True),
)

flask run
# Debug UI → http://127.0.0.1:5000/flowsurgeon

Security

FlowSurgeon is designed for local development — these settings keep it safe and contained.

Kill Switch

The middleware is a no-op unless explicitly enabled, so it is safe to ship in your codebase. Enable it per-environment via an env var or your settings layer.

# Enable without modifying code
FLOWSURGEON_ENABLED=1 uvicorn myapp:app

from flowsurgeon import Config

# Or enable in code
Config(enabled=True)

Allowed Hosts

Only requests from listed IPs can access the debug panel. By default this is restricted to localhost addresses.

from flowsurgeon import Config

Config(
    # Defaults — loopback only
    allowed_hosts=["127.0.0.1", "::1", "localhost"],
)

To allow an additional machine on your local network:

from flowsurgeon import Config

Config(
    allowed_hosts=["127.0.0.1", "::1", "localhost", "192.168.1.50"],
)

Header Redaction

Sensitive header values are replaced with [redacted] before being stored in the database. The default list covers the most common credential-bearing headers.

from flowsurgeon import Config

Config(
    # Defaults — authorization, session cookies redacted
    strip_sensitive_headers=["authorization", "cookie", "set-cookie"],
)

To redact an additional header:

from flowsurgeon import Config

Config(
    strip_sensitive_headers=["authorization", "cookie", "set-cookie", "x-api-key"],
)

Database File

FlowSurgeon writes request data to a local SQLite file (flowsurgeon.db by default). Add it to your project's .gitignore so it is never committed.

# .gitignore
flowsurgeon.db

To store the database elsewhere:

from flowsurgeon import Config

Config(db_path="/tmp/flowsurgeon.db")

SQL query tracking

SQLAlchemy

from sqlalchemy import create_engine
from flowsurgeon import FlowSurgeon, Config
from flowsurgeon.trackers.sqlalchemy import SQLAlchemyTracker

engine = create_engine("sqlite:///mydb.db")
tracker = SQLAlchemyTracker(engine, capture_stacktrace=False)

app = FlowSurgeon(
    asgi_app,
    config=Config(enabled=True),
    trackers=[tracker],
)

DB-API 2.0 (sqlite3, psycopg2, …)

import sqlite3
from flowsurgeon import FlowSurgeon, Config
from flowsurgeon.trackers import DBAPITracker

raw_conn = sqlite3.connect("mydb.db")
tracker = DBAPITracker(raw_conn)
conn = tracker.connection  # use this instead of raw_conn everywhere

app = FlowSurgeon(
    wsgi_app,
    config=Config(enabled=True),
    trackers=[tracker],
)

DBAPITracker works via a transparent proxy — replace your connection object with tracker.connection and every cursor().execute() call is automatically timed and recorded.

Configuration

from flowsurgeon import Config

Config(
    # Master switch — default False. Also controlled by FLOWSURGEON_ENABLED env var.
    enabled=True,

    # Only serve the debug panel to requests from these hosts.
    allowed_hosts=["127.0.0.1", "::1", "localhost"],

    # SQLite file for request history storage.
    db_path="flowsurgeon.db",

    # Prune oldest records when this limit is exceeded.
    max_stored_requests=1000,

    # URL prefix for the built-in debug UI.
    debug_route="/flowsurgeon",

    # Headers replaced with "[redacted]" before storage.
    strip_sensitive_headers=["authorization", "cookie", "set-cookie"],

    # SQL query tracking options.
    track_queries=True,
    slow_query_threshold_ms=100.0,
    capture_query_stacktrace=False,

    # Call-stack profiling (cProfile). Off by default; adds ~1-10% overhead.
    # Also controlled by FLOWSURGEON_PROFILING env var.
    enable_profiling=False,
    profile_top_n=50,              # keep top N functions by cumulative time
    profile_user_code_only=True,   # filter out stdlib + third-party frames

    # Manually register routes shown in the UI before any traffic.
    # Flask and FastAPI routes are auto-discovered; use this for other cases.
    known_routes=[("GET", "/health"), ("POST", "/webhooks/stripe")],
)

Debug UI

URL	Description
/flowsurgeon	Requests grid — all captured requests with latency and query info
/flowsurgeon?view=profiling	Profiling tab — list of profiled requests with top hotspot function
/flowsurgeon?q=/books	Filter requests by path
/flowsurgeon?order=duration	Sort by duration (also: queries, path)
/flowsurgeon/{request_id}	Request detail: headers, response body, SQL, tracebacks, profile

Requests view

Displays all captured requests as a card grid, sorted by number of queries by default. Each card shows: status code, HTTP method, path, total duration, query time and count. Supports filtering by path and ordering by query count, duration, or path.

Request detail — four tabs

• Details — stat cards (status, duration, SQL count, SQL time); request headers; response headers and body (up to 128 KB for text/JSON content types)
• SQL — every captured query with timing, slow badge (exceeds threshold), dup badge (same SQL run more than once), and bound params
• Traceback — Python stack trace per query (requires capture_query_stacktrace=True)
• Profile — cProfile call-stack stats: function name, file:line, calls, own time, total time, visual time bar, and a native <details> callers drilldown per function (requires enable_profiling=True)

Call-stack profiling

app = FlowSurgeon(
    _app,
    config=Config(
        enabled=True,
        enable_profiling=True,         # enable cProfile per request
        profile_top_n=50,              # keep top 50 functions by cumulative time
        profile_user_code_only=True,   # hide stdlib + third-party frames
    ),
)

Or via environment variable (no code changes needed):

FLOWSURGEON_ENABLED=1 FLOWSURGEON_PROFILING=1 uvicorn myapp:app

ASGI note: cProfile measures CPU time while the coroutine is on-thread. I/O-wait time (e.g. awaiting a DB call) appears as event-loop time rather than in the awaited coroutine's frame. CPU-bound hotspots are captured accurately.

Running the examples

# FastAPI + SQLAlchemy
uv run --group examples uvicorn examples.fastapi.demo_fastapi:app --reload

# Flask + DB-API (sqlite3)
uv run --group examples python examples/flask/demo_flask.py

Debug UI:

• FastAPI → http://127.0.0.1:8000/flowsurgeon
• Flask → http://127.0.0.1:5000/flowsurgeon

Both demos expose these routes:

Route	What it demonstrates
GET /books	Normal query — 1 SQL
GET /books/{id}	Parametrised query
GET /books/duplicates	Same query twice → dup badge
GET /books/slow	Query exceeds threshold → slow badge
GET /slow	Slow endpoint, no SQL
GET /boom	500 error

Environment variable

See Security for the FLOWSURGEON_ENABLED kill switch and other security settings.

License

MIT