Coverage for app\routers\billing.py: 31%

1from fastapi import APIRouter, Depends, HTTPException, Query 

2from sqlalchemy.orm import Session 

3from typing import List, Any 

4from .. import db as models 

5from ..schemas import billing 

6from ..db import SessionLocal 

7from ..deps import get_db 

8 

9router = APIRouter() 

10 

11@router.post("/plans", response_model=billing.Plan) 

12def create_plan(plan_in: billing.PlanCreate, db: Session = Depends(get_db)): 

13 plan = db.query(models.Plan).filter(models.Plan.code == plan_in.code).first() 

14 if plan: 

15 raise HTTPException(status_code=400, detail="Plan with this code already exists") 

16 plan = models.Plan(**plan_in.model_dump()) 

17 db.add(plan) 

18 db.commit() 

19 db.refresh(plan) 

20 return plan 

21 

22@router.get("/plans", response_model=List[billing.Plan]) 

23def read_plans(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)): 

24 plans = db.query(models.Plan).offset(skip).limit(limit).all() 

25 return plans 

26 

27@router.put("/plans/{plan_id}", response_model=billing.Plan) 

28def update_plan(plan_id: str, plan_in: billing.PlanUpdate, db: Session = Depends(get_db)): 

29 plan = db.query(models.Plan).filter(models.Plan.id == plan_id).first() 

30 if not plan: 

31 raise HTTPException(status_code=404, detail="Plan not found") 

32 

33 update_data = plan_in.model_dump(exclude_unset=True) 

34 for field, value in update_data.items(): 

35 setattr(plan, field, value) 

36 

37 db.add(plan) 

38 db.commit() 

39 db.refresh(plan) 

40 return plan 

41 

42@router.delete("/plans/{plan_id}") 

43def delete_plan(plan_id: str, db: Session = Depends(get_db)): 

44 plan = db.query(models.Plan).filter(models.Plan.id == plan_id).first() 

45 if not plan: 

46 raise HTTPException(status_code=404, detail="Plan not found") 

47 

48 # Check if plan is used in any subscription 

49 sub = db.query(models.Subscription).filter(models.Subscription.plan_id == plan_id).first() 

50 if sub: 

51 raise HTTPException(status_code=400, detail="Cannot delete plan that is in use by subscriptions") 

52 

53 db.delete(plan) 

54 db.commit() 

55 return {"ok": True} 

56 

57@router.post("/subscriptions", response_model=billing.Subscription) 

58def create_subscription(sub_in: billing.SubscriptionCreate, db: Session = Depends(get_db)): 

59 school = db.query(models.School).filter(models.School.id == sub_in.school_id).first() 

60 if not school: 

61 raise HTTPException(status_code=404, detail="School not found") 

62 

63 plan = db.query(models.Plan).filter(models.Plan.id == sub_in.plan_id).first() 

64 if not plan: 

65 raise HTTPException(status_code=404, detail="Plan not found") 

66 

67 # Deactivate existing active subscriptions for this school 

68 existing = db.query(models.Subscription).filter( 

69 models.Subscription.school_id == sub_in.school_id, 

70 models.Subscription.status == "active" 

71 ).all() 

72 for sub in existing: 

73 sub.status = "upgraded" 

74 

75 new_sub = models.Subscription(**sub_in.model_dump()) 

76 db.add(new_sub) 

77 db.commit() 

78 db.refresh(new_sub) 

79 return new_sub 

80 

81@router.get("/subscriptions", response_model=List[billing.Subscription]) 

82def read_subscriptions(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)): 

83 subscriptions = db.query(models.Subscription).offset(skip).limit(limit).all() 

84 return subscriptions 

85 

86@router.put("/subscriptions/{subscription_id}", response_model=billing.Subscription) 

87def update_subscription(subscription_id: str, sub_in: billing.SubscriptionUpdate, db: Session = Depends(get_db)): 

88 sub = db.query(models.Subscription).filter(models.Subscription.id == subscription_id).first() 

89 if not sub: 

90 raise HTTPException(status_code=404, detail="Subscription not found") 

91 

92 update_data = sub_in.model_dump(exclude_unset=True) 

93 for field, value in update_data.items(): 

94 setattr(sub, field, value) 

95 

96 db.add(sub) 

97 db.commit() 

98 db.refresh(sub) 

99 return sub 

100 

101@router.get("/usage", response_model=List[billing.UsageRecord]) 

102def read_all_usage(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)): 

103 usage = db.query(models.UsageRecord).offset(skip).limit(limit).all() 

104 return usage 

105 

106@router.get("/schools/{school_id}/usage", response_model=List[billing.UsageRecord]) 

107def get_usage(school_id: str, db: Session = Depends(get_db)): 

108 usage = db.query(models.UsageRecord).filter(models.UsageRecord.school_id == school_id).all() 

109 return usage