File size: 3,370 Bytes
7c6ffa6 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 | import logging
from typing import NoReturn, TypeVar
from fastapi import HTTPException, status
from sqlalchemy.orm import Session
ModelT = TypeVar("ModelT")
logger = logging.getLogger(__name__)
# ── Student-safe canonical messages ─────────────────────────────────────────
# Keep these terse and actionable. Never reference provider names, file paths,
# internal services, or exception class names.
MSG_DOCUMENT_NOT_PROCESSABLE = (
"This document could not be processed. Please try uploading it again."
)
MSG_AUDIO_GENERATION_FAILED = "Audio generation failed. Please try again."
MSG_VIDEO_RENDER_FAILED = "Video generation failed. Please try again."
MSG_VIDEO_VALIDATION_FAILED = (
"Video settings are invalid. Check your scene plan and try again."
)
MSG_EXTRACTION_FAILED = (
"Questions could not be extracted from this file. Please try again."
)
def student_safe_error(
*,
http_status: int,
student_message: str,
error_code: str,
debug_exc: BaseException | None = None,
) -> NoReturn:
"""Log *debug_exc* server-side (WARNING level) and raise an HTTPException
whose ``detail`` only contains the student-safe *student_message*.
Usage::
except SomeInternalError as exc:
student_safe_error(
http_status=status.HTTP_422_UNPROCESSABLE_ENTITY,
student_message=MSG_AUDIO_GENERATION_FAILED,
error_code="AUDIO_GENERATION_FAILED",
debug_exc=exc,
)
"""
if debug_exc is not None:
logger.warning(
"[%s] %s: %s",
error_code,
type(debug_exc).__name__,
debug_exc,
)
raise HTTPException(
status_code=http_status,
detail={"code": error_code, "message": student_message},
)
def rate_limited_error(
message: str = "You have reached your plan limit for this month. Upgrade to continue.",
) -> NoReturn:
"""Raise HTTP 429 RATE_LIMITED with the standard student-safe error shape.
Use this after ``check_usage_limit`` returns ``(False, reason)`` to block
the request with a consistent, student-friendly message that never leaks
internal details.
"""
raise HTTPException(
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
detail={"code": "RATE_LIMITED", "message": message},
)
def get_or_404(db: Session, model: type[ModelT], object_id: str, label: str) -> ModelT:
instance = db.get(model, object_id)
if instance is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"{label} not found",
)
return instance
def ensure_user_exists(db: Session, user_id: str) -> None:
from app.models.user import User
if db.get(User, user_id) is not None:
return
if user_id != "usr_demo_student":
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="User not found",
)
db.add(
User(
id="usr_demo_student",
name="Demo Student",
email="student@example.com",
role="student",
class_level="Plus Two",
syllabus="Kerala HSE",
preferred_language="English",
)
)
db.commit()
|