34 lines
1.3 KiB
Python
34 lines
1.3 KiB
Python
from datetime import datetime
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from backend.app.models.system import AuditLog
|
|
|
|
|
|
class AuditRepository:
|
|
def create_log(self, session: Session, payload: dict) -> AuditLog:
|
|
log = AuditLog(**payload)
|
|
session.add(log)
|
|
session.flush()
|
|
return log
|
|
|
|
def list_logs(self, session: Session, filters: dict) -> list[AuditLog]:
|
|
stmt = select(AuditLog)
|
|
|
|
if filters.get("biz_type"):
|
|
stmt = stmt.where(AuditLog.biz_type == filters["biz_type"])
|
|
if filters.get("biz_id") is not None:
|
|
stmt = stmt.where(AuditLog.biz_id == filters["biz_id"])
|
|
if filters.get("operator_name"):
|
|
stmt = stmt.where(AuditLog.operator_name.contains(filters["operator_name"]))
|
|
if filters.get("operate_type"):
|
|
stmt = stmt.where(AuditLog.operate_type == filters["operate_type"])
|
|
if filters.get("start_time"):
|
|
stmt = stmt.where(AuditLog.operate_time >= datetime.strptime(filters["start_time"], "%Y-%m-%d %H:%M:%S"))
|
|
if filters.get("end_time"):
|
|
stmt = stmt.where(AuditLog.operate_time <= datetime.strptime(filters["end_time"], "%Y-%m-%d %H:%M:%S"))
|
|
|
|
stmt = stmt.order_by(AuditLog.id.desc())
|
|
return list(session.execute(stmt).scalars())
|