43 lines
1.2 KiB
Python
43 lines
1.2 KiB
Python
import json
|
|
from uuid import uuid4
|
|
|
|
from sqlalchemy import asc, select
|
|
|
|
from app.db.session import AsyncSessionLocal
|
|
from app.models.audit_event import AuditEventRecord
|
|
|
|
|
|
async def create_audit_event(
|
|
*,
|
|
scheme_id: str,
|
|
event_type: str,
|
|
object_type: str,
|
|
object_ref: str | None = None,
|
|
details: dict | None = None,
|
|
) -> str:
|
|
audit_event_id = uuid4().hex
|
|
|
|
async with AsyncSessionLocal() as session:
|
|
row = AuditEventRecord(
|
|
audit_event_id=audit_event_id,
|
|
scheme_id=scheme_id,
|
|
event_type=event_type,
|
|
object_type=object_type,
|
|
object_ref=object_ref,
|
|
details_json=json.dumps(details, ensure_ascii=False) if details is not None else None,
|
|
)
|
|
session.add(row)
|
|
await session.commit()
|
|
|
|
return audit_event_id
|
|
|
|
|
|
async def list_audit_events(scheme_id: str) -> list[AuditEventRecord]:
|
|
async with AsyncSessionLocal() as session:
|
|
result = await session.execute(
|
|
select(AuditEventRecord)
|
|
.where(AuditEventRecord.scheme_id == scheme_id)
|
|
.order_by(asc(AuditEventRecord.created_at), asc(AuditEventRecord.id))
|
|
)
|
|
return list(result.scalars().all())
|