Chapter 163 · Omnibus Instrument Product Analytics
Subchapter 163.14
references/EXAMPLE-flask.mdMarkdown32 KBView on GitHub
Repository: https://github.com/PostHog/context-mill Path: basics/flask
A Flask application demonstrating PostHog integration for analytics, feature flags, and error tracking.
Create and activate a virtual environment:
python -m venv venv
source venv/bin/activate # On Windows: venv\Scripts\activateInstall dependencies:
pip install -r requirements.txtCopy the environment file and configure:
cp .env.example .env
# Edit .env with your PostHog project keyRun the application:
python run.pyOpen http://localhost:5001 (opens in a new tab) and either:
admin@example.com / adminNew users are identified and tracked on signup using the context-based API:
with new_context():
identify_context(user.email)
tag('email', user.email)
tag('is_staff', user.is_staff)
capture('user_signed_up', properties={'signup_method': 'form'})Users are identified on login with their properties:
with new_context():
identify_context(user.email)
tag('email', user.email)
tag('is_staff', user.is_staff)
capture('user_logged_in', properties={'login_method': 'password'})Custom events are captured throughout the app:
with new_context():
identify_context(current_user.email)
capture('burrito_considered', properties={'total_considerations': count})The dashboard demonstrates feature flag checking:
show_new_feature = posthog.feature_enabled(
'new-dashboard-feature',
current_user.email,
person_properties={'email': current_user.email, 'is_staff': current_user.is_staff}
)
feature_config = posthog.get_feature_flag_payload('new-dashboard-feature', current_user.email)The example demonstrates two approaches to error tracking:
Manual capture for specific critical operations** (app/api/routes.py).
try:
# Critical operation that might fail
result = process_payment()
except Exception as e:
# Manually capture this specific exception
with new_context():
identify_context(current_user.email)
event_id = posthog.capture_exception(e)
return jsonify({
"error": "Operation failed",
"error_id": event_id,
"message": f"Error captured in PostHog. Reference ID: {event_id}"
}), 500The /api/test-error endpoint demonstrates manual exception capture. Use ?capture=true to capture in PostHog, or ?capture=false to skip tracking.
basics/flask/
├── app/
│ ├── __init__.py # Application factory
│ ├── config.py # Configuration classes
│ ├── extensions.py # Extension instances
│ ├── models.py # User model (SQLAlchemy)
│ ├── main/
│ │ ├── __init__.py # Main blueprint
│ │ └── routes.py # View functions
│ ├── templates/ # HTML templates
│ └── api/
│ ├── __init__.py # API blueprint
│ └── routes.py # API endpoints
├── .env.example
├── .gitignore
├── requirements.txt
├── README.md
└── run.py # Entry pointPOSTHOG_PROJECT_TOKEN=<ph_project_token>
POSTHOG_HOST=https://us.i.posthog.com
FLASK_SECRET_KEY=your-secret-key-here
FLASK_DEBUG=True
POSTHOG_DISABLED=False
"""Flask application factory."""
import posthog
from flask import Flask, g, jsonify, render_template, request
from flask_login import current_user
from posthog import identify_context, new_context
from werkzeug.exceptions import HTTPException
from app.config import config
from app.extensions import db, login_manager
def create_app(config_name="default"):
"""Application factory."""
app = Flask(__name__)
app.config.from_object(config[config_name])
# Initialize extensions
db.init_app(app)
login_manager.init_app(app)
# Initialize PostHog
if not app.config["POSTHOG_DISABLED"]:
posthog.api_key = app.config["POSTHOG_PROJECT_TOKEN"]
posthog.host = app.config["POSTHOG_HOST"]
posthog.debug = app.config["DEBUG"]
# Import models after db is initialized
from app.models import User
# User loader for Flask-Login
@login_manager.user_loader
def load_user(user_id):
return User.get_by_id(user_id)
# Simple error handlers - no automatic PostHog capture
# Capture exceptions manually only where it makes sense (e.g., test endpoints)
@app.errorhandler(404)
def page_not_found(e):
if request.path.startswith('/api/'):
return jsonify({"error": "Not found"}), 404
return render_template('errors/404.html'), 404
@app.errorhandler(500)
def internal_server_error(e):
if request.path.startswith('/api/'):
return jsonify({"error": "Internal server error"}), 500
return render_template('errors/500.html'), 500
# Register blueprints
from app.api import api_bp
from app.main import main_bp
app.register_blueprint(main_bp)
app.register_blueprint(api_bp, url_prefix="/api")
# Create database tables and seed default admin user
with app.app_context():
db.create_all()
if not User.get_by_email("admin@example.com"):
User.create_user(
email="admin@example.com",
password="admin",
is_staff=True,
)
return app
"""API blueprint registration."""
from flask import Blueprint
api_bp = Blueprint("api", __name__)
from app.api import routes # noqa: E402, F401
"""API endpoints demonstrating PostHog integration patterns."""
import posthog
from flask import jsonify, request, session
from flask_login import current_user, login_required
from posthog import capture, identify_context, new_context
from app.api import api_bp
@api_bp.route("/burrito/consider", methods=["POST"])
@login_required
def consider_burrito():
"""Track burrito consideration event."""
# Increment session counter
burrito_count = session.get("burrito_count", 0) + 1
session["burrito_count"] = burrito_count
# PostHog: Capture custom event
with new_context():
identify_context(current_user.email)
capture("burrito_considered", properties={"total_considerations": burrito_count})
return jsonify({"success": True, "count": burrito_count})
@api_bp.route("/test-error", methods=["POST"])
@login_required
def test_error():
"""Test endpoint demonstrating manual exception capture in PostHog.
Shows how to intentionally capture specific errors in PostHog.
Use this pattern for critical operations where you want error tracking.
Query params:
- capture: "true" to capture the exception in PostHog, "false" to just raise it
"""
should_capture = request.args.get("capture", "true").lower() == "true"
try:
# Simulate a critical operation failure
raise Exception("Test exception from critical operation")
except Exception as e:
if should_capture:
# Manually capture this specific exception in PostHog
with new_context():
identify_context(current_user.email)
event_id = posthog.capture_exception(e)
return jsonify({
"error": "Operation failed",
"error_id": event_id,
"message": f"Error captured in PostHog. Reference ID: {event_id}"
}), 500
else:
# Just return error without PostHog capture
return jsonify({"error": str(e)}), 500
"""Flask application configuration."""
import os
from dotenv import load_dotenv
load_dotenv()
class Config:
"""Base configuration."""
SECRET_KEY = os.environ.get("FLASK_SECRET_KEY", "dev-secret-key-change-in-production")
# Database configuration (SQLite like Django example)
SQLALCHEMY_DATABASE_URI = os.environ.get("DATABASE_URL", "sqlite:///db.sqlite3")
SQLALCHEMY_TRACK_MODIFICATIONS = False
# PostHog configuration
POSTHOG_PROJECT_TOKEN = os.environ.get("POSTHOG_PROJECT_TOKEN", "<ph_project_token>")
POSTHOG_HOST = os.environ.get("POSTHOG_HOST", "https://us.i.posthog.com")
POSTHOG_DISABLED = os.environ.get("POSTHOG_DISABLED", "False").lower() == "true"
class DevelopmentConfig(Config):
"""Development configuration."""
DEBUG = True
class ProductionConfig(Config):
"""Production configuration."""
DEBUG = False
config = {
"development": DevelopmentConfig,
"production": ProductionConfig,
"default": DevelopmentConfig,
}
"""Flask extensions initialized without binding to app."""
from flask_login import LoginManager
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()
login_manager = LoginManager()
login_manager.login_view = "main.home"
login_manager.login_message = "Please log in to access this page."
"""Main blueprint registration."""
from flask import Blueprint
main_bp = Blueprint("main", __name__, template_folder="../templates")
from app.main import routes # noqa: E402, F401
"""Core view functions demonstrating PostHog integration patterns."""
import posthog
from flask import flash, redirect, render_template, request, session, url_for
from flask_login import current_user, login_required, login_user, logout_user
from posthog import capture, identify_context, new_context, tag
from app.main import main_bp
from app.models import User
@main_bp.route("/", methods=["GET", "POST"])
def
"""User model with SQLite persistence (similar to Django's auth.User)."""
from datetime import datetime, timezone
from flask_login import UserMixin
from werkzeug.security import check_password_hash, generate_password_hash
from app.extensions import db
class User(UserMixin, db.Model):
"""User model with SQLite persistence."""
__tablename__ = "users"
id = db.Column(db.Integer, primary_key=True)
email = db.Column(db.String(254), unique=True, nullable=False)
password_hash = db.Column(db.String(256), nullable=False)
is_staff = db.Column(db.Boolean, default=False)
is_active = db.Column(db.Boolean, default=True)
date_joined = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
def set_password(self, password):
"""Hash and set the user's password."""
self.password_hash = generate_password_hash(password)
def check_password(self, password):
"""Verify the password against the hash."""
return check_password_hash(self.password_hash, password)
@classmethod
def create_user(cls, email, password, is_staff=False):
"""Create and save a new user."""
user = cls(email=email, is_staff=is_staff)
# nosemgrep: python.django.security.audit.unvalidated-password.unvalidated-password
user.set_password(password)
db.session.add(user)
db.session.commit()
return user
@classmethod
def get_by_id(cls, user_id):
"""Get user by ID."""
return cls.query.get(int(user_id))
@classmethod
def get_by_email(cls, email):
"""Get user by email."""
return cls.query.filter_by(email=email).first()
@classmethod
def authenticate(cls, email, password):
"""Authenticate user with email and password."""
user = cls.get_by_email(email)
if user and user.check_password(password):
return user
return None
def __repr__(self):
return f"<User {self.email}>"
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>{% block title %}PostHog Flask Example{% endblock %}</title>
<style>
* {
box-sizing
{% extends "base.html" %}
{% block title %}Burrito - PostHog Flask Example{% endblock %}
{% block content %}
<div class="card">
<h1>Burrito Consideration Tracker</h1>
<p>This page demonstrates custom event tracking with PostHog.</p>
<div class="count" id="burrito-count">{{ burrito_count }}</div>
<p style="text-align: center; color: #666;">Times you've considered a burrito</p>
<div style="text-align: center; margin-top: 20px;">
<button onclick="considerBurrito()">Consider a Burrito</button>
</div>
</div>
<div class="card">
<h3>Code Example</h3>
<pre>
# API endpoint captures the event
with new_context():
identify_context(current_user.email)
capture('burrito_considered', properties={
'total_considerations': burrito_count
})</pre>
</div>
{% endblock %}
{% block scripts %}
<script>
async function considerBurrito() {
try {
const response = await fetch('/api/burrito/consider', {
method: 'POST',
headers: {
'Content-Type': 'application/json'
}
});
const data = await response.json();
if (data.success) {
document.getElementById('burrito-count').textContent = data.count;
}
} catch (error) {
console.error('Error:', error);
}
}
</script>
{% endblock %}
{% extends "base.html" %}
{% block title %}Dashboard - PostHog Flask Example{% endblock %}
{% block content %}
<div class="card">
<h1>Dashboard</h1>
<p>Welcome back, {{ current_user.username }}!</p>
</div>
<div class="card">
<h2>Feature Flags</h2>
{% if show_new_feature %}
<div class="feature-flag">
<strong>New Feature Enabled!</strong>
<p>You're seeing this because the <code>new-dashboard-feature</code> flag is enabled for you.</p>
{% if feature_config %}
<p><strong>Feature Configuration:</strong></p>
<pre>{{ feature_config | tojson(indent=2) }}</pre>
{% endif %}
</div>
{% else %}
<p>The <code>new-dashboard-feature</code> flag is not enabled for your account.</p>
{% endif %}
<h3 style="margin-top: 20px;">Code Example</h3>
<pre>
# Check if feature flag is enabled
show_new_feature = posthog.feature_enabled(
'new-dashboard-feature',
user_id,
person_properties={
'email': current_user.email,
'is_staff': current_user.is_staff
}
)
# Get feature flag payload
feature_config = posthog.get_feature_flag_payload(
'new-dashboard-feature',
user_id
)</pre>
</div>
{% endblock %}
{% extends "base.html" %}
{% block title %}404 - Page Not Found{% endblock %}
{% block content %}
<div class="card" style="text-align: center; padding: 60px 20px;">
<h1 style="font-size: 72px; color: #dc2626; margin-bottom: 10px;">404</h1>
<h2 style="color: #333; margin-bottom: 20px;">Page Not Found</h2>
<p style="font-size: 18px; color: #666; margin-bottom: 30px;">
The page you're looking for doesn't exist or has been moved.
</p>
{% if error_id %}
<div style="background: #fef3c7; border: 1px solid #fbbf24; border-radius: 8px; padding: 15px; margin: 30px 0;">
<p style="color: #92400e; margin-bottom: 5px; font-weight: 600;">Error Reference ID:</p>
<code style="background: #fff; padding: 5px 10px; border-radius: 4px; font-family: monospace; color: #1e40af;">{{ error_id }}</code>
<p style="color: #92400e; margin-top: 10px; font-size: 14px;">
Share this ID with support if you need assistance.
</p>
</div>
{% endif %}
<div style="margin-top: 40px;">
<a href="{{ url_for('main.home') }}" class="btn" style="margin-right: 10px;">Go to Home</a>
{% if current_user.is_authenticated %}
<a href="{{ url_for('main.dashboard') }}" class="btn">Go to Dashboard</a>
{% endif %}
</div>
</div>
{% endblock %}
{% extends "base.html" %}
{% block title %}500 - Internal Server Error{% endblock %}
{% block content %}
<div class="card" style="text-align: center; padding: 60px 20px;">
<h1 style="font-size: 72px; color: #dc2626; margin-bottom: 10px;">500</h1>
<h2 style="color: #333; margin-bottom: 20px;">Internal Server Error</h2>
<p style="font-size: 18px; color: #666; margin-bottom: 30px;">
Something went wrong on our end. We've been notified and are looking into it.
</p>
{% if error_id %}
<div style="background: #fef3c7; border: 1px solid #fbbf24; border-radius: 8px; padding: 15px; margin: 30px 0;">
<p style="color: #92400e; margin-bottom: 5px; font-weight: 600;">Error Reference ID:</p>
<code style="background: #fff; padding: 5px 10px; border-radius: 4px; font-family: monospace; color: #1e40af;">{{ error_id }}</code>
<p style="color: #92400e; margin-top: 10px; font-size: 14px;">
Share this ID with support if you need assistance. This error has been logged in PostHog.
</p>
</div>
{% endif %}
{% if error and config.DEBUG %}
<div style="background: #fee2e2; border: 1px solid #dc2626; border-radius: 8px; padding: 15px; margin: 30px 0; text-align: left;">
<p style="color: #7f1d1d; margin-bottom: 5px; font-weight: 600;">Debug Information:</p>
<code style="background: #fff; padding: 10px; border-radius: 4px; font-family: monospace; color: #dc2626; display: block; overflow-x: auto;">{{ error }}</code>
</div>
{% endif %}
<div style="margin-top: 40px;">
<a href="{{ url_for('main.home') }}" class="btn" style="margin-right: 10px;">Go to Home</a>
{% if current_user.is_authenticated %}
<a href="{{ url_for('main.dashboard') }}" class="btn">Go to Dashboard</a>
{% endif %}
</div>
</div>
{% endblock %}
{% extends "base.html" %}
{% block title %}Login - PostHog Flask Example{% endblock %}
{% block content %}
<div class="card">
<h1>Welcome to PostHog Flask Example</h1>
<p>This example demonstrates how to integrate PostHog with a Flask application.</p>
<form method="POST">
<label for="email">Email</label>
<input type="email" id="email" name="email" required>
<label for="password">Password</label>
<input type="password" id="password" name="password" required>
<button type="submit">Login</button>
</form>
<p style="margin-top: 16px; font-size: 14px; color: #666;">
Don't have an account? <a href="{{ url_for('main.signup') }}">Sign up here</a>
</p>
<p style="font-size: 14px; color: #666;">
<strong>Tip:</strong> Default credentials are admin@example.com/admin
</p>
</div>
<div class="card">
<h2>Features Demonstrated</h2>
<ul style="margin-left: 20px; color: #666;">
<li>User registration and identification</li>
<li>Event tracking</li>
<li>Feature flags</li>
<li>Error tracking</li>
<li>Group analytics</li>
</ul>
</div>
{% endblock %}
{% extends "base.html" %}
{% block title %}Profile - PostHog Flask Example{% endblock %}
{% block content %}
<div class="card">
<h1>Your Profile</h1>
<p>This page demonstrates error tracking with PostHog.</p>
<table>
<tr>
<th>Email</th>
<td>{{ current_user.email }}</td>
</tr>
<tr>
<th>Date Joined</th>
<td>{{ current_user.date_joined.strftime('%Y-%m-%d %H:%M') }}</td>
</tr>
<tr>
<th>Staff Status</th>
<td>{{ 'Yes' if current_user.is_staff else 'No' }}</td>
</tr>
</table>
</div>
<div class="card">
<h2>Error Tracking Demo</h2>
<p>Click a button to trigger an error and see it captured in PostHog:</p>
<div style="margin: 20px 0;">
<button class="danger" onclick="triggerError('value')">
Trigger ValueError
</button>
<button class="danger" onclick="triggerError('key')">
Trigger KeyError
</button>
<button class="danger" onclick="triggerError('generic')">
Trigger Generic Error
</button>
</div>
<div id="error-result" style="display: none;" class="message"></div>
</div>
<div class="card">
<h3>Code Example</h3>
<pre>
try:
raise ValueError('Invalid value provided')
except Exception as e:
# Capture exception and event with user context
with new_context():
identify_context(current_user.email)
posthog.capture_exception(e)
capture('error_triggered', properties={
'error_type': 'value',
'error_message': str(e)
})</pre>
</div>
{% endblock %}
{% block scripts %}
<script>
async function triggerError(errorType) {
const resultDiv = document.getElementById('error-result');
try {
const formData = new FormData();
formData.append('error_type', errorType);
const response = await fetch('/api/trigger-error', {
method: 'POST',
body: formData
});
const data = await response.json();
resultDiv.style.display = 'block';
resultDiv.className = 'message ' + (data.success ? 'success' : 'error');
resultDiv.textContent = data.message + ': ' + data.error;
} catch (error) {
console.error('Error:', error);
resultDiv.style.display = 'block';
resultDiv.className = 'message error';
resultDiv.textContent = 'Request failed: ' + error.message;
}
}
</script>
{% endblock %}
{% extends "base.html" %}
{% block title %}Sign Up - PostHog Flask Example{% endblock %}
{% block content %}
<div class="card">
<h1>Create an Account</h1>
<p>Sign up to explore the PostHog Flask integration example.</p>
<form method="POST">
<label for="email">Email *</label>
<input type="email" id="email" name="email" required>
<label for="password">Password *</label>
<input type="password" id="password" name="password" required>
<label for="password_confirm">Confirm Password *</label>
<input type="password" id="password_confirm" name="password_confirm" required>
<button type="submit">Sign Up</button>
</form>
<p style="margin-top: 16px; font-size: 14px; color: #666;">
Already have an account? <a href="{{ url_for('main.home') }}">Login here</a>
</p>
</div>
<div class="card">
<h2>PostHog Integration</h2>
<p>When you sign up, the following PostHog events are captured:</p>
<ul style="margin-left: 20px; color: #666;">
<li><code>identify_context()</code> - Associates your email with the context</li>
<li><code>tag()</code> - Sets person properties (email, etc.)</li>
<li><code>user_signed_up</code> event - Tracks the signup action</li>
</ul>
<h3 style="margin-top: 20px;">Code Example</h3>
<pre>
# After creating the user
with new_context():
identify_context(user.email)
tag('email', user.email)
tag('is_staff', user.is_staff)
tag('date_joined', user.date_joined.isoformat())
capture('user_signed_up', properties={'signup_method': 'form'})</pre>
</div>
{% endblock %}
Flask>=3.1.0
Flask-Login>=0.6.3
Flask-SQLAlchemy>=3.1.0
python-dotenv>=1.0.0
posthog>=3.0.0
Werkzeug>=3.0.0
"""Development server entry point."""
from app import create_app
app = create_app()
if __name__ == "__main__":
app.run(port=5001)