Coverage for src/cstlcore/collections/router.py: 93%

1import uuid 

2 

3from fastapi import APIRouter, Depends, HTTPException 

4from sqlmodel import Session, select 

5 

6from cstlcore.collections.models import ( 

7 Collection, 

8 CollectionCreate, 

9 CollectionDetailedPublic, 

10 CollectionPublic, 

11 CollectionUpdate, 

12) 

13from cstlcore.constellations.models import Constellation 

14from cstlcore.database.dependencies import get_session 

15from cstlcore.memberships.dependencies import require_read_access, require_write_access 

16 

17router = APIRouter() 

18 

19 

20@router.get( 

21 "/constellations/{constellation_id}/collections", 

22 response_model=list[CollectionDetailedPublic], 

23) 

24async def get_all_collections( 

25 constellation: Constellation = Depends(require_read_access), 

26): 

27 return constellation.collections 

28 

29 

30@router.get( 

31 "/constellations/{constellation_id}/collections/{collection_id}", 

32 response_model=CollectionDetailedPublic, 

33) 

34async def get_collection( 

35 collection_id: uuid.UUID, 

36 constellation: Constellation = Depends(require_read_access), 

37 session: Session = Depends(get_session), 

38): 

39 db_collection = session.exec( 

40 select(Collection).where( 

41 Collection.id == collection_id, 

42 Collection.constellation_id == constellation.id, 

43 ) 

44 ).first() 

45 if not db_collection: 

46 raise HTTPException(status_code=404, detail="Collection not found") 

47 return db_collection 

48 

49 

50@router.post( 

51 "/constellations/{constellation_id}/collections", 

52 response_model=CollectionPublic, 

53 status_code=201, 

54) 

55async def create_collection( 

56 collection: CollectionCreate, 

57 constellation: Constellation = Depends(require_write_access), 

58 session: Session = Depends(get_session), 

59): 

60 collection_db = Collection.model_validate( 

61 collection, update={"constellation_id": constellation.id} 

62 ) 

63 session.add(collection_db) 

64 session.commit() 

65 session.refresh(collection_db) 

66 return collection_db 

67 

68 

69@router.patch( 

70 "/constellations/{constellation_id}/collections/{collection_id}", 

71 response_model=CollectionPublic, 

72) 

73async def update_collection( 

74 collection_id: uuid.UUID, 

75 collection: CollectionUpdate, 

76 session: Session = Depends(get_session), 

77): 

78 db_collection = session.get(Collection, collection_id) 

79 if not db_collection: 

80 raise HTTPException(status_code=404, detail="Collection not found") 

81 collection_data = collection.model_dump(exclude_unset=True) 

82 db_collection.sqlmodel_update(collection_data) 

83 

84 session.add(db_collection) 

85 session.commit() 

86 session.refresh(db_collection) 

87 return db_collection 

88 

89 

90@router.delete("/constellations/{constellation_id}/collections/{collection_id}") 

91async def delete_collection( 

92 collection_id: uuid.UUID, session: Session = Depends(get_session) 

93): 

94 db_collection = session.get(Collection, collection_id) 

95 if not db_collection: 

96 raise HTTPException(status_code=404, detail="Collection not found") 

97 session.delete(db_collection) 

98 session.commit() 

99 return {"ok": True}