← назад к разделу

Администратор может делать то, что обычным пользователям недоступно: отменить чужой заказ, заблокировать аккаунт, изменить статус вручную. Если такие действия нигде не фиксируются, при компрометации admin-аккаунта невозможно восстановить ни что было сделано, ни к каким данным получили доступ. Журнал действий решает эту проблему.

Что фиксировать и в какой форме

Каждое действие администратора, которое меняет данные, должно оставлять запись. Минимальный набор полей:

  • actor_id — кто сделал (идентификатор из JWT, не email);
  • action — что сделал ("cancel-order", "block-customer");
  • resource_type и resource_id — к чему применил ("Order", "42");
  • occurred_at — когда (временная метка с часовым поясом);
  • metadata — детали в свободном JSON: статус до/после, причина, владелец ресурса.

Email и другие персональные данные в metadata не кладут — только идентификатор (sub из токена).

Таблица в базе данных

Одна таблица на сервис (или отдельная на каждый агрегат, если структура метаданных принципиально различается):

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 — таблица только для добавления записей. Изменять или удалять строки нельзя. Это гарантирует, что записи не исчезнут даже при ошибке в коде приложения.

metadata JSONB не имеет фиксированной схемы — добавляете туда нужные поля по мере необходимости: previous_status, new_status, reason, owner_customer_id.

SQLAlchemy-модель и репозиторий

# 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 не вызывает commit — транзакцией управляет хэндлер или FastAPI-зависимость. Это важно: запись в журнал должна попасть в ту же транзакцию, что и бизнес-операция.

Одна транзакция для бизнес-операции и журнала

Самая распространённая ошибка — писать в журнал уже после commit бизнес-изменения. Тогда при сбое между двумя операциями данные изменятся, но запись не появится. Правильно — держать обе операции в одной транзакции:

# 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() охватывает и сохранение заказа, и запись в журнал. Если хэндлер бросил исключение — оба откатились. Не выносите журнал в отдельный HTTP-вызов или очередь: событие может потеряться между commit и publish, и изменение окажется незафиксированным.

Способ 1 — декоратор

Для типовых admin-команд удобен декоратор: нельзя случайно забыть добавить журналирование при написании нового хэндлера.

# 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

Декоратор срабатывает только при principal.is_admin() — обычные пользователи через тот же хэндлер не оставляют записей в журнале.

Способ 2 — явный вызов

Когда в журнале нужны детали, доступные только внутри метода (статус до изменения, владелец ресурса), явный вызов читается нагляднее:

# 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

Выбор между декоратором и явным вызовом: если метаданные простые — декоратор; если нужны данные из середины метода (статус до изменения, вычисленные поля) — явный вызов.

Те же правила для других агрегатов

Паттерн одинаков для любой admin-команды — блокировка пользователя, принудительная публикация товара, ручная корректировка баланса:

# 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

Частые ошибки

Журнал после commit. Если audit.append вызывается уже после того, как транзакция закрыта, то при любом сбое между ними изменение есть, а записи нет. Используйте одну сессию для обеих операций.

Email вместо идентификатора. В metadata и actor_id кладут только sub из JWT — стабильный идентификатор, не адрес почты. Email — персональные данные, которые меняются и не должны попадать в постоянные записи в открытом виде.

Только деструктивные действия. Критичные операции чтения тоже подлежат фиксации — например, когда администратор просматривает персональные данные чужого аккаунта.

Отдельная очередь для журнала. Событие может потеряться между commit бизнес-операции и отправкой в очередь. Одна транзакция надёжнее.

Изменяемая таблица. Строки журнала нельзя редактировать или удалять — это снижает доверие к записям. REVOKE UPDATE, DELETE закрывает эту возможность на уровне базы.

Коротко

  • Каждое изменяющее данные действие администратора фиксируется в admin_audit_log.
  • Обязательные поля: actor_id, action, resource_type, resource_id, occurred_at.
  • В metadata — детали (статус до/после, причина); персональные данные туда не кладут.
  • Таблица только для добавления записей: REVOKE UPDATE, DELETE FROM app_role.
  • Запись в журнал и бизнес-операция — в одной транзакции SQLAlchemy.
  • Декоратор @admin_audit — для типовых команд; явный вызов — когда нужны данные из середины метода.
  • Не выносите журналирование в очередь или отдельный HTTP-вызов.

Что почитать дальше