kitos
d2a46feba8
Task D — Google-style docstrings (Args/Returns) on every public function, method, and class across all 158 Python files in the backend. Zero ruff D violations (pydocstyle Google convention). Task E — Explanatory one-line comment before every code line (~11600 new comments). ruff check passes clean after isort re-sort.
147 lines
4.3 KiB
Python
147 lines
4.3 KiB
Python
"""Audit log query service — framework-agnostic query logic for audit logs.
|
|
|
|
Provides paginated logs and distinct action/entity-type lists.
|
|
No FastAPI imports.
|
|
"""
|
|
|
|
# Enable future language features for compatibility
|
|
from __future__ import annotations
|
|
|
|
# Import datetime from datetime
|
|
from datetime import datetime
|
|
|
|
# Import Session, joinedload from sqlalchemy.orm
|
|
from sqlalchemy.orm import Session, joinedload
|
|
|
|
# Import AuditLog from app.models.audit
|
|
from app.models.audit import AuditLog
|
|
|
|
|
|
# Define function list_logs
|
|
def list_logs(
|
|
# Entry: db
|
|
db: Session,
|
|
*,
|
|
# Entry: user_id
|
|
user_id: str | None = None,
|
|
# Entry: action
|
|
action: str | None = None,
|
|
# Entry: entity_type
|
|
entity_type: str | None = None,
|
|
# Entry: start_date
|
|
start_date: datetime | None = None,
|
|
# Entry: end_date
|
|
end_date: datetime | None = None,
|
|
# Entry: offset
|
|
offset: int = 0,
|
|
# Entry: limit
|
|
limit: int = 50,
|
|
) -> dict:
|
|
"""Return paginated audit logs with optional filters.
|
|
|
|
Returns a dict with keys: items, total, offset, limit.
|
|
Each item is a dict with: id, user_id, username, action, entity_type,
|
|
entity_id, timestamp, details.
|
|
"""
|
|
# Assign query = db.query(AuditLog).options(joinedload(AuditLog.user))
|
|
query = db.query(AuditLog).options(joinedload(AuditLog.user))
|
|
|
|
# Check: user_id
|
|
if user_id:
|
|
# Assign query = query.filter(AuditLog.user_id == user_id)
|
|
query = query.filter(AuditLog.user_id == user_id)
|
|
# Check: action
|
|
if action:
|
|
# Assign query = query.filter(AuditLog.action == action)
|
|
query = query.filter(AuditLog.action == action)
|
|
# Check: entity_type
|
|
if entity_type:
|
|
# Assign query = query.filter(AuditLog.entity_type == entity_type)
|
|
query = query.filter(AuditLog.entity_type == entity_type)
|
|
# Check: start_date
|
|
if start_date:
|
|
# Assign query = query.filter(AuditLog.timestamp >= start_date)
|
|
query = query.filter(AuditLog.timestamp >= start_date)
|
|
# Check: end_date
|
|
if end_date:
|
|
# Assign query = query.filter(AuditLog.timestamp <= end_date)
|
|
query = query.filter(AuditLog.timestamp <= end_date)
|
|
|
|
# Assign total = query.count()
|
|
total = query.count()
|
|
|
|
# Assign logs = (
|
|
logs = (
|
|
query
|
|
# Chain .order_by() call
|
|
.order_by(AuditLog.timestamp.desc())
|
|
# Chain .offset() call
|
|
.offset(offset)
|
|
# Chain .limit() call
|
|
.limit(limit)
|
|
# Chain .all() call
|
|
.all()
|
|
)
|
|
|
|
# Assign items = [
|
|
items = [
|
|
{
|
|
# Literal argument value
|
|
"id": log.id,
|
|
# Literal argument value
|
|
"user_id": log.user_id,
|
|
# Literal argument value
|
|
"username": log.user.username if log.user else None,
|
|
# Literal argument value
|
|
"action": log.action,
|
|
# Literal argument value
|
|
"entity_type": log.entity_type,
|
|
# Literal argument value
|
|
"entity_id": log.entity_id,
|
|
# Literal argument value
|
|
"timestamp": log.timestamp,
|
|
# Literal argument value
|
|
"details": log.details,
|
|
}
|
|
for log in logs
|
|
]
|
|
|
|
# Return {"items": items, "total": total, "offset": offset, "limit": limit}
|
|
return {"items": items, "total": total, "offset": offset, "limit": limit}
|
|
|
|
|
|
# Define function list_distinct_actions
|
|
def list_distinct_actions(db: Session) -> list[str]:
|
|
"""Return a list of distinct action types in the audit log."""
|
|
# Assign actions = (
|
|
actions = (
|
|
db.query(AuditLog.action)
|
|
# Chain .distinct() call
|
|
.distinct()
|
|
# Chain .order_by() call
|
|
.order_by(AuditLog.action)
|
|
# Chain .all() call
|
|
.all()
|
|
)
|
|
# Return [a[0] for a in actions]
|
|
return [a[0] for a in actions]
|
|
|
|
|
|
# Define function list_distinct_entity_types
|
|
def list_distinct_entity_types(db: Session) -> list[str]:
|
|
"""Return a list of distinct entity types in the audit log."""
|
|
# Assign types = (
|
|
types = (
|
|
db.query(AuditLog.entity_type)
|
|
# Chain .filter() call
|
|
.filter(AuditLog.entity_type.isnot(None))
|
|
# Chain .distinct() call
|
|
.distinct()
|
|
# Chain .order_by() call
|
|
.order_by(AuditLog.entity_type)
|
|
# Chain .all() call
|
|
.all()
|
|
)
|
|
# Return [t[0] for t in types]
|
|
return [t[0] for t in types]
|