UVF IT: PostHog Desktop

Welcome to another official UVF IT build guide. Today we are constructing a minimal, self-hosted analytics engine inspired by PostHog, designed to help you understand user behavior without sending data to third-party servers.

Using Python and SQLite, you will build a lightweight event tracker that captures page views and custom interactions. This project focuses on the core mechanics of event ingestion, storage, and visualization, providing a foundational understanding of how modern product analytics tools operate under the hood.

Prerequisites & Constraints

Requires Python 3.9+ and basic familiarity with the command line; ensure pip is installed and accessible in your PATH.
This is a simplified clone: it lacks user authentication, complex funnel analysis, and real-time WebSocket updates found in the full PostHog suite.
SQLite is used for simplicity; it is not suitable for high-concurrency production workloads. Do not expose this endpoint to the public internet without a reverse proxy and authentication.

System Architecture

Phase 1: Scaffold Project & Database Schema

Initialize the Python environment and create the SQLite database schema. We will define a simple events table to store the event name, timestamp, and arbitrary properties as JSON.

Install Flask for the web server and sqlite3 is built-in. Create a schema.sql file to define the table structure, ensuring we can handle flexible event properties.

setup.sh

#!/bin/bash
mkdir -p data
pip install flask
python3 -c "
import sqlite3
import os

# Ensure data directory exists
os.makedirs('data', exist_ok=True)

# Connect to SQLite database
db_path = 'data/analytics.db'
conn = sqlite3.connect(db_path)
cursor = conn.cursor()

# Create events table
cursor.execute('''
CREATE TABLE IF NOT EXISTS events (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    event_name TEXT NOT NULL,
    timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
    properties TEXT DEFAULT '{}'
)
''')

conn.commit()
conn.close()
print('Database schema created successfully.')
"

schema.sql

CREATE TABLE IF NOT EXISTS events (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    event_name TEXT NOT NULL,
    timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
    properties TEXT DEFAULT '{}'
);

Phase 2: Core Feature 1 - Event Ingestion API

Build the backend logic to receive events. Create a app.py file that initializes the Flask app and connects to SQLite.

Implement a /capture endpoint that accepts POST requests with JSON payloads. This endpoint will validate the input and insert the event into the database.

app.py

from flask import Flask, request, jsonify
import sqlite3
import json

app = Flask(__name__)

DB_PATH = 'data/analytics.db'

def get_db():
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    return conn

@app.route('/capture', methods=['POST'])
def capture():
    data = request.get_json()
    if not data or 'event' not in data:
        return jsonify({'status': 'error', 'message': 'Missing event field'}), 400
    
    event_name = data['event']
    properties = json.dumps(data.get('properties', {}))
    
    try:
        conn = get_db()
        cursor = conn.cursor()
        cursor.execute(
            'INSERT INTO events (event_name, properties) VALUES (?, ?)',
            (event_name, properties)
        )
        conn.commit()
        conn.close()
        return jsonify({'status': 'success', 'id': cursor.lastrowid}), 201
    except Exception as e:
        return jsonify({'status': 'error', 'message': str(e)}), 500

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

Phase 3: Core Feature 2 - Data Retrieval & Dashboard

Create an endpoint to retrieve recent events for visualization. We will add a /events endpoint that returns the last 100 events.

Build a simple HTML dashboard that fetches these events and displays them in a table. This provides immediate visual feedback for the analytics data.

app.py

from flask import Flask, request, jsonify
import sqlite3
import json

app = Flask(__name__)

DB_PATH = 'data/analytics.db'

def get_db():
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    return conn

@app.route('/capture', methods=['POST'])
def capture():
    data = request.get_json()
    if not data or 'event' not in data:
        return jsonify({'status': 'error', 'message': 'Missing event field'}), 400
    
    event_name = data['event']
    properties = json.dumps(data.get('properties', {}))
    
    try:
        conn = get_db()
        cursor = conn.cursor()
        cursor.execute(
            'INSERT INTO events (event_name, properties) VALUES (?, ?)',
            (event_name, properties)
        )
        conn.commit()
        conn.close()
        return jsonify({'status': 'success', 'id': cursor.lastrowid}), 201
    except Exception as e:
        return jsonify({'status': 'error', 'message': str(e)}), 500

@app.route('/events', methods=['GET'])
def get_events():
    try:
        conn = get_db()
        cursor = conn.cursor()
        cursor.execute('SELECT * FROM events ORDER BY timestamp DESC LIMIT 100')
        rows = cursor.fetchall()
        events = []
        for row in rows:
            events.append({
                'id': row['id'],
                'event_name': row['event_name'],
                'timestamp': row['timestamp'],
                'properties': json.loads(row['properties'])
            })
        conn.close()
        return jsonify(events)
    except Exception as e:
        return jsonify({'status': 'error', 'message': str(e)}), 500

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

dashboard.html

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>PostHog Clone Dashboard</title>
    <style>
        body { font-family: Arial, sans-serif; margin: 20px; }
        table { width: 100%; border-collapse: collapse; margin-top: 20px; }
        th, td { border: 1px solid #ddd; padding: 8px; text-align: left; }
        th { background-color: #f2f2f2; }
        .refresh-btn { padding: 10px 20px; background-color: #007bff; color: white; border: none; cursor: pointer; }
        .refresh-btn:hover { background-color: #0056b3; }
    </style>
</head>
<body>
    <h1>PostHog Clone Dashboard</h1>
    <button class="refresh-btn" onclick="loadEvents()">Refresh Events</button>
    <table id="events-table">
        <thead>
            <tr>
                <th>ID</th>
                <th>Event Name</th>
                <th>Timestamp</th>
                <th>Properties</th>
            </tr>
        </thead>
        <tbody id="events-body">
        </tbody>
    </table>

    <script>
        async function loadEvents() {
            try {
                const response = await fetch('/events');
                const events = await response.json();
                const tbody = document.getElementById('events-body');
                tbody.innerHTML = '';
                events.forEach(event => {
                    const row = document.createElement('tr');
                    row.innerHTML = `
                        <td>${event.id}</td>
                        <td>${event.event_name}</td>
                        <td>${event.timestamp}</td>
                        <td>${JSON.stringify(event.properties)}</td>
                    `;
                    tbody.appendChild(row);
                });
            } catch (error) {
                console.error('Error loading events:', error);
            }
        }

        // Load events on page load
        document.addEventListener('DOMContentLoaded', loadEvents);
    </script>
</body>
</html>

Phase 4: Run Locally & Simulate Traffic

Now that our API and dashboard are ready, we need to bring the system to life. We will start the Flask development server and then simulate user activity by sending raw HTTP requests to our ingestion endpoint.

This step verifies that the data pipeline is intact: the browser sends a POST request, Flask processes it, and SQLite persists the record. We use curl here because it gives us precise control over the payload structure, mimicking how a real client-side script would behave.

run.sh

#!/bin/bash

# Start the Flask server in the background
# We use nohup to keep it running while we execute curl commands
nohup python app.py > server.log 2>&1 &
SERVER_PID=$!

# Give the server a moment to start
sleep 2

echo "Server started with PID: $SERVER_PID"

# Simulate a 'page_view' event
curl -s -X POST http://localhost:5000/capture \
  -H "Content-Type: application/json" \
  -d '{"event": "page_view", "properties": {"page": "/home", "referrer": "google.com"}}'

echo ""

# Simulate a 'button_click' event
curl -s -X POST http://localhost:5000/capture \
  -H "Content-Type: application/json" \
  -d '{"event": "button_click", "properties": {"button_name": "signup", "location": "footer"}}'

echo ""

# Simulate a 'form_submit' event
curl -s -X POST http://localhost:5000/capture \
  -H "Content-Type: application/json" \
  -d '{"event": "form_submit", "properties": {"form_id": "contact_us", "fields_filled": 3}}'

echo ""
echo "Test events sent. Check http://localhost:5000/dashboard.html to verify."

# Kill the server when done (optional, for script cleanup)
# kill $SERVER_PID

Phase 5: Verify & Extend

With the server running and test events in the database, open your browser to http://localhost:5000/dashboard.html. You should see a table populated with the 'page_view', 'button_click', and 'form_submit' events you just triggered via curl.

To make this truly useful like PostHog, we need a client-side snippet. This JavaScript code allows any HTML page to automatically send events to our local API without manual curl commands. We will create a simple fetch-based tracker that mimics the standard analytics SDK pattern.

snippet.js

/**
 * UVF IT Analytics Snippet
 * Drop this script into your HTML <head> to start tracking.
 * Configure the API endpoint to match your server.
 */

(function() {
  const CONFIG = {
    apiEndpoint: 'http://localhost:5000/capture',
    projectId: 'uvf-it-demo'
  };

  // Core capture function
  window.uvfTrack = function(eventName, properties = {}) {
    const payload = {
      event: eventName,
      properties: {
        ...properties,
        timestamp: new Date().toISOString(),
        project_id: CONFIG.projectId
      }
    };

    // Use fetch to send data to our Flask API
    fetch(CONFIG.apiEndpoint, {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json'
      },
      body: JSON.stringify(payload)
    })
    .then(response => {
      if (!response.ok) throw new Error('Network response was not ok');
      console.log(`[UVF IT] Event captured: ${eventName}`);
    })
    .catch(error => {
      console.error(`[UVF IT] Failed to capture event: ${eventName}`, error);
    });
  };

  // Auto-track page views
  window.uvfTrack('page_view', {
    page: window.location.pathname,
    title: document.title,
    url: window.location.href
  });

  // Auto-track clicks on elements with data-uvf-track attribute
  document.addEventListener('click', function(e) {
    const target = e.target.closest('[data-uvf-track]');
    if (target) {
      const eventName = target.getAttribute('data-uvf-track');
      window.uvfTrack(eventName, {
        element_tag: target.tagName,
        element_text: target.innerText.substring(0, 50)
      });
    }
  });

  console.log('[UVF IT] Analytics snippet loaded.');
})();

Verifying the Stack

Check	Command / Why it matters
Server is running and accepting connections	curl -s http://localhost:5000/ \| grep -q 'Flask' && echo 'Server OK'
Events are persisted in SQLite	sqlite3 data/analytics.db 'SELECT count(*) FROM events;' \| grep -q '[1-9]' && echo 'Data OK'
Dashboard renders recent events	Open http://localhost:5000/dashboard.html and visually confirm the table is not empty
Client-side snippet triggers events	Load a test HTML page with snippet.js, then refresh dashboard to see a new 'page_view' event

Next Steps

You have built a functional analytics tracker. To extend this, consider adding user identification, session tracking, and a more robust frontend framework for the dashboard. Remember to secure your API endpoints before deploying to production.