Coverage for app/api/routes/items.py: 93%

1import uuid 

2from typing import Any 

3 

4from fastapi import APIRouter, HTTPException 

5from sqlmodel import func, select 

6 

7from app.api.deps import CurrentUser, SessionDep 

8from app.models import Item, ItemCreate, ItemPublic, ItemsPublic, ItemUpdate, Message 

9 

10router = APIRouter() 

11 

12 

13@router.get("/", response_model=ItemsPublic) 

14def read_items( 

15 session: SessionDep, current_user: CurrentUser, skip: int = 0, limit: int = 100 

16) -> Any: 

17 """ 

18 Retrieve items. 

19 """ 

20 

21 if current_user.is_superuser: 

22 count_statement = select(func.count()).select_from(Item) 

23 count = session.exec(count_statement).one() 

24 statement = select(Item).offset(skip).limit(limit) 

25 items = session.exec(statement).all() 

26 else: 

27 count_statement = ( 

28 select(func.count()) 

29 .select_from(Item) 

30 .where(Item.owner_id == current_user.id) 

31 ) 

32 count = session.exec(count_statement).one() 

33 statement = ( 

34 select(Item) 

35 .where(Item.owner_id == current_user.id) 

36 .offset(skip) 

37 .limit(limit) 

38 ) 

39 items = session.exec(statement).all() 

40 

41 return ItemsPublic(data=items, count=count) 

42 

43 

44@router.get("/{id}", response_model=ItemPublic) 

45def read_item(session: SessionDep, current_user: CurrentUser, id: uuid.UUID) -> Any: 

46 """ 

47 Get item by ID. 

48 """ 

49 item = session.get(Item, id) 

50 if not item: 

51 raise HTTPException(status_code=404, detail="Item not found") 

52 if not current_user.is_superuser and (item.owner_id != current_user.id): 

53 raise HTTPException(status_code=400, detail="Not enough permissions") 

54 return item 

55 

56 

57@router.post("/", response_model=ItemPublic) 

58def create_item( 

59 *, session: SessionDep, current_user: CurrentUser, item_in: ItemCreate 

60) -> Any: 

61 """ 

62 Create new item. 

63 """ 

64 item = Item.model_validate(item_in, update={"owner_id": current_user.id}) 

65 session.add(item) 

66 session.commit() 

67 session.refresh(item) 

68 return item 

69 

70 

71@router.put("/{id}", response_model=ItemPublic) 

72def update_item( 

73 *, 

74 session: SessionDep, 

75 current_user: CurrentUser, 

76 id: uuid.UUID, 

77 item_in: ItemUpdate, 

78) -> Any: 

79 """ 

80 Update an item. 

81 """ 

82 item = session.get(Item, id) 

83 if not item: 

84 raise HTTPException(status_code=404, detail="Item not found") 

85 if not current_user.is_superuser and (item.owner_id != current_user.id): 

86 raise HTTPException(status_code=400, detail="Not enough permissions") 

87 update_dict = item_in.model_dump(exclude_unset=True) 

88 item.sqlmodel_update(update_dict) 

89 session.add(item) 

90 session.commit() 

91 session.refresh(item) 

92 return item 

93 

94 

95@router.delete("/{id}") 

96def delete_item( 

97 session: SessionDep, current_user: CurrentUser, id: uuid.UUID 

98) -> Message: 

99 """ 

100 Delete an item. 

101 """ 

102 item = session.get(Item, id) 

103 if not item: 

104 raise HTTPException(status_code=404, detail="Item not found") 

105 if not current_user.is_superuser and (item.owner_id != current_user.id): 

106 raise HTTPException(status_code=400, detail="Not enough permissions") 

107 session.delete(item) 

108 session.commit() 

109 return Message(message="Item deleted successfully")