routeflow-python 0.1.3
Python SDK for Routeflow backend telemetry
License: MIT
Requires Python: >=3.10

# 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](./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

```bash
pip install routeflow-python
```

Or install with your framework:

```bash
# 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

```bash
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

```python
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

```python
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`:

```python
# 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:

```python
# 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:

```python
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:

```python
from routeflow_python import get_routeflow_stats

stats = get_routeflow_stats()
print(stats)
```

Output:
```python
{
    '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:

```javascript
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:

```python
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:

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

Or via environment:
```bash
export ROUTEFLOW_SAMPLE_RATE=0.1
```

### Disable in Development

```bash
export ROUTEFLOW_ENABLED=false
```

Or:
```python
init_routeflow(app, enabled=False)
```

## Example Application

```python
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:
```bash
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:
```bash
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](https://github.com/routeflow/routeflow).
# routeflow-python