Coverage for server / utilities / model_parser.py: 89%
146 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-10-04 09:33 +0000
1import json
2import random
3import secrets
4import string
5from typing import Dict, Optional, Union
6from bson import ObjectId, datetime
7from fastapi import HTTPException, Request, status
8from server.models.account import (
9 Account,
10 Registration,
11 SubscriberAccount,
12 SubscriberRegistration,
13 UpdatedAccount,
14 UpdatedSubscriberAccount,
15)
16from server.models.classes import ClassModel
17from server.models.users import Student, Teacher
18from pydantic import ValidationError
21def account_parser(account):
22 if "role" in account:
23 if account["role"].strip() == "admin":
24 created_account = Registration.model_validate(account)
25 created_account = Account(**created_account.model_dump())
26 else:
27 raise InvalidRoleException("Invalid Roles")
28 return created_account
29 else:
30 raise InvalidRoleException("Role is required")
33def staff_parser(account):
34 if "role" in account:
35 if account["role"].strip() == "staff":
36 created_account = Registration.model_validate(account)
37 created_account = Account(**created_account.model_dump())
38 else:
39 raise InvalidRoleException("Invalid Roles")
40 return created_account
41 else:
42 raise InvalidRoleException("Role is required")
45async def user_account_parser(account_data):
46 """
47 Parse and validate user registration data.
49 Converts and validates user registration data into the appropriate Pydantic model
50 based on the user's role. Implements comprehensive validation with user-friendly
51 error messages.
53 Parameters
54 ----------
55 account_data : Union[dict, BaseModel]
56 User registration data with the following fields:
58 Required:
59 email : str
60 Valid email address in standard format
61 password : str
62 Secure password meeting all requirements
63 role : str
64 User type, either "student" or "teacher"
65 first_name : str
66 User's first name
67 last_name : str
68 User's last name
70 Optional:
71 middle_name : str | None
72 User's middle name if available
73 contact_person : dict | None
74 Emergency contact information (required for students)
75 department : str | None
76 Academic department (required for teachers)
77 employee_id : str | None
78 School-issued ID (required for teachers)
80 Returns
81 -------
82 Union[Student, Teacher]
83 Validated user model instance matching the specified role:
84 Student:
85 - All base user fields
86 - contact_person : ContactPerson
87 - school_id : ObjectId
89 Teacher:
90 - All base user fields
91 - department : str
92 - employee_id : str
93 - school_id : ObjectId
95 Raises
96 ------
97 HTTPException
98 status_code: 400
99 description: Bad Request
100 content:
101 application/json:
102 examples:
103 invalid_format:
104 summary: Invalid Data Format
105 value: {"detail": "User data must be a dictionary or Pydantic model"}
106 missing_field:
107 summary: Missing Required Field
108 value: {"detail": "First Name is required"}
109 invalid_password:
110 summary: Password Validation Failed
111 value: {"detail": "Invalid password"}
112 field_error:
113 summary: Field Validation Error
114 value: {"detail": "Email: Invalid email format"}
116 InvalidRoleException
117 description: Role Validation Error
118 content:
119 application/json:
120 examples:
121 missing_role:
122 summary: Role Not Provided
123 value: {"detail": "Role is required"}
124 invalid_role:
125 summary: Invalid Role Type
126 value: {"detail": "Invalid Role: admin"}
128 Notes
129 -----
130 Validation Process:
131 1. Role Validation
132 - Verifies role field exists
133 - Validates role is either "student" or "teacher"
135 2. Input Format Normalization
136 - Converts Pydantic models to dictionaries
137 - Creates safe copy of dictionary inputs
138 - Validates input format
140 3. Data Standardization
141 - Normalizes role to lowercase
142 - Sets default account status
143 - Sanitizes input fields
145 4. Model-specific Validation
146 - Maps role to appropriate Pydantic model
147 - Validates all required fields
148 - Performs model-specific validations
150 Error Handling:
151 - Field names in errors are formatted for readability:
152 - Underscores replaced with spaces
153 - Words capitalized
154 Example: "first_name" → "First Name"
156 - Error message types:
157 - Missing fields: "{Field Name} is required"
158 - Password errors: "Invalid password"
159 - Other errors: "{Field Name}: {Error Message}"
161 Examples
162 --------
163 Successful Registration:
164 >>> data = {
165 ... "role": "teacher",
166 ... "email": "teacher@school.edu",
167 ... "password": "SecurePass123!",
168 ... "first_name": "John",
169 ... "last_name": "Doe",
170 ... "department": "Mathematics",
171 ... "employee_id": "T123456"
172 ... }
173 >>> teacher = await user_account_parser(data)
175 Error Cases:
176 >>> # Missing required field
177 >>> data_missing_name = {
178 ... "role": "teacher",
179 ... "email": "teacher@school.edu"
180 ... }
181 >>> await user_account_parser(data_missing_name)
182 HTTPException: First Name is required
184 >>> # Invalid role
185 >>> data_invalid_role = {
186 ... "role": "admin",
187 ... "email": "admin@school.edu"
188 ... }
189 >>> await user_account_parser(data_invalid_role)
190 InvalidRoleException: Invalid Role: admin
191 """
192 # Step 1: Verify required role field exists
193 if "role" not in account_data:
194 raise InvalidRoleException("Role is required")
196 # Step 2: Normalize input data to dictionary format
197 if hasattr(account_data, "model_dump"):
198 # Handle Pydantic model input by converting to dict
199 normalized_account = account_data.model_dump()
200 elif isinstance(account_data, dict):
201 # Create safe copy of dictionary to prevent mutations
202 normalized_account = account_data.copy()
203 else:
204 # Reject invalid input types (neither dict nor Pydantic model)
205 raise HTTPException(
206 status_code=status.HTTP_400_BAD_REQUEST,
207 detail="User data must be a dictionary or Pydantic model"
208 )
210 # Step 3: Standardize data fields
211 normalized_account["role"] = normalized_account["role"].strip().lower() # Normalize role to lowercase
212 normalized_account["status"] = "pending_activation" # Set default account status to pending activation
214 try:
215 # Step 4: Map role to appropriate user model
216 user_model_map = {
217 "student": Student, # Student Pydantic model with contact_person
218 "teacher": Teacher # Teacher Pydantic model with department/employee_id
219 }
221 # Get appropriate model class or raise exception for invalid role
222 selected_model = user_model_map.get(
223 normalized_account["role"],
224 lambda: InvalidRoleException(f"Invalid Role: {normalized_account['role']}")
225 )
227 # Step 5: Create and validate model instance with all fields
228 return selected_model(**normalized_account)
230 except ValidationError as validation_error:
231 # Extract the first validation error for user-friendly message
232 error = validation_error.errors()[0]
234 # Extract location of the error
235 field_location = error.get("loc", [])
237 # Extract field name (last part of the location)
238 field_name = field_location[-1] if field_location else "unknown"
239 formatted_field_name = field_name.replace("_", " ").capitalize()
241 # Check if the field is inside `contact_person`
242 is_inside_contact_person = "contact_person" in field_location
244 # Generate error message based on location
245 if error["type"] == "missing":
246 if is_inside_contact_person:
247 raise HTTPException(
248 status_code=status.HTTP_400_BAD_REQUEST,
249 detail=f"{formatted_field_name} is required in contact person"
250 )
251 else:
252 raise HTTPException(
253 status_code=status.HTTP_400_BAD_REQUEST,
254 detail=f"{formatted_field_name} is required"
255 )
257 # Format other validation errors
258 error_msg = error["msg"].capitalize()
259 raise HTTPException(
260 status_code=status.HTTP_400_BAD_REQUEST,
261 detail=f"{formatted_field_name}: {error_msg}"
262 )
264 except HTTPException as http_error:
265 # Handle specific validation errors from other validation layers
266 error_detail = str(http_error.detail).lower()
268 # Handle email validation errors
269 if "email" in error_detail and "invalid" in error_detail:
270 raise HTTPException(
271 status_code=status.HTTP_400_BAD_REQUEST,
272 detail="Email is invalid"
273 )
275 # Re-raise other HTTP exceptions without modification
276 raise http_error
278 except Exception as error:
279 # Convert any unexpected errors to HTTP 400 responses
280 raise HTTPException(
281 status_code=status.HTTP_400_BAD_REQUEST,
282 detail=str(error)
283 )
286def updated_account_parser(account):
287 if "role" in account:
288 if account["role"].strip() == "admin" or account["role"].strip() == "staff":
289 updated_account = UpdatedAccount.model_validate(account)
290 elif account["role"].strip() == "subscriber":
291 updated_account = UpdatedSubscriberAccount.model_validate(account)
292 else:
293 raise InvalidRoleException("Invalid Role")
294 return updated_account
295 else:
296 raise InvalidRoleException("Role is required")
299class JSONEncoder(json.JSONEncoder):
300 def default(self, o):
301 if isinstance(o, ObjectId):
302 return str(o)
303 if isinstance(o, datetime.datetime):
304 return str(o)
305 return json.JSONEncoder.default(self, o)
308def parse_response(result, exclude_dates=False):
309 result = json.loads(JSONEncoder().encode(result))
310 return generate_response_payload(result, exclude_dates)
313def generate_response_payload(result, exclude_dates):
314 response_payload = []
315 total_count = 0
316 for res in result:
317 question = {}
318 item = {}
319 # total_count = res['total_count']
320 if res["question_type"] == "STAAR":
321 item["grade_level"] = res["grade_level"]
322 item["student_expectations"] = res["student_expectations"]
323 item["category"] = res["category"]
324 item["release_date"] = res["release_date"]
325 item["keywords"] = res["keywords"]
326 elif res["question_type"] == "College Level":
327 item["classification"] = res["classification"]
328 item["test_code"] = res["test_code"]
329 item["keywords"] = res["keywords"]
330 elif res["question_type"] == "Mathworld":
331 item["subject"] = res["subject"]
332 item["topic"] = res["topic"]
333 item["teks_code"] = res["teks_code"]
334 item["category"] = res["category"]
335 item["student_expectations"] = res["student_expectations"]
336 item["keywords"] = res["keywords"]
337 item["difficulty"] = res["difficulty"]
338 item["points"] = res["points"]
339 if item:
340 question["id"] = res["_id"]
341 question["question_type"] = res["question_type"]
342 question["response_type"] = res["response_type"]
343 question["question_content"] = res["question_content"]
344 question["question_img"] = res["question_img"]
345 question["question_status"] = res["question_status"]
347 if not exclude_dates:
348 question["created_by"] = res["created_by"]
349 question["created_at"] = res["created_at"]
350 question["updated_by"] = res["updated_by"]
351 question["updated_at"] = res["updated_at"]
352 question["reviewed_by"] = res["reviewed_by"]
353 question["reviewed_at"] = res["reviewed_at"]
355 question["metadata"] = item
356 question["options"] = res["options"]
357 response_payload.append(question)
358 return response_payload
361MAX_CODE_GENERATION_RETRIES = 10
364async def generate_unique_code(length: int = 6) -> str:
365 """
366 Generate a cryptographically secure unique class code.
368 Args:
369 length: Length of the code (default 6)
371 Returns:
372 Unique alphanumeric class code
374 Raises:
375 RuntimeError: If unable to generate unique code after MAX_CODE_GENERATION_RETRIES
376 """
377 # Use uppercase + digits only for better readability (no l/1, O/0 confusion)
378 characters = string.ascii_uppercase + string.digits
380 for attempt in range(MAX_CODE_GENERATION_RETRIES):
381 code = "".join(secrets.choice(characters) for _ in range(length))
383 # Check if the generated code already exists - FIXED: use correct field name
384 existing_code = await ClassModel.find_one({"class_code": code})
385 if not existing_code:
386 return code
388 raise RuntimeError(
389 f"Failed to generate unique class code after {MAX_CODE_GENERATION_RETRIES} attempts"
390 )
393class InvalidRoleException(Exception):
394 def __init__(self, message="Invalid role"):
395 self.message = message
396 super().__init__(self.message)
399def validate_params(page_num: int, page_size: int):
400 if page_num <= 0:
401 raise HTTPException(
402 status.HTTP_400_BAD_REQUEST,
403 detail="Page number should not be equal or less than to 0",
404 )
406 if page_size <= 0:
407 raise HTTPException(
408 status.HTTP_400_BAD_REQUEST,
409 detail="Page size should not be equal or less than to 0",
410 )
413async def normalize_query_params(request: Request) -> Dict[str, Optional[str]]:
414 params = request.query_params
415 normalized_params = {}
417 for key, value in params.items():
418 normalized_params[key.lower()] = value
420 return normalized_params