Coverage for app\routers\audit.py: 97%

1from typing import List, Optional 

2from datetime import datetime 

3from fastapi import APIRouter, Depends, Query 

4from sqlalchemy.orm import Session 

5from pydantic import BaseModel 

6 

7from ..db import SessionLocal, AuditLog, User 

8from ..deps import get_db, get_current_user 

9 

10router = APIRouter() 

11 

12class AuditLogSchema(BaseModel): 

13 id: str 

14 actor_id: str 

15 action: str 

16 entity: str 

17 entity_id: str 

18 before: Optional[dict] = None 

19 after: Optional[dict] = None 

20 created_at: datetime 

21 

22 class Config: 

23 from_attributes = True 

24 

25@router.get("", response_model=List[AuditLogSchema]) 

26def get_audit_logs( 

27 skip: int = 0, 

28 limit: int = 100, 

29 entity: Optional[str] = None, 

30 action: Optional[str] = None, 

31 actor_id: Optional[str] = None, 

32 db: Session = Depends(get_db), 

33 current_user: User = Depends(get_current_user) 

34): 

35 query = db.query(AuditLog) 

36 

37 if entity: 

38 query = query.filter(AuditLog.entity == entity) 

39 if action: 

40 query = query.filter(AuditLog.action == action) 

41 if actor_id: 

42 query = query.filter(AuditLog.actor_id == actor_id) 

43 

44 logs = query.order_by(AuditLog.created_at.desc()).offset(skip).limit(limit).all() 

45 return logs