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

1import uuid (empty)

2from typing import Any (empty)

3 

4from fastapi import APIRouter, HTTPException (empty)

5from sqlmodel import col, func, select (empty)

6 

7from app.api.deps import CurrentUser, SessionDep (empty)

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

9 

10router = APIRouter(prefix="/items", tags=["items"]) (empty)

11 

12 

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

14def read_items( (empty)

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: 1 ctx1c

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

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

24 statement = ( 1 ctx1c

25 select(Item).order_by(col(Item.created_at).desc()).offset(skip).limit(limit) 

26 ) 

27 items = session.exec(statement).all() 1 ctx1c

28 else: 

29 count_statement = ( 

30 select(func.count()) 

31 .select_from(Item) 

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

33 ) 

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

35 statement = ( 

36 select(Item) 

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

38 .order_by(col(Item.created_at).desc()) 

39 .offset(skip) 

40 .limit(limit) 

41 ) 

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

43 

44 items_public = [ItemPublic.model_validate(item) for item in items] 1 ctx1c

45 return ItemsPublic(data=items_public, count=count) 1 ctx1c

46 

47 

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

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

50 """ 

51 Get item by ID. 

52 """ 

53 item = session.get(Item, id) 3 ctx1fgj

54 if not item: 3 ctx1fgj

55 raise HTTPException(status_code=404, detail="Item not found") 1 ctx1j

56 if not current_user.is_superuser and (item.owner_id != current_user.id): 2 ctx1fg

57 raise HTTPException(status_code=403, detail="Not enough permissions") 1 ctx1g

58 return item 1 ctx1f

59 

60 

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

62def create_item( (empty)

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

64) -> Any: 

65 """ 

66 Create new item. 

67 """ 

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

69 session.add(item) 1 ctx1e

70 session.commit() 1 ctx1e

71 session.refresh(item) 1 ctx1e

72 return item 1 ctx1e

73 

74 

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

76def update_item( (empty)

77 *, 

78 session: SessionDep, 

79 current_user: CurrentUser, 

80 id: uuid.UUID, 

81 item_in: ItemUpdate, 

82) -> Any: 

83 """ 

84 Update an item. 

85 """ 

86 item = session.get(Item, id) 3 ctx1bhk

87 if not item: 3 ctx1bhk

88 raise HTTPException(status_code=404, detail="Item not found") 1 ctx1k

89 if not current_user.is_superuser and (item.owner_id != current_user.id): 2 ctx1bh

90 raise HTTPException(status_code=403, detail="Not enough permissions") 1 ctx1h

91 update_dict = item_in.model_dump(exclude_unset=True) 1 ctx1b

92 item.sqlmodel_update(update_dict) 1 ctx1b

93 session.add(item) 1 ctx1b

94 session.commit() 1 ctx1b

95 session.refresh(item) 1 ctx1b

96 return item 1 ctx1b

97 

98 

99@router.delete("/{id}") (empty)

100def delete_item( (empty)

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

102) -> Message: 

103 """ 

104 Delete an item. 

105 """ 

106 item = session.get(Item, id) 3 ctx1dil

107 if not item: 3 ctx1dil

108 raise HTTPException(status_code=404, detail="Item not found") 1 ctx1l

109 if not current_user.is_superuser and (item.owner_id != current_user.id): 2 ctx1di

110 raise HTTPException(status_code=403, detail="Not enough permissions") 1 ctx1i

111 session.delete(item) 1 ctx1d

112 session.commit() 1 ctx1d

113 return Message(message="Item deleted successfully") 1 ctx1d