GovOPlaN/govoplan-audit
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(api): share full audit delta response

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-21 13:29:19 +02:00
parent 86b20c65cb
commit a34935da02
1 changed files with 44 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

80
src/govoplan_audit/backend/api/v1/routes.py
View File

@@ -196,6 +196,36 @@ def _audit_delta_response_watermark(
return encode_sequence_watermark(entries[-1].id) if has_more and entries else _audit_delta_watermark(session, effective_scope=effective_scope, tenant_id=tenant_id)
def _full_audit_delta_response(
session: Session,
*,
context: AuditAdminQueryContext,
page_query: Any,
start_cursor: str | None,
sort_by: str,
sort_direction: str,
) -> AuditAdminDeltaResponse:
rows_plus_one = page_query.order_by(context.order, AuditLog.id.desc()).limit(context.effective_page_size + 1).all()
rows = rows_plus_one[:context.effective_page_size]
next_cursor = (
_audit_cursor_for_row(rows[-1], sort_by=sort_by, sort_direction=sort_direction, fingerprint=context.fingerprint)
if len(rows_plus_one) > context.effective_page_size and rows else None
)
return AuditAdminDeltaResponse(
total=context.total,
page=1,
page_size=context.effective_page_size,
pages=context.pages,
cursor=start_cursor,
next_cursor=next_cursor,
items=_audit_items(session, rows, context.access_admin),
deleted=[],
watermark=_audit_delta_watermark(session, effective_scope=context.effective_scope, tenant_id=context.resolved_tenant_id),
has_more=False,
full=True,
)
def _audit_items(session: Session, rows: list[AuditLog], access_admin: AccessAdministration) -> list[AuditAdminItem]:
actor_email_by_user_id = access_admin.actor_email_by_user_id(session, {row.user_id for row in rows if row.user_id})
return [
@@ -530,24 +560,13 @@ def list_admin_audit_delta(
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc
if since is None:
rows_plus_one = page_query.order_by(context.order, AuditLog.id.desc()).limit(context.effective_page_size + 1).all()
rows = rows_plus_one[:context.effective_page_size]
next_cursor = (
_audit_cursor_for_row(rows[-1], sort_by=sort_by, sort_direction=sort_direction, fingerprint=context.fingerprint)
if len(rows_plus_one) > context.effective_page_size and rows else None
)
return AuditAdminDeltaResponse(
total=context.total,
page=1,
page_size=context.effective_page_size,
pages=context.pages,
cursor=start_cursor,
next_cursor=next_cursor,
items=_audit_items(session, rows, context.access_admin),
deleted=[],
watermark=_audit_delta_watermark(session, effective_scope=context.effective_scope, tenant_id=context.resolved_tenant_id),
has_more=False,
full=True,
return _full_audit_delta_response(
session,
context=context,
page_query=page_query,
start_cursor=start_cursor,
sort_by=sort_by,
sort_direction=sort_direction,
)
entries, has_more = _audit_delta_entries(
@@ -558,24 +577,13 @@ def list_admin_audit_delta(
limit=context.effective_page_size,
)
if entries is None:
rows_plus_one = page_query.order_by(context.order, AuditLog.id.desc()).limit(context.effective_page_size + 1).all()
rows = rows_plus_one[:context.effective_page_size]
next_cursor = (
_audit_cursor_for_row(rows[-1], sort_by=sort_by, sort_direction=sort_direction, fingerprint=context.fingerprint)
if len(rows_plus_one) > context.effective_page_size and rows else None
)
return AuditAdminDeltaResponse(
total=context.total,
page=1,
page_size=context.effective_page_size,
pages=context.pages,
cursor=start_cursor,
next_cursor=next_cursor,
items=_audit_items(session, rows, context.access_admin),
deleted=[],
watermark=_audit_delta_watermark(session, effective_scope=context.effective_scope, tenant_id=context.resolved_tenant_id),
has_more=False,
full=True,
return _full_audit_delta_response(
session,
context=context,
page_query=page_query,
start_cursor=start_cursor,
sort_by=sort_by,
sort_direction=sort_direction,
)
changed_ids = [entry.resource_id for entry in entries if entry.resource_type == "audit_log"]