feat(platform): add audit logs and activity feed

This commit is contained in:
Schubert Ferenc 2026-07-03 00:04:44 +02:00
parent 92cb8d1286
commit c816e9869d
34 changed files with 1592 additions and 43 deletions

View file

@ -0,0 +1,80 @@
from datetime import datetime
from fastapi import APIRouter, Depends, Query
from sqlalchemy.orm import Session
from app.core.rbac import get_user_permission_names, require_permission
from app.db.database import get_db
from app.models.user import User
from app.repositories.audit_repository import AuditRepository
from app.schemas.api_response import ApiSuccess
from app.schemas.audit import ActivityFeedItem, AuditLogListResponse
from app.services.audit_service import audit_to_activity
router = APIRouter(tags=["Audit"])
@router.get("/audit-logs", response_model=ApiSuccess)
def get_audit_logs(
page: int = Query(default=1, ge=1),
page_size: int = Query(default=25, ge=1, le=100),
entity_type: str | None = None,
actor_user_id: int | None = None,
action: str | None = None,
date_from: datetime | None = None,
date_to: datetime | None = None,
db: Session = Depends(get_db),
current_user: User = Depends(require_permission("audit_logs.read")),
):
items, total = AuditRepository.list(
db,
page=page,
page_size=page_size,
entity_type=entity_type,
actor_user_id=actor_user_id,
action=action,
date_from=date_from,
date_to=date_to,
)
return ApiSuccess(
data=AuditLogListResponse(
items=items,
total=total,
page=page,
page_size=page_size,
),
message="Audit Logs geladen",
)
def can_read_activity(action: str, permissions: set[str]) -> bool:
if action.startswith("users."):
return "users.read" in permissions
if action.startswith("customers.") or action.startswith("customer_contacts."):
return "customers.read" in permissions
if action.startswith("roles."):
return "roles.read" in permissions
if action.startswith("audit_logs."):
return "audit_logs.read" in permissions
if action.startswith("auth."):
return "audit_logs.read" in permissions
return False
@router.get("/activity-feed", response_model=ApiSuccess)
def get_activity_feed(
limit: int = Query(default=10, ge=1, le=50),
db: Session = Depends(get_db),
current_user: User = Depends(require_permission("dashboard.read")),
):
permissions = get_user_permission_names(current_user)
activities: list[ActivityFeedItem] = []
for audit_log in AuditRepository.latest(db, limit=100):
if not can_read_activity(audit_log.action, permissions):
continue
activities.append(ActivityFeedItem(**audit_to_activity(audit_log)))
if len(activities) >= limit:
break
return ApiSuccess(data=activities, message="Aktivitäten geladen")