Coverage for server / services / growthbook / teacher_assignment_quota.py: 71%
112 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-10-04 09:33 +0000
1"""
2Teacher-made assignment creation quota from GrowthBook entitlements.
4Each teacher may create up to 10 assignments **per class** when
5teacher.teacher_made_tests.limit_10 is enabled (unlimited per class on premium).
6"""
8from dataclasses import dataclass
9from typing import Any, Optional
11from bson import ObjectId
12from fastapi import HTTPException, Request, status
14from server.connection.database import db
15from server.services.growthbook.context import TargetingContext, resolve_targeting_context
16from server.services.growthbook.evaluation import feature_evaluation_service
18TEACHER_MADE_ASSIGNMENT_LIMIT = 10
19LIMIT_10_KEY = "teacher.teacher_made_tests.limit_10"
20UNLIMITED_KEY = "teacher.teacher_made_tests.unlimited"
21QUOTA_SCOPE = "per_class"
24def _teacher_id_from_request(request: Request) -> str:
25 user_details = request.state.user_details
26 return str(user_details.get("mongodb_id") or user_details.get("uuid") or "")
29def _created_by_values(teacher_id: str) -> list[Any]:
30 """Match assignments stored with created_by as string or ObjectId."""
31 teacher_id_str = str(teacher_id)
32 values: list[Any] = [teacher_id_str]
33 if ObjectId.is_valid(teacher_id_str):
34 values.append(ObjectId(teacher_id_str))
35 return values
38def _class_id_values(class_id: str) -> list[Any]:
39 class_id_str = str(class_id)
40 values: list[Any] = [class_id_str]
41 if ObjectId.is_valid(class_id_str):
42 values.append(ObjectId(class_id_str))
43 return values
46async def resolve_class_scope(
47 *,
48 class_code: Optional[str] = None,
49 class_id: Optional[str] = None,
50) -> tuple[Optional[str], Optional[str]]:
51 """Return (class_id, class_code) from explicit id or class_code lookup."""
52 if class_id:
53 class_id_str = str(class_id)
54 if class_code:
55 return class_id_str, class_code
56 doc = await db["class_collection"].find_one(
57 {"_id": ObjectId(class_id_str), "deleted": {"$ne": True}}
58 )
59 return class_id_str, doc.get("class_code") if doc else None
61 if class_code:
62 doc = await db["class_collection"].find_one(
63 {"class_code": class_code, "deleted": {"$ne": True}}
64 )
65 if not doc:
66 raise HTTPException(
67 status_code=status.HTTP_404_NOT_FOUND,
68 detail=f"Class with code {class_code} not found",
69 )
70 return str(doc["_id"]), class_code
72 return None, None
75@dataclass(frozen=True)
76class TeacherAssignmentQuota:
77 can_create: bool
78 max_assignments: Optional[int]
79 current_count: int
80 remaining: Optional[int]
81 enabled_feature: Optional[str]
82 scope: str = QUOTA_SCOPE
83 class_id: Optional[str] = None
84 class_code: Optional[str] = None
86 def to_dict(self) -> dict[str, Any]:
87 return {
88 "scope": self.scope,
89 "class_id": self.class_id,
90 "class_code": self.class_code,
91 "can_create": self.can_create,
92 "max_assignments_per_class": self.max_assignments,
93 "max_assignments": self.max_assignments,
94 "current_count": self.current_count,
95 "remaining": self.remaining,
96 "enabled_feature": self.enabled_feature,
97 }
100async def count_teacher_created_assignments(
101 teacher_id: str,
102 *,
103 class_id: Optional[str] = None,
104) -> int:
105 """Count live assignments created by the teacher for one class."""
106 if not teacher_id:
107 return 0
109 query: dict[str, Any] = {
110 "deleted": {"$ne": True},
111 "created_by": {"$in": _created_by_values(teacher_id)},
112 }
113 if class_id:
114 query["assigned_class"] = {"$in": _class_id_values(class_id)}
116 return await db["assignments_collection"].count_documents(query)
119async def is_teacher_made_assignment_access_enabled(ctx: TargetingContext) -> bool:
120 """True when the district plan includes teacher-made assignment entitlements."""
121 if await feature_evaluation_service.is_enabled(UNLIMITED_KEY, ctx):
122 return True
123 return await feature_evaluation_service.is_enabled(LIMIT_10_KEY, ctx)
126async def resolve_teacher_assignment_quota(ctx: TargetingContext) -> tuple[bool, Optional[int], Optional[str]]:
127 """
128 Return (can_create, max_assignments_per_class, enabled_feature_key).
130 max_assignments is None when unlimited.
131 """
132 if await feature_evaluation_service.is_enabled(UNLIMITED_KEY, ctx):
133 return True, None, UNLIMITED_KEY
135 if await feature_evaluation_service.is_enabled(LIMIT_10_KEY, ctx):
136 return True, TEACHER_MADE_ASSIGNMENT_LIMIT, LIMIT_10_KEY
138 return False, 0, None
141async def get_teacher_assignment_quota(
142 ctx: TargetingContext,
143 teacher_id: str,
144 *,
145 class_id: Optional[str] = None,
146 class_code: Optional[str] = None,
147) -> TeacherAssignmentQuota:
148 can_create, max_assignments, enabled_feature = await resolve_teacher_assignment_quota(ctx)
150 if not can_create:
151 return TeacherAssignmentQuota(
152 can_create=False,
153 max_assignments=0,
154 current_count=0,
155 remaining=0,
156 enabled_feature=enabled_feature,
157 class_id=class_id,
158 class_code=class_code,
159 )
161 current_count = 0
162 if class_id:
163 current_count = await count_teacher_created_assignments(
164 teacher_id,
165 class_id=class_id,
166 )
168 if max_assignments is None:
169 return TeacherAssignmentQuota(
170 can_create=True,
171 max_assignments=None,
172 current_count=current_count,
173 remaining=None,
174 enabled_feature=enabled_feature,
175 class_id=class_id,
176 class_code=class_code,
177 )
179 if not class_id:
180 return TeacherAssignmentQuota(
181 can_create=True,
182 max_assignments=max_assignments,
183 current_count=0,
184 remaining=max_assignments,
185 enabled_feature=enabled_feature,
186 class_id=class_id,
187 class_code=class_code,
188 )
190 remaining = max(max_assignments - current_count, 0)
191 return TeacherAssignmentQuota(
192 can_create=current_count < max_assignments,
193 max_assignments=max_assignments,
194 current_count=current_count,
195 remaining=remaining,
196 enabled_feature=enabled_feature,
197 class_id=class_id,
198 class_code=class_code,
199 )
202async def build_teacher_assignment_quota_payload(
203 request: Request,
204 *,
205 class_code: Optional[str] = None,
206 class_id: Optional[str] = None,
207) -> dict[str, Any]:
208 ctx = await resolve_targeting_context(request, require_district=True)
209 teacher_id = _teacher_id_from_request(request)
210 resolved_class_id, resolved_class_code = await resolve_class_scope(
211 class_code=class_code,
212 class_id=class_id,
213 )
214 return (
215 await get_teacher_assignment_quota(
216 ctx,
217 teacher_id,
218 class_id=resolved_class_id,
219 class_code=resolved_class_code,
220 )
221 ).to_dict()
224async def enforce_teacher_assignment_quota(
225 request: Request,
226 assigned_class_ids: Optional[list[Any]] = None,
227) -> None:
228 """Raise 403 when the teacher cannot create another assignment for a target class."""
229 ctx = await resolve_targeting_context(request, require_district=True)
230 teacher_id = _teacher_id_from_request(request)
232 if not assigned_class_ids:
233 quota = await get_teacher_assignment_quota(ctx, teacher_id)
234 if not quota.can_create and quota.max_assignments == 0:
235 raise HTTPException(
236 status_code=status.HTTP_403_FORBIDDEN,
237 detail={
238 "message": "Teacher-made assignment creation is not included in your district plan.",
239 "reason": "teacher_made_assignments_not_enabled",
240 "hint": (
241 f"Enable {LIMIT_10_KEY} or {UNLIMITED_KEY} for this district."
242 ),
243 "quota": quota.to_dict(),
244 },
245 )
246 return
248 seen_classes: set[str] = set()
249 for raw_class_id in assigned_class_ids:
250 class_id = str(raw_class_id)
251 if class_id in seen_classes:
252 continue
253 seen_classes.add(class_id)
255 class_doc = await db["class_collection"].find_one(
256 {"_id": ObjectId(class_id) if ObjectId.is_valid(class_id) else class_id}
257 )
258 class_code = class_doc.get("class_code") if class_doc else None
259 quota = await get_teacher_assignment_quota(
260 ctx,
261 teacher_id,
262 class_id=class_id,
263 class_code=class_code,
264 )
266 if not quota.can_create and quota.max_assignments == 0:
267 raise HTTPException(
268 status_code=status.HTTP_403_FORBIDDEN,
269 detail={
270 "message": "Teacher-made assignment creation is not included in your district plan.",
271 "reason": "teacher_made_assignments_not_enabled",
272 "hint": (
273 f"Enable {LIMIT_10_KEY} or {UNLIMITED_KEY} for this district."
274 ),
275 "quota": quota.to_dict(),
276 },
277 )
279 if quota.max_assignments is not None and quota.current_count >= quota.max_assignments:
280 raise HTTPException(
281 status_code=status.HTTP_403_FORBIDDEN,
282 detail={
283 "message": (
284 f"Teacher-made assignment limit reached for this class "
285 f"({quota.max_assignments} per class)."
286 ),
287 "reason": "teacher_made_assignment_limit_reached",
288 "hint": (
289 "Upgrade to a plan with unlimited teacher-made assignments, "
290 "or delete an existing assignment in this class before creating another."
291 ),
292 "quota": quota.to_dict(),
293 },
294 )
297async def enforce_teacher_made_assignment_access(request: Request) -> None:
298 """Raise 403 when teacher-made assignments are not included in the district plan."""
299 ctx = await resolve_targeting_context(request, require_district=True)
300 if await is_teacher_made_assignment_access_enabled(ctx):
301 return
303 raise HTTPException(
304 status_code=status.HTTP_403_FORBIDDEN,
305 detail={
306 "message": "Teacher-made assignments are not included in your district plan.",
307 "reason": "teacher_made_assignments_not_enabled",
308 "hint": (
309 f"Enable {LIMIT_10_KEY} or {UNLIMITED_KEY} for this district."
310 ),
311 },
312 )