Posttrail

Posttrail is a Django history table implementation with automatic versioning. It provides automatic versioning where updates create new rows instead of modifying existing ones, while maintaining full audit trails and supporting real-time streaming.

Features

Core Versioning

• Auto-incrementing ID versioning: History records use auto-incrementing integer IDs for simple, efficient ordering
• Automatic version creation: Updates and deletes automatically create new versions
• ForeignKey linking: History records link to parent records via ForeignKey (object_id) for efficient queries
• Soft deletes: Deletions create new versions with deleted=True
• Timestamp-based ordering: History records include created_at timestamps for chronological ordering
• Filtering support: Query history records with Django filter syntax via the Shape API

History Tracking

• SQL trigger-based tracking: Database-level tracking via @track_history_trigger (catches all changes, including raw SQL)
• Sparse history: Only stores fields that changed, reducing storage

Real-time Streaming

• Durable Streams API: HTTP-based protocol for streaming incremental updates
• Server-Sent Events (SSE): Real-time live streaming support
• Long-polling: Efficient polling for new changes
• Offset-based pagination: Resume from any point in history using integer ID offsets
• Filtering: Filter streams by field values (e.g., where={"status_id": 5})

Additional Features

• Shape API: Flexible querying with filtering, column selection, and pagination
• Window functions: Advanced SQL capabilities for analytics
• Functional indexes: Optimized database queries

Requirements

• Python 3.8+
• PostgreSQL 13+
• Django 4.2
• dbsamizdat (for managing Postgres functions and triggers)
• UV (Python package manager and virtual environment tool)

Quick Start

1. Install UV (if not already installed)

# On macOS/Linux
curl -LsSf https://astral.sh/uv/install.sh | sh

# On Windows
powershell -c "irm https://astral.sh/uv/install.ps1 | iex"

# Or via pip
pip install uv

2. Setup Database

Using Podman (recommended, see Justfile):

just start

Or manually:

podman run -d --name posttrail-db -p 5434:5432 -e POSTGRES_PASSWORD=password docker.io/library/postgres:14

Or use Docker:

docker run -d --name posttrail-db -p 5434:5432 -e POSTGRES_PASSWORD=password postgres:14

3. Install Dependencies

# Clone the repository
git clone https://github.com/joshbrooks/posttrail.git
cd posttrail

# Install dependencies (UV automatically creates .venv/)
uv sync

4. Configure Database

Update posttrail_project/settings.py with your database credentials:

DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.postgresql',
        'NAME': 'postgres',
        'USER': 'postgres',
        'PASSWORD': 'password',
        'HOST': 'localhost',
        'PORT': '5434',
    }
}

5. Run Migrations

uv run python manage.py migrate

6. Use in Your Models

from posttrail.decoupled import track_history_trigger
from django.db import models

@track_history_trigger()
class Article(models.Model):
    # Standard auto-incrementing integer primary key (or any PK type)
    title = models.CharField(max_length=200)
    body = models.TextField()

# Configure dbsamizdat in settings.py
DBSAMIZDAT_MODULES = [
    'posttrail.decoupled.triggers',
]

# Deploy triggers
# uv run python manage.py shell
from dbsamizdat.django_api import sync
sync(dbconn="default", samizdatmodules=('posttrail.decoupled.triggers',))

Note: Trigger-based tracking captures ALL database changes, including raw SQL updates, admin changes, and bulk operations.

7. Query History

# Query current state
current = Article.objects.all()  # Only current versions

# Query history
all_versions = Article.history.all()  # All versions
versions_of_entity = Article.history.versions_of_object(article.id)

# Soft delete
article.delete()  # Creates version with deleted=True

Architecture

History Table Structure

History tables use a simple structure:

• id: Auto-incrementing integer primary key (provides chronological ordering)
• object_id: ForeignKey to the parent model instance (links to original record's PK)
• created_at: Timestamp when the history record was created
• deleted: Boolean flag indicating if this is a deletion record
• changed_fields: JSON array of field names that changed
• All model fields: Same fields as parent model (nullable, sparse storage)

Version Creation Flow

When you update a record:

1. Django detects it's an update (has PK, not in adding state)
2. Compares old vs new field values
3. Creates new history record with:
   • Auto-generated integer ID (provides chronological ordering)
   • object_id → ForeignKey to parent record's PK
   • created_at → current timestamp
   • Only changed fields populated (sparse storage)
   • changed_fields → JSON array of changed field names
4. Original record remains unchanged

Querying

• Current state: Model.objects.all() returns current versions
• All versions: Model.history.all() returns all history records
• Version chain: Model.history.versions_of_object(pk) gets all versions for a specific object
• Incremental sync: Model.history.changes_since(timestamp) filters by created_at timestamp
• Filtering: Use Django filter syntax via Shape API where parameter

Project Structure

posttrail/
├── posttrail/
│   ├── decoupled/          # History tracking implementation
│   │   ├── trigger_decorator.py  # SQL trigger-based @track_history_trigger
│   │   ├── sql_builders.py       # SQL generation for triggers
│   │   ├── sql_identifier.py    # Safe SQL identifier handling
│   │   ├── triggers.py           # dbsamizdat trigger discovery
│   │   ├── history_manager.py    # History query manager
│   │   ├── change_tracker.py     # Field change detection utilities
│   │   └── models.py             # HistoryLink model
│   ├── durable_streams/    # Real-time streaming API
│   ├── api/                # Shape API for flexible querying
├── example/                # Example Django app
│   ├── models.py           # Example models
│   └── migrations/         # Database migrations
├── tests/                   # Test suite
│   ├── test_trigger_history.py  # Trigger-based tests
│   ├── test_sql_builders.py    # SQL builder tests
│   ├── test_sql_identifier.py  # Identifier validation tests
│   ├── test_parsers.py         # API parser tests
│   └── test_change_tracker.py   # Change tracking tests
├── doc/                     # Additional documentation
└── README.md               # This file

Development

Running Tests

Posttrail uses pytest with fixtures (not Django's TestCase) for testing.

# Run all tests
uv run pytest

# Run only unit tests (fast, no database)
uv run pytest -m unit

# Run integration tests (require database)
uv run pytest -m integration

# Run specific test file
uv run pytest tests/test_trigger_history.py

# Run with verbose output
uv run pytest -v

Prerequisites:

• PostgreSQL 18 running (see Setup Database above)
• Test database will be created automatically by pytest-django

See tests/TESTING.md for detailed testing guide.

Running the Example App

# Deploy triggers
uv run python manage.py shell
>>> from dbsamizdat.django_api import sync
>>> sync(dbconn="default", samizdatmodules=('posttrail.decoupled.triggers',))

# Run migrations
uv run python manage.py migrate

# Create superuser (optional)
uv run python manage.py createsuperuser

# Run development server
uv run python manage.py runserver

Then visit:

• http://localhost:8000/ - Homepage
• http://localhost:8000/articles/ - Article list
• http://localhost:8000/admin/ - Django admin

See doc/example-app.md for detailed instructions.

Documentation

• Tutorial - Step-by-step guide to getting started with history tracking
• Example App Guide - Running and using the example application
• Filtering Guide - Filtering history records with Django filter syntax
• Durable Streams Specification - HTTP-based streaming protocol
• Streaming Guide - Real-time streaming with SSE
• Trigger-based History - SQL trigger implementation
• Implementation Details - Architecture and design decisions
• Testing Guide - Comprehensive testing documentation

API Reference

Durable Streams API

Stream incremental updates via HTTP:

# Catch-up read
GET /v1/stream/{app_label}/{model_name}?offset=-1

# Long-poll for new data
GET /v1/stream/{app_label}/{model_name}?offset={offset}&live=long-poll

# Server-Sent Events
GET /v1/stream/{app_label}/{model_name}?offset={offset}&live=sse

See doc/durable-streams-specification.md for full API documentation.

Shape API

Flexible querying with filtering and column selection:

GET /v1/shape?table={app_label}.{model_name}&offset={id}&limit=100&where={...}&columns=[...]

Parameters:

• table: Table name (e.g., example.tutorial_product_history)
• offset: Integer ID offset (or "-1" for start of stream)
• limit: Maximum number of records (default: 100)
• where: JSON string with Django filter syntax (e.g., {"deleted": false} or {"status_id": 5})
• columns: JSON array of column names to select (optional)

Filtering Examples:

# Filter by deleted status
GET /v1/shape?table=example.tutorial_product_history&where={"deleted":false}

# Filter by foreign key
GET /v1/shape?table=example.task_history&where={"status_id":5}

# Multiple conditions (AND by default)
GET /v1/shape?table=example.task_history&where={"status_id":5,"deleted":false}

How It Works

Single Table Design

All versions are stored in a separate history table, linked via ForeignKey:

• History records: Each history record has an auto-incrementing integer ID and links to the parent record via object_id ForeignKey
• HistoryManager: Returns all history versions with methods for version traversal and filtering

Database Structure

History tables use standard Django ForeignKeys and auto-incrementing IDs:

• id: Auto-incrementing integer primary key (provides chronological ordering)
• object_id: ForeignKey to the original model instance
• Standard PostgreSQL indexes on object_id and created_at for efficient queries
• No special database functions required for basic versioning

Indexes

Fast queries for history records:

CREATE INDEX model_history_object_id_idx ON model_history(object_id);
CREATE INDEX model_history_created_at_idx ON model_history(created_at);

These indexes are automatically created by Django migrations.

License

MIT

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

Support

For issues, questions, or contributions, please open an issue on GitHub.