routeflow-python 0.1.3

Python SDK for Routeflow backend telemetry

Routeflow Python SDK

Production-ready backend instrumentation for Flask, FastAPI, and Django applications.

Automatically capture and send backend request telemetry to Routeflow with zero per-route changes. Non-blocking, efficient, and privacy-focused.

Features

• ✅ One-line integration - Just init_routeflow(app)
• ✅ SQL query capture - Automatically track database queries (see SQL_CAPTURE.md)
• ✅ Non-blocking - Background worker with batching, no request latency
• ✅ Privacy-first - No request bodies, headers (except trace headers), or query strings
• ✅ Production-ready - Retry logic, queue limits, graceful degradation
• ✅ Lightweight - Minimal dependencies (just requests)
• ✅ Observable - Built-in stats for monitoring SDK behavior

Installation

pip install routeflow-python

Or install with your framework:

# For Flask
pip install routeflow-python[flask]

# For FastAPI
pip install routeflow-python[fastapi]

# For Django
pip install routeflow-python[django]

# For all frameworks
pip install routeflow-python[all]

Quick Start

1. Set Environment Variables

export ROUTEFLOW_INGEST_KEY=rf_live_your-ingest-key-here
export ROUTEFLOW_ENV=production

The Routeflow SDK automatically sends data to Routeflow Cloud.

2. Initialize in Your App

Choose your framework:

Flask

from flask import Flask
from routeflow_python import init_routeflow

app = Flask(__name__)

# ONE LINE - that's it!
init_routeflow(app)

@app.route('/api/users')
def get_users():
    return {'users': []}

@app.route('/api/users/<user_id>')
def get_user(user_id):
    return {'id': user_id, 'name': 'John'}

if __name__ == '__main__':
    app.run()

FastAPI

from fastapi import FastAPI
from routeflow_python import init_routeflow_fastapi

app = FastAPI()

# ONE LINE - that's it!
init_routeflow_fastapi(app)

@app.get("/api/users")
def get_users():
    return {"users": []}

@app.get("/api/users/{user_id}")
def get_user(user_id: int):
    return {"id": user_id, "name": "John"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Django

Add middleware in settings.py:

# settings.py
from routeflow_python import init_routeflow_django

MIDDLEWARE = [
    'django.middleware.security.SecurityMiddleware',
    # ... other middleware
    'routeflow_python.django.RouteflowDjangoMiddleware',  # Add this
]

# Initialize Routeflow
init_routeflow_django()

Then define views as normal:

# views.py
from django.http import JsonResponse

def get_users(request):
    return JsonResponse({"users": []})

def get_user(request, user_id):
    return JsonResponse({"id": user_id, "name": "John"})

# urls.py
from django.urls import path
from .views import get_users, get_user

urlpatterns = [
    path('api/users/', get_users),
    path('api/users/<int:user_id>/', get_user),
]

3. That's It! 🎉

Routeflow will now automatically capture:

• Request timestamps
• HTTP methods and routes (e.g., GET /api/users/<user_id>)
• Status codes
• Latency
• Environment info
• Code version (from CI variables)

What Data is Collected?

✅ Collected

• Timestamp: When the request started
• Trace ID: From x-routeflow-trace-id header or auto-generated
• Frontend Route: From x-routeflow-frontend-route header (e.g., /checkout)
• Backend Method: HTTP method (GET, POST, etc.)
• Backend Path: Route pattern (e.g., /api/orders, /api/users/{user_id})
• Handler: Endpoint/view name (e.g., api.create_order, myapp.views.get_user)
• Status Code: HTTP response status
• Latency: Request duration in milliseconds
• Environment: From ROUTEFLOW_ENV
• Code Version: From ROUTEFLOW_CODE_VERSION or CI variables

❌ NOT Collected

• Request bodies
• Request headers (except trace headers)
• Query strings
• Path parameters
• User data
• Secrets or credentials
• Exception messages or stack traces

Privacy is built-in. Routes are automatically sanitized to remove query strings.

Configuration

Environment Variables

Variable	Required	Default	Description
ROUTEFLOW_INGEST_KEY	✅ Yes	-	API key for authentication
ROUTEFLOW_ENV	No	development	Environment name
ROUTEFLOW_ENABLED	No	true	Enable/disable SDK
ROUTEFLOW_SAMPLE_RATE	No	1.0	Sampling rate (0.0-1.0)
ROUTEFLOW_CODE_VERSION	No	Auto-detected	Code/git version
ROUTEFLOW_MAX_EVENTS_PER_BATCH	No	50	Max events per batch
ROUTEFLOW_FLUSH_INTERVAL_MS	No	1000	Flush interval in ms
ROUTEFLOW_QUEUE_MAXSIZE	No	5000	Max queue size
ROUTEFLOW_TIMEOUT_MS	No	1500	HTTP timeout in ms

Programmatic Configuration

You can override environment variables programmatically:

init_routeflow(
    app,
    ingest_key="rf_live_your-key",
    environment="production",
    sample_rate=0.1,  # Sample 10% of requests
)

Performance & Behavior

Non-Blocking Design

The SDK is designed to never slow down your application:

1. No I/O on request thread - Events are enqueued instantly
2. Background worker - Separate thread batches and sends events
3. Bounded queue - If queue fills, events are dropped (not blocked)
4. Graceful degradation - Backend down? Events are dropped safely

Batching

Events are batched for efficiency:

• Batches up to 50 events (configurable)
• Flushes every 1000ms (configurable)
• Automatic flush on process exit (best-effort)

Retry Logic

• 5xx errors: Retry with exponential backoff (3 attempts max)
• 4xx errors: No retry (validation/auth errors)
• Network errors: Retry with backoff
• Final failure: Drop batch and increment counter

Queue Behavior

If the queue fills up (5000 events by default):

• New events are dropped immediately
• A warning is logged
• Counter dropped_queue_full is incremented
• Your app is never blocked

Monitoring SDK Health

Get real-time statistics:

from routeflow_python import get_routeflow_stats

stats = get_routeflow_stats()
print(stats)

Output:

{
    'enqueued': 1523,
    'sent': 1500,
    'dropped_queue_full': 0,
    'dropped_sampled': 23,
    'dropped_validation': 0,
    'dropped_http_4xx': 0,
    'dropped_http_5xx': 0,
    'retries': 2,
    'last_error_at': None
}

Metrics Explained

• enqueued: Events successfully queued
• sent: Events successfully delivered
• dropped_sampled: Events skipped due to sampling
• dropped_queue_full: Events dropped due to full queue
• dropped_validation: Events failed validation
• dropped_http_4xx: Events dropped due to 4xx errors
• dropped_http_5xx: Events dropped after retries failed
• retries: Number of retry attempts
• last_error_at: Timestamp of last error (ISO 8601)

Frontend Integration

Your frontend should send these headers:

fetch('/api/users', {
  headers: {
    'x-routeflow-trace-id': generateTraceId(), // UUID
    'x-routeflow-frontend-route': window.location.pathname, // e.g., "/checkout"
  }
})

If these headers are not present:

• trace_id is auto-generated
• frontend_route defaults to "__unknown__"

Advanced Usage

Manual Shutdown

For testing or graceful shutdown:

from routeflow_python import shutdown_routeflow

# Flush events and stop worker (waits up to 2 seconds)
shutdown_routeflow(timeout_sec=2.0)

This is called automatically on process exit via atexit.

Sampling

Sample only a percentage of requests:

# Sample 10% of requests
init_routeflow(app, sample_rate=0.1)

Or via environment:

export ROUTEFLOW_SAMPLE_RATE=0.1

Disable in Development

export ROUTEFLOW_ENABLED=false

Or:

init_routeflow(app, enabled=False)

Example Application

from flask import Flask, jsonify
from routeflow_python import init_routeflow, get_routeflow_stats

app = Flask(__name__)
init_routeflow(app)

@app.route('/')
def index():
    return {'message': 'Hello World'}

@app.route('/api/users/<int:user_id>')
def get_user(user_id):
    return {'id': user_id, 'name': 'John'}

@app.route('/_stats/routeflow')
def routeflow_stats():
    """Endpoint to check SDK health."""
    return jsonify(get_routeflow_stats())

if __name__ == '__main__':
    app.run(debug=True)

Test it:

curl http://localhost:5000/api/users/123
curl http://localhost:5000/_stats/routeflow

Error Handling

The SDK is designed to never crash your application:

• All exceptions are caught and logged
• Failed events are dropped, not retried indefinitely
• Queue overflow drops events instead of blocking
• Invalid configuration logs warnings but doesn't crash

Code Version Detection

The SDK automatically detects code version from:

1. ROUTEFLOW_CODE_VERSION (explicit)
2. VERCEL_GIT_COMMIT_SHA (Vercel)
3. GIT_SHA (generic)
4. COMMIT_SHA (generic)
5. SOURCE_VERSION (Azure)
6. HEROKU_SLUG_COMMIT (Heroku)

Or set it explicitly:

export ROUTEFLOW_CODE_VERSION=$(git rev-parse HEAD)

Requirements

• Python 3.10+
• requests >= 2.28.0

Framework-specific:

• Flask 2.x or 3.x (for Flask integration)
• FastAPI 0.100+ (for FastAPI integration)
• Django 3.2+ / 4.x / 5.x (for Django integration)

License

MIT

Support

For issues, questions, or feature requests, visit the Routeflow repository.

routeflow-python