Coverage for app\routers\trust.py: 40%

1from fastapi import APIRouter, Depends, HTTPException 

2from sqlalchemy.orm import Session 

3from typing import List 

4from datetime import datetime, UTC 

5from .. import db as models 

6from ..schemas import trust 

7from ..db import SessionLocal 

8from ..deps import get_db 

9 

10router = APIRouter() 

11 

12@router.post("/logs", response_model=trust.InferenceLog) 

13def create_inference_log(log_in: trust.InferenceLogCreate, db: Session = Depends(get_db)): 

14 """ 

15 Log an AI inference result and record usage for billing. 

16 """ 

17 # 1. Create Log 

18 log = models.InferenceLog(**log_in.model_dump()) 

19 db.add(log) 

20 

21 # 2. Billing Integration 

22 # Check for active subscription 

23 subscription = db.query(models.Subscription).filter( 

24 models.Subscription.school_id == log_in.school_id, 

25 models.Subscription.status == "active" 

26 ).first() 

27 

28 # Record usage 

29 # If no subscription, it still records usage (maybe for pay-as-you-go or audit) 

30 usage = models.UsageRecord( 

31 school_id=log_in.school_id, 

32 subscription_id=subscription.id if subscription else None, 

33 metric_key="ai_inference", 

34 value=1.0, # 1 request count 

35 dimension=log_in.module_id or "general", 

36 recorded_at=datetime.now(UTC) 

37 ) 

38 db.add(usage) 

39 

40 db.commit() 

41 db.refresh(log) 

42 return log 

43 

44@router.get("/logs", response_model=List[trust.InferenceLog]) 

45def get_inference_logs(skip: int = 0, limit: int = 100, school_id: str = None, db: Session = Depends(get_db)): 

46 query = db.query(models.InferenceLog) 

47 if school_id: 

48 query = query.filter(models.InferenceLog.school_id == school_id) 

49 logs = query.order_by(models.InferenceLog.timestamp.desc()).offset(skip).limit(limit).all() 

50 return logs 

51 

52@router.get("/logs/{log_id}", response_model=trust.InferenceLog) 

53def get_inference_log(log_id: str, db: Session = Depends(get_db)): 

54 log = db.query(models.InferenceLog).filter(models.InferenceLog.id == log_id).first() 

55 if not log: 

56 raise HTTPException(status_code=404, detail="Log not found") 

57 return log 

58 

59@router.post("/reviews", response_model=trust.JudgementReview) 

60def create_judgement_review(review_in: trust.JudgementReviewCreate, reviewer_id: str = "current_user_id", db: Session = Depends(get_db)): 

61 # In a real app, reviewer_id would come from the current authenticated user 

62 log = db.query(models.InferenceLog).filter(models.InferenceLog.id == review_in.inference_id).first() 

63 if not log: 

64 raise HTTPException(status_code=404, detail="Inference Log not found") 

65 

66 review = models.JudgementReview(**review_in.model_dump(), reviewer_id=reviewer_id) 

67 db.add(review) 

68 db.commit() 

69 db.refresh(review) 

70 return review