Cheng Zhou
70 lines
1.8 KiB
Python
70 lines
1.8 KiB
Python
from __future__ import annotations
|
|
|
|
import uuid
|
|
from typing import Sequence
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.audit_log import AuditLog
|
|
|
|
|
|
async def log_action(
|
|
db: AsyncSession,
|
|
*,
|
|
study_id: uuid.UUID | None,
|
|
entity_type: str,
|
|
entity_id: uuid.UUID | None,
|
|
action: str,
|
|
detail: str | None,
|
|
operator_id: uuid.UUID,
|
|
operator_role: str,
|
|
auto_commit: bool = True,
|
|
) -> AuditLog:
|
|
log = AuditLog(
|
|
study_id=study_id,
|
|
entity_type=entity_type,
|
|
entity_id=entity_id,
|
|
action=action,
|
|
detail=detail,
|
|
operator_id=operator_id,
|
|
operator_role=operator_role,
|
|
)
|
|
db.add(log)
|
|
if auto_commit:
|
|
await db.commit()
|
|
await db.refresh(log)
|
|
else:
|
|
await db.flush()
|
|
return log
|
|
|
|
|
|
async def list_logs(
|
|
db: AsyncSession,
|
|
*,
|
|
study_id: uuid.UUID,
|
|
entity_type: str | None = None,
|
|
entity_id: uuid.UUID | None = None,
|
|
action: str | None = None,
|
|
operator_id: uuid.UUID | None = None,
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
) -> Sequence[AuditLog]:
|
|
stmt = select(AuditLog).where(AuditLog.study_id == study_id)
|
|
if entity_type:
|
|
stmt = stmt.where(AuditLog.entity_type == entity_type)
|
|
if entity_id:
|
|
stmt = stmt.where(AuditLog.entity_id == entity_id)
|
|
if action:
|
|
stmt = stmt.where(AuditLog.action == action)
|
|
if operator_id:
|
|
stmt = stmt.where(AuditLog.operator_id == operator_id)
|
|
stmt = stmt.order_by(AuditLog.created_at.desc()).offset(skip).limit(limit)
|
|
result = await db.execute(stmt)
|
|
return result.scalars().all()
|
|
|
|
|
|
async def get_log(db: AsyncSession, log_id: uuid.UUID) -> AuditLog | None:
|
|
result = await db.execute(select(AuditLog).where(AuditLog.id == log_id))
|
|
return result.scalar_one_or_none()
|