sqlalchemy-activitylog 0.1.0

A comprehensive activity logging package for SQLAlchemy 2.0, inspired by Laravel Spatie Activity Log

sqlalchemy-activitylog

A comprehensive activity logging package for SQLAlchemy 2.0, inspired by Laravel Spatie Activity Log but adapted for the Python ecosystem.

Track model changes automatically, log manual activities, query audit trails, and maintain complete change history with a clean, typed API.

Features

• ✅ Automatic logging of model create/update/delete events
• ✅ Manual activity logging with a fluent builder API
• ✅ Dirty tracking - log only changed attributes
• ✅ Change history - detailed before/after values in JSON
• ✅ Subject & Causer support - track what changed and who changed it
• ✅ Batch operations - group activities with shared batch UUID
• ✅ Request metadata - automatically capture request context
• ✅ Query API - fluent helpers for searching activities
• ✅ Soft delete support - pluggable soft delete strategies
• ✅ FastAPI integration - automatic middleware and user extraction
• ✅ Flask integration - request context integration
• ✅ PostgreSQL, MySQL, SQLite support
• ✅ Fully typed with Python 3.11+ syntax
• ✅ Production-ready with comprehensive test coverage

Installation

pip install sqlalchemy-activitylog

With FastAPI support:

pip install sqlalchemy-activitylog[fastapi]

With Flask support:

pip install sqlalchemy-activitylog[flask]

With CLI tools:

pip install sqlalchemy-activitylog[cli]

With PostgreSQL support:

pip install sqlalchemy-activitylog[postgres]

Quick Start

1. Define Your Models

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import declarative_base
from sqlalchemy_activitylog import LogsActivity, LogOptions

Base = declarative_base()


class User(Base, LogsActivity):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    name = Column(String)
    email = Column(String)

    __activitylog__ = (
        LogOptions.defaults()
        .use_log_name("users")
        .log_only(["name", "email"])
        .log_only_dirty()
        .dont_submit_empty_logs()
        .set_description(lambda event, subject, causer: f"User {event}")
    )

2. Create Activity Table

Option A: Using Alembic (Recommended)

Initialize the migration file using the Spatie-style CLI tool. It will automatically detect your project's current migration head and append the activities table creation seamlessly:

# 1. Publish the migration template into your alembic directory
activitylog init

# 2. Run the migration to apply changes to your database
activitylog migrate

Note: You can also use the standard alembic upgrade head command after running activitylog init if you prefer.

Option B: Manual Creation (For projects not using Alembic)

If you do not use a migration manager, you can import the model and create the table directly through SQLAlchemy's metadata engine:

from my_project.database import engine, Base
# Importing the model registers it with your local Base metadata
from sqlalchemy_activitylog.models import Activity

# Create the activities table immediately
Base.metadata.create_all(engine)

3. Initialize the Logger

from sqlalchemy_activitylog import ActivityLogger, Activity, set_activity_logger

# Create logger
logger = ActivityLogger(Activity)
logger.set_causer_resolver(lambda: current_user)
logger.register_events()  # Register SQLAlchemy event listeners

# Set global logger
set_activity_logger(logger)

4. Use It!

Automatic logging:

user = User(name="John Doe", email="[e-mail]")
db.add(user)
db.commit()  # Automatically logs "create" event

Manual logging:

from sqlalchemy_activitylog import activity

activity()
    .event("published")
    .performed_on(post)
    .caused_by(current_user)
    .with_properties({"status": "published"})
    .log("Post published to production", session=db)

Query activities:

# Get all activities for a user
activities = db.query(Activity).filter_by(subject_type="User", subject_id=user.id).all()

# Get latest activities
latest = db.query(Activity).order_by(Activity.created_at.desc()).limit(10).all()

# Get activities in a batch
batch_activities = db.query(Activity).filter_by(batch_uuid="...").all()

View changes:

activity = db.query(Activity).filter_by(event="update").first()

# Get formatted changes
changes = activity.changes
# {
#     "name": {"old": "John", "new": "Johnny"},
#     "email": {"old": "[e-mail]", "new": "[e-mail]"}
# }

Core Concepts

Activity Model

The Activity model stores audit trail records:

class Activity:
    id: int                    # Primary key
    log_name: str             # Channel name (e.g., "users", "posts")
    event: str                # Event type (create, update, delete, custom)
    description: str          # Human-readable description
    subject_type: str         # Type of modified model
    subject_id: int           # ID of modified model
    causer_type: str          # Type of user who made change
    causer_id: int            # ID of user who made change
    properties: dict          # JSON change tracking
    batch_uuid: str           # Batch operation ID
    created_at: datetime      # Timestamp

Properties Structure

Changes are stored in a single JSON column named properties:

Create event:

{
  "attributes": {
    "name": "John Doe",
    "email": "[e-mail]"
  }
}

Update event:

{
  "attributes": {
    "name": "Johnny",
    "email": "[e-mail]"
  },
  "old": {
    "name": "John Doe",
    "email": "[e-mail]"
  }
}

Delete event:

{
  "old": {
    "name": "John Doe",
    "email": "[e-mail]"
  }
}

LogOptions Builder

Configure logging behavior per model:

__activitylog__ = (
    LogOptions.defaults()

    # Set log channel
    .use_log_name("users")

    # Configure attribute tracking
    .log_all()                              # Log all attributes (default)
    .log_only(["name", "email"])            # Log only these
    .log_except(["password", "token"])      # Log all except these
    .exclude_attributes(["updated_at"])     # Alias for log_except

    # Dirty tracking
    .log_only_dirty()                       # Only log changed attributes

    # Empty log handling
    .dont_submit_empty_logs()               # Don't log if nothing changed

    # Attribute filters
    .dont_log_if_attributes_changed_only(["updated_at"])  # Ignore timestamp-only changes

    # Descriptions
    .set_description(
        lambda event, subject, causer: f"User {subject.name} was {event}"
    )
)

Manual Logging

Log custom activities with the fluent builder:

from sqlalchemy_activitylog import activity

activity()
    .event("published")                          # Custom event name
    .performed_on(post)                          # What was affected
    .caused_by(current_user)                     # Who did it
    .with_properties({
        "ip": request.client.host,
        "status": "published"
    })                                           # Custom metadata
    .use_log("posts")                            # Log channel
    .tap(lambda a: setattr(a, "custom", "value"))  # Modify before save
    .log("Post published to production", session=db)  # Description + save

Batch Operations

Group related activities:

from sqlalchemy_activitylog import activity_batch

with activity_batch():
    # Create multiple records
    db.add(User(name="Alice"))
    db.add(User(name="Bob"))
    db.commit()

    # All activities share the same batch_uuid

Integration Guides

FastAPI

from fastapi import FastAPI
from sqlalchemy_activitylog.integrations.fastapi import setup_activitylog
from sqlalchemy_activitylog import Activity

app = FastAPI()

async def get_current_user():
    # Your user extraction logic
    pass

# Setup activity logging
logger = await setup_activitylog(
    app,
    Activity,
    causer_resolver=get_current_user,
    enable_middleware=True,  # Capture request metadata
)

Flask

from flask import Flask
from sqlalchemy_activitylog.integrations.flask import setup_activitylog
from sqlalchemy_activitylog import Activity
from flask_login import current_user

app = Flask(__name__)

# Setup activity logging
logger = setup_activitylog(
    app,
    Activity,
    causer_resolver=lambda: current_user,
    enable_middleware=True,
)

Configuration

from sqlalchemy_activitylog import ActivityLogConfig, set_config

config = ActivityLogConfig(
    enabled=True,
    table_name="activities",
    database_schema=None,
    default_log_name="default",
    submit_empty_logs=False,
    store_request_metadata=True,
)

set_config(config)

Advanced Usage

Soft Delete Support

from sqlalchemy_activitylog import SoftDeleteMixin

class User(Base, LogsActivity, SoftDeleteMixin):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    name = Column(String)
    deleted_at = Column(DateTime, nullable=True)

    __activitylog__ = LogOptions.defaults().use_log_name("users")

# Soft delete
user.soft_delete()

# Restore
user.restore()

Custom Activity Model

from sqlalchemy_activitylog import create_activity_model

CustomActivity = create_activity_model(
    table_name="audit_logs",
    schema="public"
)

Request Metadata

Automatically capture request context:

# In FastAPI/Flask, request metadata is auto-captured:
# - request id
# - IP address
# - user agent
# - URL
# - HTTP method

# Access in activity.properties:
activity.properties["request"]["ip"]
activity.properties["request"]["user_agent"]

API Reference

See API.md for complete API documentation.

Testing

Run tests with pytest:

pytest tests/

With coverage:

pytest tests/ --cov=sqlalchemy_activitylog --cov-report=html

License

MIT License - see LICENSE file for details.

Contributing

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

Inspiration

This package is inspired by Laravel Spatie Activity Log and adapted for SQLAlchemy 2.0 and the Python ecosystem.