Framework Middleware

Flask Extension

Flask extension with decorators and middleware

Flask Extension

The entrolytics-flask-middleware package provides a Flask extension for automatic analytics tracking with decorators, middleware, and seamless Flask integration.

Installation

pip install entrolytics-flask-middleware

Quick Start

Basic Setup

from flask import Flask
from entrolytics_flask import Entrolytics

app = Flask(__name__)

# Initialize Entrolytics
entrolytics = Entrolytics(app,
    website_id='your-website-id',
    api_key='your-api-key'
)

@app.route('/')
def home():
    return 'Hello World!'

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

Factory Pattern

from flask import Flask
from entrolytics_flask import Entrolytics

def create_app():
    app = Flask(__name__)

    # Configure Entrolytics
    app.config['ENTROLYTICS_WEBSITE_ID'] = 'your-website-id'
    app.config['ENTROLYTICS_API_KEY'] = 'your-api-key'

    # Initialize extension
    entrolytics = Entrolytics(app)

    return app

app = create_app()

Decorator Usage

from flask import Flask
from entrolytics_flask import Entrolytics, track_event, track_page

app = Flask(__name__)
entrolytics = Entrolytics(app, website_id='your-website-id')

@app.route('/users/<int:user_id>')
@track_page('user_profile')
def user_profile(user_id):
    user = get_user(user_id)
    return render_template('profile.html', user=user)

@app.route('/api/users', methods=['POST'])
@track_event('user_creation_attempt')
def create_user():
    user_data = request.json

    try:
        user = create_user_in_db(user_data)

        # Track success event
        entrolytics.track('user_created', {
            'user_id': user.id,
            'email': user.email
        })

        return jsonify({'user_id': user.id}), 201
    except Exception as e:
        # Track error
        entrolytics.track('user_creation_failed', {
            'error': str(e),
            'email': user_data.get('email')
        })
        raise

Configuration

Extension Options

from flask import Flask
from entrolytics_flask import Entrolytics

app = Flask(__name__)

# Direct initialization
entrolytics = Entrolytics(app,
    website_id='your-website-id',    # required

    api_key='your-api-key',          # optional

    host='https://entrolytics.dev',  # custom host

    auto_track=True,                 # auto-track requests

    track_errors=True,               # track exceptions

    track_forms=True,                # track form submissions

    exclude_patterns=[r'/health', r'/static/.*'],  # exclude paths

    include_headers=['User-Agent', 'X-Forwarded-For'],  # include headers

    debug=False                      # debug mode

)

Configuration via Flask Config

# config.py
class Config:
    ENTROLYTICS_WEBSITE_ID = 'your-website-id'
    ENTROLYTICS_API_KEY = 'your-api-key'
    ENTROLYTICS_HOST = 'https://entrolytics.dev'
    ENTROLYTICS_AUTO_TRACK = True
    ENTROLYTICS_TRACK_ERRORS = True
    ENTROLYTICS_TRACK_FORMS = True
    ENTROLYTICS_EXCLUDE_PATTERNS = [r'/health', r'/static/.*']
    ENTROLYTICS_INCLUDE_HEADERS = ['User-Agent', 'X-Forwarded-For']
    ENTROLYTICS_DEBUG = False

class DevelopmentConfig(Config):
    ENTROLYTICS_DEBUG = True

class ProductionConfig(Config):
    ENTROLYTICS_DEBUG = False

Environment Variables

# .env
ENTROLYTICS_WEBSITE_ID=your-website-id
ENTROLYTICS_API_KEY=your-api-key
ENTROLYTICS_HOST=https://entrolytics.dev
ENTROLYTICS_AUTO_TRACK=true
ENTROLYTICS_TRACK_ERRORS=true
ENTROLYTICS_DEBUG=false

API Reference

Entrolytics Extension

class Entrolytics:
    def __init__(self, app=None, **kwargs)

    def init_app(self, app, **kwargs)

    def track(self, event, properties=None, user_id=None, session_id=None)

    def track_page(self, url=None, properties=None, user_id=None, session_id=None)

    def identify(self, user_id, traits=None)

    def set_user_properties(self, properties)

    def track_batch(self, events)

    def flush(self)

    def set_enabled(self, enabled)

    def is_enabled(self)

    def get_config(self)

Decorators

@track_event(event_name, properties=None)
def view_function(*args, **kwargs):
    pass

@track_page(page_name, properties=None)
def view_function(*args, **kwargs):
    pass

@identify_user(user_id_func=None, traits_func=None)
def view_function(*args, **kwargs):
    pass

@track_form(form_name=None, properties_func=None)
def view_function(*args, **kwargs):
    pass

Request Context

# In Flask routes
from flask import request

# Access entrolytics via request context
request.entrolytics.track('custom_event', {'property': 'value'})
request.entrolytics.identify('user-123', {'email': 'user@example.com'})
request.entrolytics.page('/custom-page', {'title': 'Custom Page'})

# Get current user/session info
user_id = request.entrolytics.get_user_id()
session_id = request.entrolytics.get_session_id()

Advanced Usage

Custom Middleware

from flask import Flask, request, g
from entrolytics_flask import Entrolytics

app = Flask(__name__)
entrolytics = Entrolytics(app)

@app.before_request
def track_request_info():
    # Track API version
    api_version = request.headers.get('API-Version')
    if api_version:
        request.entrolytics.track('api_request', {
            'version': api_version,
            'method': request.method,
            'endpoint': request.endpoint
        })

    # Track authenticated users
    if hasattr(g, 'user') and g.user:
        request.entrolytics.identify(g.user.id, {
            'email': g.user.email,
            'role': g.user.role
        })

@app.after_request
def track_response_info(response):
    # Track response status
    if response.status_code >= 400:
        request.entrolytics.track('http_error', {
            'status_code': response.status_code,
            'url': request.url,
            'method': request.method
        })

    return response

Form Tracking

from flask import Flask, request, render_template, redirect, url_for
from entrolytics_flask import Entrolytics, track_form

app = Flask(__name__)
entrolytics = Entrolytics(app, track_forms=True)

@app.route('/contact', methods=['GET', 'POST'])
@track_form('contact_form')
def contact():
    if request.method == 'POST':
        # Form submission is automatically tracked
        name = request.form.get('name')
        email = request.form.get('email')
        message = request.form.get('message')

        # Track custom form properties
        request.entrolytics.track('contact_submission', {
            'name_length': len(name),
            'has_message': len(message) > 0,
            'source': request.form.get('source', 'direct')
        })

        # Process form...
        return redirect(url_for('thank_you'))

    return render_template('contact.html')

# Custom form tracking
@app.route('/signup', methods=['GET', 'POST'])
def signup():
    if request.method == 'POST':
        # Track form with custom properties
        form_data = {
            'email': request.form.get('email'),
            'plan': request.form.get('plan', 'free'),
            'source': request.form.get('source')
        }

        request.entrolytics.track('signup_form_submission', form_data)

        # Process signup...
        return redirect(url_for('welcome'))

    return render_template('signup.html')

User Authentication Integration

from flask import Flask, session, request, g
from entrolytics_flask import Entrolytics, identify_user

app = Flask(__name__)
app.secret_key = 'your-secret-key'
entrolytics = Entrolytics(app)

@app.before_request
def load_user():
    if 'user_id' in session:
        g.user = get_user_by_id(session['user_id'])

@app.route('/login', methods=['POST'])
def login():
    username = request.form.get('username')
    password = request.form.get('password')

    user = authenticate_user(username, password)
    if user:
        session['user_id'] = user.id

        # Track login success
        request.entrolytics.identify(user.id, {
            'username': user.username,
            'email': user.email,
            'role': user.role
        })

        request.entrolytics.track('login_success', {
            'method': 'form',
            'username': username
        })

        return redirect(url_for('dashboard'))
    else:
        # Track login failure
        request.entrolytics.track('login_failed', {
            'username': username,
            'reason': 'invalid_credentials'
        })

        return render_template('login.html', error='Invalid credentials')

@app.route('/logout')
def logout():
    user_id = session.get('user_id')

    # Track logout
    request.entrolytics.track('logout', {
        'user_id': user_id,
        'method': 'explicit'
    })

    session.clear()
    return redirect(url_for('home'))

API Integration

from flask import Flask, request, jsonify
from entrolytics_flask import Entrolytics

app = Flask(__name__)
entrolytics = Entrolytics(app)

@app.route('/api/users', methods=['POST'])
def create_user():
    user_data = request.json

    # Track API call
    request.entrolytics.track('api_call', {
        'endpoint': '/api/users',
        'method': 'POST',
        'user_agent': request.headers.get('User-Agent'),
        'content_type': request.content_type
    })

    try:
        user = create_user_in_database(user_data)

        # Track success
        request.entrolytics.track('user_created', {
            'user_id': user.id,
            'email': user.email,
            'source': 'api'
        })

        return jsonify({
            'user_id': user.id,
            'email': user.email
        }), 201

    except ValidationError as e:
        # Track validation error
        request.entrolytics.track('validation_error', {
            'endpoint': '/api/users',
            'errors': e.errors,
            'invalid_fields': list(e.errors.keys())
        })

        return jsonify({'errors': e.errors}), 400

    except Exception as e:
        # Track server error
        request.entrolytics.track('server_error', {
            'endpoint': '/api/users',
            'error': str(e),
            'error_type': type(e).__name__
        })

        raise

# API middleware for authentication
@app.before_request
def api_auth():
    if request.path.startswith('/api/'):
        token = request.headers.get('Authorization', '').replace('Bearer ', '')

        if token:
            try:
                user = decode_jwt_token(token)
                g.user = user

                # Identify API user
                request.entrolytics.identify(user.id, {
                    'email': user.email,
                    'api_key': token[:10] + '...',
                    'permissions': user.permissions
                })

                request.entrolytics.track('api_authenticated', {
                    'endpoint': request.endpoint,
                    'method': request.method
                })

            except InvalidTokenError:
                request.entrolytics.track('api_auth_failed', {
                    'endpoint': request.endpoint,
                    'token': token[:10] + '...'
                })

Error Tracking

from flask import Flask, request
from entrolytics_flask import Entrolytics

app = Flask(__name__)
entrolytics = Entrolytics(app, track_errors=True)

@app.errorhandler(404)
def not_found(error):
    request.entrolytics.track('404_error', {
        'url': request.url,
        'method': request.method,
        'user_agent': request.headers.get('User-Agent'),
        'referrer': request.referrer
    })

    return render_template('404.html'), 404

@app.errorhandler(500)
def internal_error(error):
    request.entrolytics.track('500_error', {
        'url': request.url,
        'method': request.method,
        'error_message': str(error),
        'user_id': getattr(request, 'user_id', None)
    })

    return render_template('500.html'), 500

# Custom error tracking
@app.route('/risky-operation')
def risky_operation():
    try:
        result = perform_risky_operation()
        return result
    except SpecificError as e:
        request.entrolytics.track('specific_error', {
            'operation': 'risky_operation',
            'error_code': e.code,
            'error_message': e.message,
            'context': e.context
        })
        raise
    except Exception as e:
        request.entrolytics.track('unexpected_error', {
            'operation': 'risky_operation',
            'error_type': type(e).__name__,
            'error_message': str(e)
        })
        raise

Testing

Unit Tests

import pytest
from flask import Flask
from entrolytics_flask import Entrolytics
from unittest.mock import Mock, patch

@pytest.fixture
def app():
    app = Flask(__name__)
    app.config['TESTING'] = True
    app.config['ENTROLYTICS_WEBSITE_ID'] = 'test-website-id'

    with patch('entrolytics_flask.client.EntrolyticsClient') as mock_client:
        mock_instance = Mock()
        mock_client.return_value = mock_instance

        entrolytics = Entrolytics(app)
        entrolytics_mock = mock_instance

        yield app

@pytest.fixture
def client(app):
    return app.test_client()

def test_basic_tracking(app, client):
    with app.test_request_context('/'):
        entrolytics_mock.track.assert_not_called()

        # Make request
        response = client.get('/')

        # Check if automatic tracking was called
        entrolytics_mock.track.assert_called()

def test_decorator_tracking(app, client):
    @app.route('/test')
    def test_route():
        return 'test'

    with app.test_request_context('/test'):
        from entrolytics_flask import track_event

        @track_event('test_event')
        def decorated_view():
            return 'test'

        # Test decorator
        response = client.get('/test')
        entrolytics_mock.track.assert_called_with('test_event', None, None, None)

Integration Tests

import pytest
from flask import Flask
from entrolytics_flask import Entrolytics

def test_full_integration():
    app = Flask(__name__)
    app.config['ENTROLYTICS_WEBSITE_ID'] = 'integration-test-id'

    # Use real client for integration testing
    entrolytics = Entrolytics(app, debug=True)

    @app.route('/test')
    def test_route():
        request.entrolytics.track('integration_test', {'test': True})
        return 'test'

    with app.test_client() as client:
        response = client.get('/test')

        # Verify response
        assert response.status_code == 200
        assert response.data == b'test'

        # In real integration, you would verify
        # that events were actually sent to the API

Performance Optimization

Async Operations

import asyncio
from concurrent.futures import ThreadPoolExecutor
from flask import Flask
from entrolytics_flask import Entrolytics

app = Flask(__name__)
entrolytics = Entrolytics(app)

# Background thread for async operations
executor = ThreadPoolExecutor(max_workers=2)

def track_background(event, properties):
    """Track events in background thread"""
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)

    try:
        # Use async client in background
        client = get_async_client()
        loop.run_until_complete(client.track(event, properties))
    finally:
        loop.close()

@app.route('/heavy-operation')
def heavy_operation():
    # Track event without blocking
    executor.submit(track_background, 'heavy_operation_start', {
        'timestamp': time.time()
    })

    # Perform heavy operation
    result = perform_heavy_operation()

    # Track completion
    executor.submit(track_background, 'heavy_operation_complete', {
        'result_size': len(result)
    })

    return result

Batching

from flask import Flask
from entrolytics_flask import Entrolytics
import threading
import time

app = Flask(__name__)
entrolytics = Entrolytics(app)

# Global event batch
event_batch = []
batch_lock = threading.Lock()

def flush_batch():
    """Flush events in background"""
    while True:
        time.sleep(30)  # Flush every 30 seconds

        with batch_lock:
            if event_batch:
                # Send batch
                entrolytics.track_batch(event_batch.copy())
                event_batch.clear()

# Start background thread
batch_thread = threading.Thread(target=flush_batch, daemon=True)
batch_thread.start()

@app.route('/track-batch')
def track_batch_endpoint():
    with batch_lock:
        event_batch.append({
            'event': 'batch_test',
            'properties': {'timestamp': time.time()}
        })

    return 'Event added to batch'

Troubleshooting

Best Practices

Migration Guide

From Manual Analytics

# Before - Manual tracking
import analytics

@app.route('/old-way')
def old_way():
    analytics.track('page_view', {'url': '/old-way'})
    return 'old'

# After - Flask extension
from entrolytics_flask import Entrolytics, track_page

app = Flask(__name__)
entrolytics = Entrolytics(app)

@app.route('/new-way')
@track_page('new_page')
def new_way():
    return 'new'

From Other Flask Analytics

# Before - Other analytics
from flask_analytics import Analytics

app = Flask(__name__)
analytics = Analytics(app)

# After - Entrolytics
from entrolytics_flask import Entrolytics

app = Flask(__name__)
entrolytics = Entrolytics(app)

Flask extension for Entrolytics - First-party growth analytics for the edge

Edit on GitHub

On this page

Flask ExtensionInstallationQuick StartBasic SetupFactory PatternDecorator UsageConfigurationExtension OptionsConfiguration via Flask ConfigEnvironment VariablesAPI ReferenceEntrolytics ExtensionDecoratorsRequest ContextAdvanced UsageCustom MiddlewareForm TrackingUser Authentication IntegrationAPI IntegrationError TrackingTestingUnit TestsIntegration TestsPerformance OptimizationAsync OperationsBatchingTroubleshootingBest PracticesMigration GuideFrom Manual AnalyticsFrom Other Flask Analytics