Event System API Reference¶

class EventType(str, enum.Enum):
    CREATE = "create"
    UPDATE = "update"
    DELETE = "delete"
    LOGIN = "login"
    LOGOUT = "logout"
    FAILED_LOGIN = "failed_login"

class EventStatus(str, enum.Enum):
    SUCCESS = "success"
    FAILURE = "failure"
    WARNING = "warning"

class EventService:
    def __init__(self, db_config):
        self.db_config = db_config
        self.crud_events: FastCRUD[Any, Any, Any, Any, Any, Any] = FastCRUD(
            db_config.AdminEventLog
        )
        self.crud_audits: FastCRUD[Any, Any, Any, Any, Any, Any] = FastCRUD(
            db_config.AdminAuditLog
        )
        self.json_encoder = CustomJSONEncoder()

    def _serialize_dict(self, data: dict | None) -> dict:
        if not data:
            return {}
        return cast(dict, json.loads(self.json_encoder.encode(data)))

    async def log_event(
        self,
        db: AsyncSession,
        event_type: EventType,
        status: EventStatus,
        user_id: int,
        session_id: str,
        request: Request,
        resource_type: str | None = None,
        resource_id: str | None = None,
        details: dict | None = None,
        commit: bool = True,
    ) -> AdminEventLogRead:
        ip_address = request.client.host if request.client else "unknown"

        event_data = AdminEventLogCreate(
            event_type=event_type,
            status=status,
            user_id=user_id,
            session_id=session_id,
            ip_address=ip_address,
            user_agent=request.headers.get("user-agent", "")[:USER_AGENT_LENGTH],
            resource_type=resource_type,
            resource_id=resource_id,
            details=self._serialize_dict(redact_secrets(details)),
        )

        result = await self.crud_events.create(
            db=db,
            object=event_data,
            schema_to_select=AdminEventLogRead,
            return_as_model=False,
            commit=commit,
        )
        return AdminEventLogRead(**cast(dict, result))

    async def create_audit_log(
        self,
        db: AsyncSession,
        event_id: int,
        resource_type: str,
        resource_id: str,
        action: str,
        previous_state: dict | None = None,
        new_state: dict | None = None,
        metadata: dict | None = None,
        commit: bool = True,
    ) -> AdminAuditLogRead:
        audit_data = AdminAuditLogCreate(
            event_id=event_id,
            resource_type=resource_type,
            resource_id=resource_id,
            action=action,
            previous_state=self._serialize_dict(redact_secrets(previous_state)),
            new_state=self._serialize_dict(redact_secrets(new_state)),
            changes=self._serialize_dict(
                redact_secrets(self._compute_changes(previous_state, new_state))
            ),
            audit_metadata=self._serialize_dict(redact_secrets(metadata)),
        )

        result = await self.crud_audits.create(
            db=db,
            object=audit_data,
            schema_to_select=AdminAuditLogRead,
            return_as_model=False,
            commit=commit,
        )

        return AdminAuditLogRead(**cast(dict, result))

    def _compute_changes(
        self,
        previous_state: dict | None,
        new_state: dict | None,
    ) -> dict:
        """Compute changes between previous and new states."""
        changes: dict = {}

        if not previous_state or not new_state:
            return changes

        all_keys = set(previous_state.keys()) | set(new_state.keys())

        for key in all_keys:
            old_value = previous_state.get(key)
            new_value = new_state.get(key)

            if old_value != new_value:
                changes[key] = {"old": old_value, "new": new_value}

        return changes

    async def get_user_activity(
        self,
        db: AsyncSession,
        user_id: int,
        start_time: datetime | None = None,
        end_time: datetime | None = None,
        limit: int = 50,
        offset: int = 0,
    ) -> dict:
        """Get user activity logs."""
        filters: dict = {"user_id": user_id}

        if start_time:
            filters["timestamp__gte"] = start_time
        if end_time:
            filters["timestamp__lte"] = end_time

        result = await self.crud_events.get_multi(
            db,
            offset=offset,
            limit=limit,
            sort_columns=["timestamp"],
            sort_orders=["desc"],
            **filters,
        )

        return cast(dict, result)

    async def get_resource_history(
        self,
        db: AsyncSession,
        resource_type: str,
        resource_id: str,
        limit: int = 50,
        offset: int = 0,
    ) -> dict:
        """Get audit history for a specific resource."""
        result = await self.crud_audits.get_multi(
            db,
            offset=offset,
            limit=limit,
            sort_columns=["timestamp"],
            sort_orders=["desc"],
            resource_type=resource_type,
            resource_id=resource_id,
        )

        return cast(dict[str, Any], result)

    async def get_security_alerts(
        self, db: AsyncSession, lookback_hours: int = 24
    ) -> list[dict[str, Any]]:
        """Get security alerts based on event patterns."""
        alerts: list[dict[str, Any]] = []
        lookback_time = datetime.now(UTC) - timedelta(hours=lookback_hours)

        failed_logins = await self.crud_events.get_multi(
            db,
            event_type=EventType.FAILED_LOGIN,
            status=EventStatus.FAILURE,
            timestamp__gte=lookback_time,
        )

        failed_login_patterns: dict[tuple, int] = {}

        for login in failed_logins.get("data", []):
            key = (
                login.get("ip_address", "unknown"),
                login.get("details", {}).get("username", "unknown"),
            )
            failed_login_patterns[key] = failed_login_patterns.get(key, 0) + 1

        for (ip, username), count in failed_login_patterns.items():
            if count >= 5:
                alerts.append(
                    {
                        "type": "multiple_failed_logins",
                        "severity": "high",
                        "details": {
                            "ip_address": ip,
                            "username": username,
                            "attempts": count,
                        },
                    }
                )

        return alerts

    async def cleanup_old_logs(
        self, db: AsyncSession, retention_days: int = 90
    ) -> None:
        """Clean up old logs based on retention policy."""
        cutoff_date = datetime.now(UTC) - timedelta(days=retention_days)

        await self.crud_events.delete(
            db, allow_multiple=True, timestamp__lt=cutoff_date
        )

        await self.crud_audits.delete(
            db, allow_multiple=True, timestamp__lt=cutoff_date
        )

class EventSystemIntegration:
    def __init__(self, event_service: EventService):
        self.event_service = event_service

    async def log_model_event(
        self,
        db: AsyncSession,
        event_type: EventType,
        model: type[DeclarativeBase],
        user_id: int,
        session_id: str,
        request: Request,
        resource_id: str | None = None,
        previous_state: dict[str, Any] | None = None,
        new_state: dict[str, Any] | None = None,
        details: dict[str, Any] | None = None,
        succeeded: bool = True,
        deleted_records: list[dict[str, Any]] | None = None,
        primary_key_name: str = "id",
    ):
        """Record a model change: one event, and its audit rows if the change happened.

        A failed action is recorded as a FAILURE event without audit rows, since
        nothing changed. A delete gets one audit row per deleted record, keyed by
        the model's primary key. The event and its audit rows are written in one
        transaction, rolled back together if any write fails.
        """
        try:
            event = await self.event_service.log_event(
                db=db,
                event_type=event_type,
                status=EventStatus.SUCCESS if succeeded else EventStatus.FAILURE,
                user_id=user_id,
                session_id=session_id,
                request=request,
                resource_type=model.__name__,
                resource_id=str(resource_id) if resource_id else None,
                details=details,
                commit=False,
            )

            if succeeded and event:
                for audited_id, before, after in self._audit_entries(
                    event_type,
                    resource_id,
                    previous_state,
                    new_state,
                    deleted_records,
                    primary_key_name,
                ):
                    await self.event_service.create_audit_log(
                        db=db,
                        event_id=event.id,
                        resource_type=model.__name__,
                        resource_id=audited_id,
                        action=event_type.value,
                        previous_state=before,
                        new_state=after,
                        metadata=details,
                        commit=False,
                    )

            await db.commit()
            return event

        except Exception:
            await db.rollback()
            raise

    @staticmethod
    def _audit_entries(
        event_type: EventType,
        resource_id: str | None,
        previous_state: dict[str, Any] | None,
        new_state: dict[str, Any] | None,
        deleted_records: list[dict[str, Any]] | None,
        primary_key_name: str,
    ) -> list[tuple[str, dict[str, Any] | None, dict[str, Any] | None]]:
        """``(resource_id, previous_state, new_state)`` for each audit row to write."""
        if event_type == EventType.DELETE:
            return [
                (str(record[primary_key_name]), record, None)
                for record in deleted_records or []
                if record.get(primary_key_name) is not None
            ]
        if event_type in (EventType.CREATE, EventType.UPDATE) and resource_id:
            return [(str(resource_id), previous_state, new_state)]
        return []

    async def log_auth_event(
        self,
        db: AsyncSession,
        event_type: EventType,
        user_id: int,
        session_id: str,
        request: Request,
        success: bool,
        details: dict[str, Any] | None = None,
    ):
        """Record an authentication event; a failure is logged, never raised.

        It runs inside login and logout, which must not fail because the event
        log can't be written.
        """
        try:
            status = EventStatus.SUCCESS if success else EventStatus.FAILURE

            await self.event_service.log_event(
                db=db,
                event_type=event_type,
                status=status,
                user_id=user_id,
                session_id=session_id,
                request=request,
                details=details,
            )

        except Exception:
            logger.exception("Could not record the %s event", event_type.value)
            await db.rollback()

    async def log_security_event(
        self,
        db: AsyncSession,
        event_type: EventType,
        user_id: int,
        session_id: str,
        request: Request,
        details: dict[str, Any],
    ):
        """Log security-related events with high priority."""
        event = await self.event_service.log_event(
            db=db,
            event_type=event_type,
            status=EventStatus.WARNING,
            user_id=user_id,
            session_id=session_id,
            request=request,
            details={**details, "priority": "high", "requires_attention": True},
        )

        return event

class AdminEvents:
    """Records changes made through the admin in the event log.

    The admin's create, update and delete routes call [record][] once they know
    the outcome, with the records as they were and as they are; a route of your
    own mounted behind the admin's login can do the same. With event tracking
    off, ``record`` does nothing.

    Secrets in the records are redacted before they are stored. A failure to
    write the event is logged and absorbed, since the change it describes is
    already committed.

    Example:
        ```python
        await admin.events.record(
            request,
            admin_db,
            EventType.UPDATE,
            Product,
            record_id=product.id,
            before=before,
            after=after,
        )
        ```
    """

    def __init__(self, integration: EventSystemIntegration | None) -> None:
        self.integration = integration

    @property
    def enabled(self) -> bool:
        return self.integration is not None

    async def record(
        self,
        request: Request,
        admin_db: AsyncSession,
        event_type: EventType,
        model: type[DeclarativeBase],
        *,
        record_id: Any = None,
        before: dict[str, Any] | None = None,
        after: dict[str, Any] | None = None,
        deleted: list[dict[str, Any]] | None = None,
        succeeded: bool = True,
    ) -> None:
        """Record one create, update or delete of ``model`` by the logged-in admin.

        Args:
            request: The admin's request; the admin and the session handle are
                read from it.
            admin_db: A session on the admin database, where events are kept.
            event_type: ``CREATE``, ``UPDATE`` or ``DELETE``.
            model: The model that changed.
            record_id: The changed record's primary key.
            before: The record before an update.
            after: The record after a create or an update.
            deleted: The records a delete removed, as they were.
            succeeded: False records a refused change, without audit rows.
        """
        if self.integration is None:
            return
        admin = getattr(request.state, "user", None) or {}
        try:
            await self.integration.log_model_event(
                db=admin_db,
                event_type=event_type,
                model=model,
                user_id=admin.get("id") or 0,
                session_id=getattr(request.state, "session_handle", None) or "unknown",
                request=request,
                resource_id=None if record_id is None else str(record_id),
                previous_state=before,
                new_state=after,
                details=self._details(
                    request, event_type, model, record_id, before, after, deleted
                ),
                succeeded=succeeded,
                deleted_records=deleted,
                primary_key_name=_primary_key_name_or_id(model),
            )
        except Exception:
            logger.exception(
                "Could not record the %s of %s; the change itself stands",
                event_type.value,
                model.__name__,
            )

    @staticmethod
    def _details(
        request: Request,
        event_type: EventType,
        model: type[DeclarativeBase],
        record_id: Any,
        before: dict[str, Any] | None,
        after: dict[str, Any] | None,
        deleted: list[dict[str, Any]] | None,
    ) -> dict[str, Any]:
        """What the event log page shows: the record's changes, and the request."""
        changes: Any
        if event_type == EventType.UPDATE:
            changes = changed_fields(before, after)
        elif event_type == EventType.DELETE:
            changes = {"deleted_records": deleted or []}
        else:
            changes = after
        return {
            "resource_details": {
                "model": model.__name__,
                "id": None if record_id is None else str(record_id),
                "changes": changes,
            },
            "request_details": {
                "method": request.method,
                "path": request.url.path,
                "user_agent": request.headers.get("user-agent"),
            },
        }