naman_panwar_gnpl/pragativeer-v2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(dashboard): DPP tags properly dynamic based on user target exam

Browse Source
This commit is contained in:
ectasy663
2026-07-08 12:46:44 +05:30
parent c87b305646
commit 1a6168f724
28 changed files with 853 additions and 304 deletions
Download Patch File Download Diff File Expand all files Collapse all files

386
backend/main.py
View File

@@ -5,7 +5,7 @@ from sqlmodel import Session, select
from contextlib import asynccontextmanager
from typing import List, Dict, Any, Optional
from uuid import UUID
from datetime import datetime, timedelta
from datetime import datetime, timedelta, date
from pydantic import BaseModel
from database import init_db, get_session
@@ -14,13 +14,11 @@ from ai_coach import calculate_sm2
from analytics import generate_parent_report
from billing import generate_mock_order, verify_webhook_signature
from llm import stream_tutor_response
from analytics import generate_parent_report
from predictor import calculate_readiness_score
from growth import generate_referral_link, process_referral_reward
@asynccontextmanager
async def lifespan(app: FastAPI):
# Initialize database tables locally when the container boots up
init_db()
yield
@@ -30,10 +28,9 @@ app = FastAPI(
lifespan=lifespan
)
# Enable immediate communication with your Next.js frontend local environment
app.add_middleware(
CORSMiddleware,
allow_origins=["http://localhost:3000", "http://127.0.0.1:3000"],
allow_origins=["http://localhost:3000", "http://127.0.0.1:3000"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
@@ -46,35 +43,30 @@ def health_check():
# --- ACCESS CONTROL DEPENDENCY ---
def require_tier(allowed_tiers: List[models.TierEnum]):
"""
Dependency to block unauthorized access to premium routes.
"""
def tier_checker(user_id: UUID, session: Session = Depends(get_session)):
user = session.get(models.User, user_id)
if not user:
raise HTTPException(status_code=404, detail="User not found.")
if user.tier not in allowed_tiers:
raise HTTPException(
status_code=403,
status_code=403,
detail=f"Access denied. Requires one of: {[t.value for t in allowed_tiers]}"
)
return user
return tier_checker
# --- BASELINE AUTHENTICATION MOCKS FOR FRONTEND UI VERIFICATION ---
# --- AUTHENTICATION ---
@app.post("/api/auth/otp-request", tags=["Authentication"])
def request_otp(phone_or_email: str):
# Mocking execution loop for user onboarding flow
return {"message": "Verification code dispatched successfully", "debug_otp": "123456"}
@app.post("/api/auth/otp-verify", tags=["Authentication"])
def verify_otp(phone_or_email: str, code: str, session: Session = Depends(get_session)):
# Add your logic matching the registration & profile onboarding flow later
return {"status": "authenticated", "token": "mock_jwt_session_token_string"}
# --- CBT TEST ENVIRONMENT CONTENT DELIVERY ENDPOINTS ---
# --- CBT ENGINE ---
@app.get("/api/tests", response_model=List[models.Test], tags=["CBT Engine"])
def list_active_tests(exam_pattern: Optional[str] = None, type: Optional[str] = None, session: Session = Depends(get_session)):
query = select(models.Test).where(models.Test.is_active == True)
@@ -83,13 +75,11 @@ def list_active_tests(exam_pattern: Optional[str] = None, type: Optional[str] =
if type:
query = query.where(models.Test.type == type)
else:
# If type isn't specified, don't return DPS in regular tests query
query = query.where(models.Test.type != "DPS")
return session.exec(query).all()
@app.get("/api/tests/{test_id}", response_model=models.Test, tags=["CBT Engine"])
def get_test_by_id(test_id: UUID, user_id: UUID, session: Session = Depends(get_session), current_user = Depends(require_tier([models.TierEnum.PRO, models.TierEnum.AI_COACH]))):
def get_test_by_id(test_id: UUID, user_id: UUID, session: Session = Depends(get_session)):
test = session.get(models.Test, test_id)
if not test:
raise HTTPException(status_code=404, detail="Requested examination structure not found.")
@@ -105,114 +95,125 @@ def get_question_by_id(question_id: UUID, session: Session = Depends(get_session
@app.post("/api/attempts/submit", response_model=models.Attempt, tags=["CBT Engine"])
def submit_test_attempt(payload: Dict[str, Any], session: Session = Depends(get_session)):
user_id = payload.get("user_id")
source_type = payload.get("source_type")
source_type_raw = payload.get("source_type", "Test")
source_id = payload.get("source_id")
client_answers = payload.get("answers", []) # List of dicts: {"question_id": "...", "selected_answer": "..."}
client_answers = payload.get("answers", [])
if not user_id or not source_id:
raise HTTPException(status_code=400, detail="Missing required deployment metadata parameters.")
# 0. Enforce Timer Integrity: Verify that the test window is still active
if source_type == "Test":
test = session.get(models.Test, source_id)
if test and test.scheduled_date:
# Removed strict time cutoff for development/testing
pass
# Normalize source_type string to match SourceTypeEnum values
_st_map = {"dps": "DPS", "test": "Test", "microtest": "MicroTest"}
source_type = _st_map.get(str(source_type_raw).lower(), "Test")
# 1. Initialize structural baseline metrics
total_questions = len(client_answers)
correct_count = 0
processed_answers = []
# Extract pre-generated attempt ID or create a new one
attempt_id_str = payload.get("id")
import uuid
attempt_id = uuid.UUID(attempt_id_str) if attempt_id_str else uuid.uuid4()
# 2. Secure Server-Side Evaluation Loop
attempt_id_str = payload.get("id")
import uuid as _uuid
attempt_id = _uuid.UUID(attempt_id_str) if attempt_id_str else _uuid.uuid4()
attempt_record = models.Attempt(
id=attempt_id,
user_id=user_id,
source_type=models.SourceTypeEnum(source_type),
source_id=source_id,
started_at=datetime.utcnow() - timedelta(seconds=payload.get("time_taken", 0)),
submitted_at=datetime.utcnow(),
time_taken=payload.get("time_taken", 0),
answers=[],
score=0,
total_questions=total_questions,
accuracy=0.0,
status="Submitted"
)
session.add(attempt_record)
session.flush()
for item in client_answers:
q_id = item.get("question_id")
selected = item.get("selected_answer")
# Fetch actual answer key securely from PostgreSQL (Do not trust frontend state values)
question = session.get(models.Question, q_id)
is_subjective = item.get("is_subjective", False)
try:
question = session.get(models.Question, q_id)
except Exception:
continue
if not question:
continue
# Check correctness against JSON database configurations
is_correct = question.correct_answer.get("correct") == selected
correct_ans = question.correct_answer
# Support both {"value": "A"} and {"correct": "A"} key formats
correct_value = correct_ans.get("value") or correct_ans.get("correct")
if is_subjective:
# Subjective: award credit for any non-trivial written answer
student_text = str(selected or "").strip()
is_correct = len(student_text) > 5
else:
is_correct = bool(correct_value) and correct_value == selected
if is_correct:
correct_count += 1
processed_answers.append({
"question_id": str(q_id),
"selected_answer": selected,
"is_correct": is_correct,
"is_subjective": is_subjective,
"time_spent": item.get("time_spent", 0)
})
# 3. AI Coach Processing: If the answer is incorrect, extract it to the Mistake queue
if not is_correct:
# Check if this user has failed this specific question before to track execution loops
existing_mistake = session.exec(
select(models.Mistake).where(
models.Mistake.user_id == user_id,
models.Mistake.question_id == q_id
)
).first()
# Track mistakes only for objective wrong answers
if not is_correct and not is_subjective:
try:
existing_mistake = session.exec(
select(models.Mistake).where(
models.Mistake.user_id == user_id,
models.Mistake.question_id == q_id
)
).first()
except Exception:
existing_mistake = None
if existing_mistake:
# User failed it again: reset repetitions, assign low quality score (q=1)
interval, ef = calculate_sm2(q=1, repetitions=existing_mistake.review_count, ease_factor=2.5, previous_interval=1)
existing_mistake.review_count += 1
existing_mistake.next_review = datetime.utcnow() + timedelta(days=interval)
existing_mistake.consecutive_correct = 0
session.add(existing_mistake)
else:
# Brand new mistake: enter into the spaced repetition queue for the first time
new_mistake = models.Mistake(
user_id=user_id,
question_id=q_id,
attempt_id=attempt_id, # Linking back to original source file attempt mapping
error_type=models.ErrorTypeEnum.CONCEPTUAL, # Defaults to conceptual until AI tags it
attempt_id=attempt_id,
error_type=models.ErrorTypeEnum.CONCEPTUAL,
topic=question.topic,
subtopic=question.subtopic,
next_review=datetime.utcnow() + timedelta(days=1), # Schedule for tomorrow review loop
next_review=datetime.utcnow() + timedelta(days=1),
review_count=1,
consecutive_correct=0,
is_mastered=False
)
session.add(new_mistake)
# 4. Save Core Attempt Records
accuracy = (correct_count / total_questions) * 100 if total_questions > 0 else 0.0
attempt_record = models.Attempt(
id=attempt_id,
user_id=user_id,
source_type=source_type,
source_id=source_id,
started_at=datetime.utcnow() - timedelta(seconds=payload.get("time_taken", 0)),
submitted_at=datetime.utcnow(),
time_taken=payload.get("time_taken", 0),
answers=processed_answers,
score=correct_count,
total_questions=total_questions,
accuracy=accuracy,
status="Submitted"
)
attempt_record.answers = processed_answers
attempt_record.score = correct_count
attempt_record.accuracy = accuracy
session.add(attempt_record)
session.commit()
session.refresh(attempt_record)
return attempt_record
# --- AI COACH SPАСED REPETITION ENDPOINTS ---
# --- AI COACH SPACED REPETITION ENDPOINTS ---
@app.get("/api/mistakes/today", tags=["AI Coach"])
def get_todays_mistakes(user_id: UUID, session: Session = Depends(get_session)):
# Retrieve all un-mastered mistakes for the user scheduled on or before today
now = datetime.utcnow()
statement = select(models.Mistake, models.Question).join(
models.Question, models.Mistake.question_id == models.Question.id
@@ -222,61 +223,52 @@ def get_todays_mistakes(user_id: UUID, session: Session = Depends(get_session)):
models.Mistake.next_review <= now
)
results = session.exec(statement).all()
response_data = []
for mistake, question in results:
response_data.append({
"mistake": mistake,
"question": question
})
response_data.append({"mistake": mistake, "question": question})
return response_data
@app.post("/api/mistakes/review", tags=["AI Coach"])
def submit_mistake_review(payload: Dict[str, Any], session: Session = Depends(get_session)):
mistake_id = payload.get("mistake_id")
selected_answer = payload.get("selected_answer")
if not mistake_id or not selected_answer:
raise HTTPException(status_code=400, detail="Missing mistake_id or selected_answer")
mistake = session.get(models.Mistake, mistake_id)
if not mistake:
raise HTTPException(status_code=404, detail="Mistake not found.")
question = session.get(models.Question, mistake.question_id)
is_correct = question.correct_answer.get("correct") == selected_answer
# Calculate intervals
q_value = 4 if is_correct else 1
# Very basic interval calculation for demo
correct_value = question.correct_answer.get("value") or question.correct_answer.get("correct")
is_correct = correct_value == selected_answer
q_value = 4 if is_correct else 1
days_since_creation = (datetime.utcnow() - mistake.created_at).days
prev_interval = max(1, days_since_creation)
interval, ef = calculate_sm2(q=q_value, repetitions=mistake.review_count, ease_factor=2.5, previous_interval=prev_interval)
if is_correct:
mistake.consecutive_correct += 1
if mistake.consecutive_correct >= 3:
mistake.is_mastered = True
else:
mistake.consecutive_correct = 0
mistake.review_count += 1
mistake.next_review = datetime.utcnow() + timedelta(days=interval)
session.add(mistake)
session.commit()
session.refresh(mistake)
return {"mistake": mistake, "is_correct": is_correct, "correct_answer": question.correct_answer.get("correct")}
# --- PARENT SUBSYSTEM ENDPOINTS ---
return {"mistake": mistake, "is_correct": is_correct, "correct_answer": correct_value}
# --- PARENT SUBSYSTEM ---
@app.post("/api/parent/register", response_model=models.Parent, tags=["Parent Subsystem"])
def register_parent(parent_data: models.Parent, session: Session = Depends(get_session)):
# Persist parent baseline identity profile
session.add(parent_data)
session.commit()
session.refresh(parent_data)
@@ -287,26 +279,24 @@ def link_student_to_parent(parent_id: UUID, student_phone_or_email: str, session
parent = session.get(models.Parent, parent_id)
if not parent:
raise HTTPException(status_code=404, detail="Parent account context not found.")
# Locate the target student account by phone or email criteria
student = session.exec(
select(models.User).where(
(models.User.phone == student_phone_or_email) |
(models.User.phone == student_phone_or_email) |
(models.User.email == student_phone_or_email)
)
).first()
if not student:
raise HTTPException(status_code=404, detail="Target student profile could not be verified.")
# Append the student's UUID string to the parent's linked array
current_links = list(parent.linked_students or [])
if str(student.id) not in current_links:
current_links.append(str(student.id))
parent.linked_students = current_links
session.add(parent)
session.commit()
return {"message": "Student successfully linked to parent context.", "linked_student_id": student.id}
@app.get("/api/parent/{parent_id}/dashboard/{student_id}", tags=["Parent Subsystem"])
@@ -314,17 +304,14 @@ def get_parent_dashboard_metrics(parent_id: UUID, student_id: UUID, session: Ses
parent = session.get(models.Parent, parent_id)
if not parent:
raise HTTPException(status_code=404, detail="Unauthorized parent profile access attempt.")
# Security check: Ensure this parent is explicitly authorized to view this student's data
if str(student_id) not in (parent.linked_students or []):
raise HTTPException(status_code=403, detail="Access denied to requested student dashboard data.")
report = generate_parent_report(student_id, session)
return report
# --- BILLING & SUBSCRIPTION ENDPOINTS ---
# --- BILLING & SUBSCRIPTION ---
@app.post("/api/billing/create-order", tags=["Subscription Engine"])
def create_subscription_order(user_id: UUID, plan_tier: str, session: Session = Depends(get_session)):
@@ -332,12 +319,7 @@ def create_subscription_order(user_id: UUID, plan_tier: str, session: Session =
if not user:
raise HTTPException(status_code=404, detail="User not found.")
# Match PRD pricing parameters (amounts in paise)
pricing_map = {
"Pro": 29900, # ₹299/mo
"Al_Coach": 59900 # ₹599/mo
}
pricing_map = {"Pro": 29900, "Al_Coach": 59900}
amount = pricing_map.get(plan_tier)
if not amount:
raise HTTPException(status_code=400, detail="Invalid subscription tier requested.")
@@ -347,34 +329,25 @@ def create_subscription_order(user_id: UUID, plan_tier: str, session: Session =
@app.post("/api/billing/webhook/razorpay", tags=["Subscription Engine"])
async def razorpay_webhook_handler(request: Request, session: Session = Depends(get_session)):
"""
Receives server-to-server notifications when a payment succeeds.
"""
payload_bytes = await request.body()
payload_str = payload_bytes.decode('utf-8')
payload_str = payload_bytes.decode("utf-8")
signature = request.headers.get("X-Razorpay-Signature", "")
# Security Check: Verify signature
if not verify_webhook_signature(payload_str, signature):
raise HTTPException(status_code=400, detail="Invalid webhook signature.")
import json
data = json.loads(payload_str)
# Process the successful payment event
if data.get("event") == "payment.captured":
notes = data["payload"]["payment"]["entity"]["notes"]
user_id = notes.get("user_id")
tier_requested = notes.get("tier_requested")
user = session.get(models.User, user_id)
if user and tier_requested:
# Upgrade the user's tier in the database
user.tier = models.TierEnum(tier_requested)
session.add(user)
# Log the subscription record
from datetime import timedelta
new_sub = models.Subscription(
user_id=user.id,
tier=models.TierEnum(tier_requested),
@@ -387,24 +360,23 @@ async def razorpay_webhook_handler(request: Request, session: Session = Depends(
)
session.add(new_sub)
session.commit()
return {"status": "Webhook processed successfully"}
# --- STUDY PLAN ENDPOINTS ---
from datetime import date
# --- STUDY PLAN ---
@app.get("/api/ai-coach/study-plan/today", tags=["AI Coach"])
def get_todays_study_plan(user_id: UUID, session: Session = Depends(get_session), current_user = Depends(require_tier([models.TierEnum.AI_COACH]))):
def get_todays_study_plan(user_id: UUID, session: Session = Depends(get_session), current_user=Depends(require_tier([models.TierEnum.AI_COACH]))):
today = datetime.utcnow().date()
plan = session.exec(
select(models.StudyPlan)
.where(models.StudyPlan.user_id == user_id)
.where(models.StudyPlan.date == today)
).first()
if not plan:
raise HTTPException(status_code=404, detail="No study plan generated for today yet.")
return plan
class ChatRequest(BaseModel):
@@ -413,32 +385,22 @@ class ChatRequest(BaseModel):
@app.post("/api/ai-coach/chat/stream", tags=["AI Coach"])
async def ai_tutor_chat_stream(request: ChatRequest, session: Session = Depends(get_session)):
"""
Streams the context-aware LLM response back to the client.
"""
user = session.get(models.User, request.user_id)
if not user:
raise HTTPException(status_code=404, detail="Student profile not found.")
if user.tier != models.TierEnum.AI_COACH:
raise HTTPException(status_code=403, detail="Access denied. Gated to AI_COACH tier.")
# Enforce Daily Limit (50 messages) as per PRD A-04
# (In a full production build, you'd increment a Redis counter here)
return StreamingResponse(
stream_tutor_response(request.user_id, request.message, session),
stream_tutor_response(request.user_id, request.message, session),
media_type="text/event-stream"
)
@app.get("/api/ai-coach/readiness/{user_id}", tags=["AI Coach"])
def get_exam_readiness(user_id: UUID, session: Session = Depends(get_session), current_user = Depends(require_tier([models.TierEnum.AI_COACH]))):
"""
Returns the dynamic 0-100 Exam Readiness Score and College Prediction.
"""
def get_exam_readiness(user_id: UUID, session: Session = Depends(get_session), current_user=Depends(require_tier([models.TierEnum.AI_COACH]))):
readiness_data = calculate_readiness_score(user_id, session)
if not readiness_data:
raise HTTPException(status_code=404, detail="Could not calculate readiness score.")
return readiness_data
@app.get("/api/growth/referral/{user_id}", tags=["Viral Growth"])
@@ -446,17 +408,17 @@ def get_referral_dashboard(user_id: UUID, session: Session = Depends(get_session
user = session.get(models.User, user_id)
if not user:
raise HTTPException(status_code=404, detail="User not found.")
link = generate_referral_link(user_id)
return {
"referral_link": link,
"total_referrals": 0, # In production, query a referral tracking table
"total_referrals": 0,
"rewards_earned_days": 0,
"share_message": f"I'm on a {user.streak_count}-day study streak on PragatiVeer! Use my link to get 7 days of the AI Coach for free: {link}"
}
# --- USER CRUD & TEST LISTING ---
# --- USERS ---
@app.get("/api/users/{user_id}", response_model=models.User, tags=["Users"])
def get_user_profile(user_id: UUID, session: Session = Depends(get_session)):
user = session.get(models.User, user_id)
@@ -476,24 +438,18 @@ def update_user_profile(user_id: UUID, payload: UserUpdate, session: Session = D
user = session.get(models.User, user_id)
if not user:
raise HTTPException(status_code=404, detail="User not found")
if payload.name is not None:
user.name = payload.name
if payload.target_exam is not None:
user.target_exam = payload.target_exam
if payload.student_class is not None:
user.student_class = payload.student_class
if payload.school is not None:
user.school = payload.school
if payload.state is not None:
user.state = payload.state
if payload.name is not None: user.name = payload.name
if payload.target_exam is not None: user.target_exam = payload.target_exam
if payload.student_class is not None: user.student_class = payload.student_class
if payload.school is not None: user.school = payload.school
if payload.state is not None: user.state = payload.state
session.add(user)
session.commit()
session.refresh(user)
return user
from fastapi import UploadFile, File
from storage import upload_file
@app.post("/api/users/{user_id}/upload-avatar", tags=["Users"])
@@ -501,31 +457,22 @@ async def upload_user_avatar(user_id: UUID, file: UploadFile = File(...), sessio
user = session.get(models.User, user_id)
if not user:
raise HTTPException(status_code=404, detail="User not found")
contents = await file.read()
file_extension = file.filename.split(".")[-1] if "." in file.filename else "jpg"
object_name = f"avatar_{user_id}.{file_extension}"
public_url = upload_file(contents, object_name, content_type=file.content_type)
if not public_url:
raise HTTPException(status_code=500, detail="Failed to upload image to Minio.")
return {"url": public_url}
@app.get("/api/tests", response_model=List[models.Test], tags=["CBT Engine"])
def list_tests(exam_pattern: Optional[models.ExamEnum] = None, session: Session = Depends(get_session)):
query = select(models.Test)
if exam_pattern is not None:
query = query.where(models.Test.exam_pattern == exam_pattern)
return session.exec(query).all()
return {"url": public_url}
@app.get("/api/users/{user_id}/attempts", tags=["Users"])
def get_user_attempts(user_id: UUID, session: Session = Depends(get_session)):
query = select(models.Attempt).where(models.Attempt.user_id == user_id)
return session.exec(query).all()
# --- ADMIN ROUTES ---
@app.get("/api/admin/tests", tags=["Admin"])
@@ -533,6 +480,28 @@ def get_admin_tests(session: Session = Depends(get_session)):
tests = session.exec(select(models.Test)).all()
return tests
class ManualQuestionInput(BaseModel):
section: str
text_en: str
text_hi: Optional[str] = None
options_en: List[str]
options_hi: Optional[List[str]] = None
correct_answer: str
type: str = "SingleChoice"
topic: str = "General"
difficulty: str = "Medium"
marks: int = 4
negative_marks: int = 1
class ManualTestCreate(BaseModel):
title: str
exam_pattern: str
time_limit: int
scheduled_date: Optional[datetime] = None
end_date: Optional[datetime] = None
admin_id: UUID
questions_json: List[ManualQuestionInput]
@app.post("/api/admin/tests/upload", tags=["Admin"])
async def upload_admin_test(
file: UploadFile = File(...),
@@ -546,14 +515,13 @@ async def upload_admin_test(
import storage
import uuid
from worker import process_pdf_test
pdf_bytes = await file.read()
object_name = f"tests/{uuid.uuid4()}_{file.filename}"
url = storage.upload_file(pdf_bytes, object_name, content_type="application/pdf")
if not url:
raise HTTPException(status_code=500, detail="Failed to upload PDF to MinIO")
# Create inactive test
test_obj = models.Test(
title=title,
type="Mock",
@@ -562,14 +530,13 @@ async def upload_admin_test(
total_time=time_limit,
difficulty="Medium",
scheduled_date=scheduled_date,
is_active=False, # Inactive until processing finishes
is_active=False,
created_by=admin_id
)
session.add(test_obj)
session.commit()
session.refresh(test_obj)
# Trigger async background worker
process_pdf_test.delay(
test_id_str=str(test_obj.id),
object_name=object_name,
@@ -579,14 +546,73 @@ async def upload_admin_test(
return {"status": "processing", "test_id": test_obj.id, "message": "Test is being processed by AI"}
@app.post("/api/admin/tests/create_manual", tags=["Admin"])
def create_manual_test(
data: ManualTestCreate,
session: Session = Depends(get_session)
):
import uuid
# 1. Create the test object
test_obj = models.Test(
title=data.title,
type="Mock",
exam_pattern=models.ExamEnum(data.exam_pattern),
sections={},
total_time=data.time_limit,
difficulty="Medium",
scheduled_date=data.scheduled_date,
end_date=data.end_date,
is_active=True,
created_by=data.admin_id
)
session.add(test_obj)
session.commit()
session.refresh(test_obj)
# 2. Parse and create questions
sections_dict = {}
for q_data in data.questions_json:
q_obj = models.Question(
text_en=q_data.text_en,
text_hi=q_data.text_hi,
options_en=q_data.options_en,
options_hi=q_data.options_hi,
correct_answer=q_data.correct_answer,
type=q_data.type,
topic=q_data.topic,
difficulty=q_data.difficulty,
marks=q_data.marks,
negative_marks=q_data.negative_marks
)
session.add(q_obj)
session.commit()
session.refresh(q_obj)
sec_name = q_data.section
if sec_name not in sections_dict:
sections_dict[sec_name] = {
"name": sec_name,
"questions": [],
"time_limit": None
}
sections_dict[sec_name]["questions"].append(str(q_obj.id))
# 3. Update the test with the sections
test_obj.sections = sections_dict
session.add(test_obj)
session.commit()
return {"message": "Test created successfully", "test_id": test_obj.id}
@app.put("/api/admin/tests/{test_id}/toggle", tags=["Admin"])
def toggle_test_status(test_id: UUID, session: Session = Depends(get_session)):
test_obj = session.get(models.Test, test_id)
if not test_obj:
raise HTTPException(status_code=404, detail="Test not found")
test_obj.is_active = not test_obj.is_active
session.add(test_obj)
session.commit()
return {"status": "success", "is_active": test_obj.is_active}