Coverage for app\services\billing_service.py: 22%

1from sqlalchemy.orm import Session 

2from sqlalchemy import func 

3from .. import db as models 

4from datetime import datetime, UTC 

5from typing import Optional, Dict, Any 

6 

7class BillingService: 

8 def __init__(self, db: Session): 

9 self.db = db 

10 

11 def get_active_subscription(self, school_id: str) -> Optional[models.Subscription]: 

12 return self.db.query(models.Subscription).filter( 

13 models.Subscription.school_id == school_id, 

14 models.Subscription.status == "active" 

15 ).first() 

16 

17 def check_limit(self, school_id: str, resource_key: str, current_value: Optional[float] = None, requested_amount: float = 1.0) -> bool: 

18 """ 

19 Check if a school has enough quota for a resource. 

20  

21 Args: 

22 school_id: ID of the school 

23 resource_key: Key in the plan's 'limits' JSON (e.g., 'max_students', 'storage_gb') 

24 current_value: If provided, this value is used as current usage.  

25 If None, the service tries to fetch it from SchoolMetric or UsageRecord. 

26 requested_amount: Amount to add/check. 

27  

28 Returns: 

29 True if allowed, False if limit exceeded. 

30 """ 

31 subscription = self.get_active_subscription(school_id) 

32 if not subscription: 

33 # No active subscription usually means strict limits or no access 

34 # For now, let's assume default strict limits (0) if no plan 

35 return False 

36 

37 plan = subscription.plan 

38 if not plan or not plan.limits: 

39 return True # No limits defined means unlimited? Or 0? Assuming unlimited for now if field missing. 

40 

41 limits = plan.limits 

42 if resource_key not in limits: 

43 return True # Limit not enforced for this resource 

44 

45 limit_value = limits[resource_key] 

46 if limit_value == -1: # Convention for unlimited 

47 return True 

48 

49 # Determine current usage if not provided 

50 if current_value is None: 

51 current_value = self._get_current_usage(school_id, resource_key) 

52 

53 return (current_value + requested_amount) <= limit_value 

54 

55 def _get_current_usage(self, school_id: str, resource_key: str) -> float: 

56 """ 

57 Determine current usage based on resource key. 

58 """ 

59 if resource_key == "max_students" or resource_key == "students": 

60 # Try to get from SchoolMetric 

61 metric = self.db.query(models.SchoolMetric).filter( 

62 models.SchoolMetric.school_id == school_id 

63 ).order_by(models.SchoolMetric.last_seen_at.desc()).first() 

64 if metric: 

65 return float(metric.active_users) 

66 return 0.0 

67 

68 elif resource_key == "storage_gb": 

69 # Try to get from SchoolMetric 

70 metric = self.db.query(models.SchoolMetric).filter( 

71 models.SchoolMetric.school_id == school_id 

72 ).order_by(models.SchoolMetric.last_seen_at.desc()).first() 

73 if metric: 

74 # disk_usage is likely in bytes or MB, let's assume MB and convert to GB?  

75 # Schema says Integer. Let's assume it matches the limit unit for now. 

76 return float(metric.disk_usage) 

77 return 0.0 

78 

79 # Fallback to UsageRecord for cumulative metrics 

80 # (This might need summing up usage for the current billing period) 

81 # For simplicity, returning 0 if unknown 

82 return 0.0 

83 

84 def get_limits_status(self, school_id: str) -> Dict[str, Any]: 

85 """ 

86 Return a summary of limits and usage for a school. 

87 """ 

88 subscription = self.get_active_subscription(school_id) 

89 if not subscription: 

90 return {"status": "no_subscription"} 

91 

92 plan = subscription.plan 

93 limits = plan.limits or {} 

94 

95 status = {} 

96 for key, limit in limits.items(): 

97 usage = self._get_current_usage(school_id, key) 

98 status[key] = { 

99 "limit": limit, 

100 "usage": usage, 

101 "remaining": limit - usage if limit != -1 else "unlimited" 

102 } 

103 

104 return status