Coverage for src/cstlcore/constellations/router.py: 71%

1import uuid 

2 

3from fastapi import APIRouter, Depends, HTTPException 

4from sqlmodel import Session, select 

5 

6from cstlcore.auth.dependencies import get_current_user 

7from cstlcore.constellations.models import ( 

8 Constellation, 

9 ConstellationCreate, 

10 ConstellationPublic, 

11 ConstellationUpdate, 

12) 

13from cstlcore.database.dependencies import get_session 

14from cstlcore.memberships.dependencies import require_admin_access 

15from cstlcore.memberships.models import AccessEnum, ConstellationMembership 

16from cstlcore.users.models import User 

17 

18router = APIRouter() 

19 

20 

21@router.get("/constellations", response_model=list[ConstellationPublic]) 

22async def get_all_constellations(session: Session = Depends(get_session)): 

23 constellations = session.exec(select(Constellation)).all() 

24 return constellations 

25 

26 

27@router.post("/constellations", response_model=ConstellationPublic, status_code=201) 

28async def create_constellation( 

29 constellation: ConstellationCreate, 

30 user: User = Depends(get_current_user), 

31 session: Session = Depends(get_session), 

32): 

33 db_constellation = Constellation.model_validate(constellation) 

34 session.add(db_constellation) 

35 session.commit() 

36 session.refresh(db_constellation) 

37 

38 membership = ConstellationMembership( 

39 user_id=user.id, 

40 constellation_id=db_constellation.id, 

41 access=AccessEnum.OWNER, 

42 ) 

43 session.add(membership) 

44 session.commit() 

45 return db_constellation 

46 

47 

48@router.get("/constellations/{constellation_id}", response_model=ConstellationPublic) 

49async def get_constellation_by_id( 

50 constellation_id: uuid.UUID, 

51 session: Session = Depends(get_session), 

52): 

53 constellation = session.get(Constellation, constellation_id) 

54 if not constellation: 

55 raise HTTPException(status_code=404, detail="Constellation not found") 

56 return constellation 

57 

58 

59@router.put("/constellations/{constellation_id}", response_model=ConstellationPublic) 

60async def update_constellation( 

61 constellation_id: str, 

62 constellation: ConstellationUpdate, 

63 session: Session = Depends(get_session), 

64): 

65 db_constellation = session.get(Constellation, constellation_id) 

66 if not db_constellation: 

67 raise HTTPException(status_code=404, detail="Constellation not found") 

68 constellation_data = constellation.model_dump(exclude_unset=True) 

69 db_constellation.sqlmodel_update(constellation_data) 

70 

71 session.add(db_constellation) 

72 session.commit() 

73 session.refresh(db_constellation) 

74 return db_constellation 

75 

76 

77@router.delete( 

78 "/constellations/{constellation_id}", 

79 status_code=204, 

80) 

81async def delete_constellation( 

82 constellation: Constellation = Depends(require_admin_access), 

83 session: Session = Depends(get_session), 

84 current_user: User = Depends(get_current_user), 

85): 

86 membership = session.get( 

87 ConstellationMembership, (constellation.id, current_user.id) 

88 ) 

89 if not membership or membership.access != AccessEnum.OWNER: 

90 raise HTTPException( 

91 status_code=403, 

92 detail="You do not have permission to delete this constellation", 

93 ) 

94 

95 session.delete(constellation) 

96 session.commit()