← Back to the section

An administrator can do things that ordinary users cannot: cancel someone else's order, block an account, change a status manually. If such actions are not recorded anywhere, then when an admin account is compromised it is impossible to reconstruct either what was done or what data was accessed. An action log solves this problem.

What to record and in what form

Every administrator action that changes data must leave a record. The minimal set of fields:

  • actor_id — who did it (the identifier from the JWT, not the email);
  • action — what they did ("cancel-order", "block-customer");
  • resource_type and resource_id — what they applied it to ("Order", "42");
  • occurred_at — when (a timestamp with a time zone);
  • metadata — details in free-form JSON: status before/after, reason, resource owner.

Email and other personal data are not put into metadata — only the identifier (sub from the token).

The database table

One table per service (or a separate one per aggregate if the metadata structure differs fundamentally):

CREATE TABLE admin_audit_log (
    id            bigserial PRIMARY KEY,
    actor_id      text        NOT NULL,
    action        text        NOT NULL,
    resource_type text        NOT NULL,
    resource_id   text        NOT NULL,
    occurred_at   timestamptz NOT NULL DEFAULT now(),
    metadata      jsonb       NOT NULL DEFAULT '{}',
    request_id    text,
    trace_id      text
);

CREATE INDEX ix_admin_audit_actor    ON admin_audit_log (actor_id, occurred_at DESC);
CREATE INDEX ix_admin_audit_resource ON admin_audit_log (resource_type, resource_id);

REVOKE UPDATE, DELETE ON admin_audit_log FROM app_role;

REVOKE UPDATE, DELETE — the table is append-only. You cannot modify or delete rows. This guarantees that records will not disappear even if there is a bug in the application code.

The metadata JSONB has no fixed schema — you add the fields you need as required: previous_status, new_status, reason, owner_customer_id.

SQLAlchemy model and repository

# adapters/out/db/audit_models.py
from datetime import datetime, UTC
from sqlalchemy import Column, BigInteger, Text, DateTime, Index
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

class AdminAuditLog(Base):
    __tablename__ = "admin_audit_log"

    id            = Column(BigInteger, primary_key=True, autoincrement=True)
    actor_id      = Column(Text, nullable=False)
    action        = Column(Text, nullable=False)
    resource_type = Column(Text, nullable=False)
    resource_id   = Column(Text, nullable=False)
    occurred_at   = Column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(UTC))
    metadata      = Column(JSONB, nullable=False, default=dict)
    request_id    = Column(Text)
    trace_id      = Column(Text)

    __table_args__ = (
        Index("ix_admin_audit_actor",    "actor_id", "occurred_at"),
        Index("ix_admin_audit_resource", "resource_type", "resource_id"),
    )

# adapters/out/db/audit_repository.py
from dataclasses import dataclass
from datetime import datetime, UTC
from sqlalchemy.ext.asyncio import AsyncSession
from .audit_models import AdminAuditLog

@dataclass
class AuditRecord:
    actor_id:      str
    action:        str
    resource_type: str
    resource_id:   str
    metadata:      dict
    request_id:    str | None = None
    trace_id:      str | None = None

class AdminAuditRepository:
    def __init__(self, session: AsyncSession) -> None:
        self._session = session

    async def append(self, record: AuditRecord) -> None:
        self._session.add(AdminAuditLog(
            actor_id      = record.actor_id,
            action        = record.action,
            resource_type = record.resource_type,
            resource_id   = record.resource_id,
            occurred_at   = datetime.now(UTC),
            metadata      = record.metadata,
            request_id    = record.request_id,
            trace_id      = record.trace_id,
        ))

append does not call commit — the transaction is managed by the handler or a FastAPI dependency. This matters: the log record must land in the same transaction as the business operation.

One transaction for the business operation and the log

The most common mistake is writing to the log after the business change has already been committed. Then, if a failure occurs between the two operations, the data changes but no record appears. The right approach is to keep both operations in one transaction:

# adapters/in/http/dependencies.py
from contextlib import asynccontextmanager
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

@asynccontextmanager
async def db_session(factory: async_sessionmaker[AsyncSession]):
    async with factory() as session:
        async with session.begin():
            yield session

# adapters/in/http/routers/orders.py
@router.delete("/orders/{order_id}")
async def cancel_order(
    order_id:  uuid.UUID,
    principal: Principal = Depends(require_roles("admin", "customer")),
    session:   AsyncSession = Depends(get_session),
):
    handler = CancelOrderHandler(
        orders = SqlOrderRepository(session),
        audit  = AdminAuditRepository(session),
    )
    return await handler.handle(CancelOrderCommand(order_id=order_id), principal)

session.begin() covers both saving the order and writing to the log. If the handler raises an exception, both are rolled back. Do not move the log into a separate HTTP call or a queue: the event can be lost between commit and publish, and the change would end up unrecorded.

Option 1 — a decorator

For typical admin commands a decorator is convenient: you cannot accidentally forget to add logging when writing a new handler.

# application/audit.py
import functools
from collections.abc import Callable
from typing import Any

def admin_audit(action: str, resource_type: str, resource_id_attr: str = "id"):
    def decorator(fn: Callable) -> Callable:
        @functools.wraps(fn)
        async def wrapper(self, command, principal, *args, **kwargs) -> Any:
            result = await fn(self, command, principal, *args, **kwargs)
            if principal.is_admin():
                resource_id = str(getattr(command, resource_id_attr))
                await self._audit.append(AuditRecord(
                    actor_id      = principal.sub,
                    action        = action,
                    resource_type = resource_type,
                    resource_id   = resource_id,
                    metadata      = {},
                    request_id    = getattr(command, "request_id", None),
                    trace_id      = getattr(command, "trace_id", None),
                ))
            return result
        return wrapper
    return decorator

# application/cancel_order_handler.py
class CancelOrderHandler:
    def __init__(self, orders: OrderRepository, audit: AdminAuditRepository) -> None:
        self._orders = orders
        self._audit  = audit

    @admin_audit(action="cancel-order", resource_type="Order", resource_id_attr="order_id")
    async def handle(self, command: CancelOrderCommand, principal: Principal) -> Order:
        order = await self._orders.find_by_id(command.order_id)
        if order is None:
            raise OrderNotFoundError(command.order_id)
        if not principal.is_admin() and order.customer_id != principal.sub:
            raise ForbiddenError()
        order.cancel()
        await self._orders.save(order)
        return order

The decorator fires only when principal.is_admin() — ordinary users going through the same handler leave no records in the log.

Option 2 — an explicit call

When the log needs details available only inside the method (the status before the change, the resource owner), an explicit call reads more clearly:

# application/cancel_order_handler.py
class CancelOrderHandler:
    def __init__(self, orders: OrderRepository, audit: AdminAuditRepository) -> None:
        self._orders = orders
        self._audit  = audit

    async def handle(self, command: CancelOrderCommand, principal: Principal) -> Order:
        order = await self._orders.find_by_id(command.order_id)
        if order is None:
            raise OrderNotFoundError(command.order_id)
        if not principal.is_admin() and order.customer_id != principal.sub:
            raise ForbiddenError()

        previous_status = order.status
        order.cancel()
        await self._orders.save(order)

        if principal.is_admin():
            await self._audit.append(AuditRecord(
                actor_id      = principal.sub,
                action        = "cancel-order",
                resource_type = "Order",
                resource_id   = str(command.order_id),
                metadata      = {
                    "previous_status":   previous_status.value,
                    "new_status":        order.status.value,
                    "owner_customer_id": str(order.customer_id),
                },
                request_id = command.request_id,
                trace_id   = command.trace_id,
            ))

        return order

Choosing between a decorator and an explicit call: if the metadata is simple — a decorator; if you need data from the middle of the method (status before the change, computed fields) — an explicit call.

The same rules for other aggregates

The pattern is the same for any admin command — blocking a user, force-publishing a product, manually adjusting a balance:

# application/block_customer_handler.py
class BlockCustomerHandler:
    def __init__(self, customers: CustomerRepository, audit: AdminAuditRepository) -> None:
        self._customers = customers
        self._audit     = audit

    async def handle(self, command: BlockCustomerCommand, principal: Principal) -> Customer:
        customer = await self._customers.find_by_id(command.customer_id)
        if customer is None:
            raise CustomerNotFoundError(command.customer_id)

        previous_status = customer.status
        customer.block(reason=command.reason)
        await self._customers.save(customer)

        await self._audit.append(AuditRecord(
            actor_id      = principal.sub,
            action        = "block-customer",
            resource_type = "Customer",
            resource_id   = str(command.customer_id),
            metadata      = {
                "previous_status": previous_status.value,
                "reason":          command.reason,
            },
        ))

        return customer

Common mistakes

Logging after commit. If audit.append is called after the transaction is already closed, then on any failure between them the change exists but no record does. Use one session for both operations.

Email instead of an identifier. In metadata and actor_id you put only the sub from the JWT — a stable identifier, not an email address. Email is personal data that changes and should not end up in permanent records in plain form.

Only destructive actions. Critical read operations also warrant recording — for example, when an administrator views the personal data of someone else's account.

A separate queue for the log. The event can be lost between the commit of the business operation and its dispatch to the queue. One transaction is more reliable.

A mutable table. Log rows cannot be edited or deleted — that would undermine trust in the records. REVOKE UPDATE, DELETE removes this possibility at the database level.

In short

  • Every data-changing administrator action is recorded in admin_audit_log.
  • Mandatory fields: actor_id, action, resource_type, resource_id, occurred_at.
  • In metadata — details (status before/after, reason); personal data is not put there.
  • The table is append-only: REVOKE UPDATE, DELETE FROM app_role.
  • The log record and the business operation are in one SQLAlchemy transaction.
  • The @admin_audit decorator — for typical commands; an explicit call — when you need data from the middle of the method.
  • Do not move logging into a queue or a separate HTTP call.

Further reading