Coverage for app\routers\schools.py: 33%

1import secrets 

2from fastapi import APIRouter, Depends, HTTPException, Query 

3from sqlalchemy.orm import Session 

4from typing import List, Optional 

5from .. import db as models 

6from ..schemas import school as school_schemas 

7from ..deps import get_db, get_current_user 

8 

9router = APIRouter() 

10 

11@router.post("/", response_model=school_schemas.School) 

12def create_school(school_in: school_schemas.SchoolCreate, db: Session = Depends(get_db), current_user = Depends(get_current_user)): 

13 # Check if school with name exists 

14 existing_school = db.query(models.School).filter(models.School.name == school_in.name).first() 

15 if existing_school: 

16 raise HTTPException(status_code=400, detail="School with this name already exists") 

17 

18 # Create school 

19 token = secrets.token_urlsafe(32) 

20 school_data = school_in.model_dump(exclude={"initial_plan_code"}) 

21 

22 # Handle dict-like user object 

23 user_id = current_user["id"] if isinstance(current_user, dict) else current_user.id 

24 

25 db_school = models.School(**school_data, token=token, created_by=user_id) 

26 db.add(db_school) 

27 db.commit() 

28 db.refresh(db_school) 

29 

30 # Handle initial plan 

31 if school_in.initial_plan_code: 

32 plan = db.query(models.Plan).filter(models.Plan.code == school_in.initial_plan_code).first() 

33 if plan: 

34 subscription = models.Subscription( 

35 school_id=db_school.id, 

36 plan_id=plan.id, 

37 status="active" 

38 ) 

39 db.add(subscription) 

40 db.commit() 

41 

42 return db_school 

43 

44@router.get("/", response_model=List[school_schemas.School]) 

45def read_schools(skip: int = 0, limit: int = 100, db: Session = Depends(get_db), current_user = Depends(get_current_user)): 

46 schools = db.query(models.School).offset(skip).limit(limit).all() 

47 return schools 

48 

49@router.get("/{school_id}", response_model=school_schemas.School) 

50def read_school(school_id: str, db: Session = Depends(get_db), current_user = Depends(get_current_user)): 

51 school = db.query(models.School).filter(models.School.id == school_id).first() 

52 if not school: 

53 raise HTTPException(status_code=404, detail="School not found") 

54 return school 

55 

56@router.put("/{school_id}", response_model=school_schemas.School) 

57def update_school(school_id: str, school_in: school_schemas.SchoolUpdate, db: Session = Depends(get_db), current_user = Depends(get_current_user)): 

58 school = db.query(models.School).filter(models.School.id == school_id).first() 

59 if not school: 

60 raise HTTPException(status_code=404, detail="School not found") 

61 

62 update_data = school_in.model_dump(exclude_unset=True) 

63 for field, value in update_data.items(): 

64 setattr(school, field, value) 

65 

66 db.add(school) 

67 db.commit() 

68 db.refresh(school) 

69 return school