waudit-tools 0.1.1

Tools for waudit: SQL generation, Alembic ops, SQLAlchemy mixins

waudit-tools

Utilities for CDC-based audit log systems. SQL generation, custom Alembic operations, and SQLAlchemy mixins for PostgreSQL + Debezium pipelines.

Installation

With pip:

pip install waudit-tools

With uv:

uv pip install waudit-tools

Requires Python 3.11+.

---

Quick Start

Alembic migration (recommended)

Create an empty migration:

alembic revision -m "setup waudit"

upgrade:

from alembic import op
from myproject.models import Base

def upgrade() -> None:
    op.waudit_setup_from_base(
        Base,
        exclude={"alembic_version"},
        create_heartbeat_table=True,
    )

downgrade:

def downgrade() -> None:
    op.waudit_remove_from_base(
        Base,
        exclude={"alembic_version"},
        drop_heartbeat_table=True,
    )

---

Use Cases

1. Manual table list

When you don't use a declarative base or want full control:

def upgrade() -> None:
    op.waudit_setup(
        include_audit_tables=["users", "articles", "comments"],
        exclude_audit_tables=["alembic_version", "spatial_ref_sys"],
        create_heartbeat_table=True,
        schema="public",
    )

def downgrade() -> None:
    op.waudit_remove(
        include_audit_tables=["users", "articles", "comments"],
        exclude_audit_tables=["alembic_version"],
        drop_heartbeat_table=True,
        schema="public",
    )

2. Single table

For targeted changes or incremental rollout:

def upgrade() -> None:
    op.waudit_setup_table("users", schema="public")

def downgrade() -> None:
    op.waudit_remove_table("users", schema="public")

3. Raw SQL generation (no Alembic)

Generate SQL strings for external tooling or manual execution:

from waudit_tools.sql.identity import generate_identity_sql, apply_identity_to_tables
from waudit_tools.sql.heartbeat import generate_heartbeat_sql, generate_drop_heartbeat_sql

# Single table
sql = generate_identity_sql("users", schema="public")
# ALTER TABLE public.users REPLICA IDENTITY FULL;

# Batch
batch = apply_identity_to_tables(["users", "articles", "comments"], schema="cms")

# Heartbeat
create_hb = generate_heartbeat_sql(schema="public")
drop_hb = generate_drop_heartbeat_sql(schema="public")

4. SQLAlchemy model mixins

Add last_modified_by to your models:

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from waudit_tools.sqlalchemy.mixins.str_mixin import WAuditStrMixin

class Base(DeclarativeBase):
    pass

class User(Base, WAuditStrMixin):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    # last_modified_by: str | None  -- injected by mixin

For UUID-based actor tracking:

from waudit_tools.sqlalchemy.mixins.uuid_mixin import WAuditUUIDMixin

class User(Base, WAuditUUIDMixin):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    # last_modified_by: UUID | None  -- injected by mixin

5. Programmatic Alembic utils

If you build custom migration frameworks on top of Alembic:

from waudit_tools.alembic.utils import create_heartbeat, drop_heartbeat
from sqlalchemy.engine import Connection

def my_custom_setup(conn: Connection) -> None:
    create_heartbeat(conn, schema="public")
    # ... your logic

---

Why REPLICA IDENTITY FULL?

Debezium needs "before" values for UPDATE and DELETE events. PostgreSQL only sends the changed columns by default. REPLICA IDENTITY FULL forces Postgres to include the complete old row in the WAL, enabling full change-data-capture audit trails.

---

Architecture

waudit_tools/
├── alembic/
│   ├── setup.py      # WauditSetupOp, WauditSetupFromBaseOp, WauditSetupTableOp
│   ├── remove.py     # WauditRemoveOp, WauditRemoveFromBaseOp, WauditRemoveTableOp
│   ├── utils.py      # create_heartbeat, drop_heartbeat
│   └── __init__.py   # side-effect registration
├── sql/
│   ├── identity.py   # generate_identity_sql, apply_identity_to_tables
│   └── heartbeat.py  # generate_heartbeat_sql, generate_drop_heartbeat_sql
└── sqlalchemy/
    └── mixins/
        ├── str_mixin.py   # WAuditStrMixin
        └── uuid_mixin.py  # WAuditUUIDMixin

---

Development

This project uses uv for dependency management and builds.

git clone https://github.com/Mac3g0d/waudit-tools.git
cd waudit-tools

# Install with dev dependencies (uv will use the uv.lock file)
uv sync --group dev

# Run tests
uv run pytest

# Lint & typecheck
uv run ruff check src tests
uv run mypy src

# Build wheel + sdist
uv build

Quality gates:

• ruff: select = ["ALL"]
• mypy: strict mode
• pytest-cov: 85% minimum (currently 100%)

---

License

MIT — see LICENSE.

Author

Wootya [e-mail]

Links

• Repository: https://github.com/Mac3g0d/waudit-tools
• Issues: https://github.com/Mac3g0d/waudit-tools/issues