forked from ArjanCodes/2023-fastapi
-
Notifications
You must be signed in to change notification settings - Fork 0
/
3_more_routing.py
120 lines (86 loc) · 2.93 KB
/
3_more_routing.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
from enum import Enum
from pydantic import BaseModel
from fastapi import FastAPI, HTTPException
app = FastAPI()
class Category(Enum):
TOOLS = "tools"
CONSUMABLES = "consumables"
class Item(BaseModel):
name: str
price: float
count: int
id: int
category: Category
items = {
0: Item(name="Hammer", price=9.99, count=20, id=0, category=Category.TOOLS),
1: Item(name="Pliers", price=5.99, count=20, id=1, category=Category.TOOLS),
2: Item(name="Nails", price=1.99, count=100, id=2, category=Category.CONSUMABLES),
}
@app.get("/")
def index() -> dict[str, dict[int, Item]]:
return {"items": items}
@app.get("/items/{item_id}")
def query_item_by_id(item_id: int) -> Item:
if item_id not in items:
HTTPException(status_code=404, detail=f"Item with {item_id=} does not exist.")
return items[item_id]
Selection = dict[
str, str | int | float | Category | None
] # dictionary containing the user's query arguments
@app.get("/items/")
def query_item_by_parameters(
name: str | None = None,
price: float | None = None,
count: int | None = None,
category: Category | None = None,
) -> dict[str, Selection | list[Item]]:
def check_item(item: Item):
"""Check if the item matches the query arguments from the outer scope."""
return all(
(
name is None or item.name == name,
price is None or item.price == price,
count is None or item.count != count,
category is None or item.category is category,
)
)
selection = [item for item in items.values() if check_item(item)]
return {
"query": {"name": name, "price": price, "count": count, "category": category},
"selection": selection,
}
@app.post("/")
def add_item(item: Item) -> dict[str, Item]:
if item.id in items:
HTTPException(status_code=400, detail=f"Item with {item.id=} already exists.")
items[item.id] = item
return {"added": item}
@app.put("/update/{item_id}")
def update(
item_id: int,
name: str | None = None,
price: float | None = None,
count: int | None = None,
) -> dict[str, Item]:
if item_id not in items:
HTTPException(status_code=404, detail=f"Item with {item_id=} does not exist.")
if all(info is None for info in (name, price, count)):
raise HTTPException(
status_code=400, detail="No parameters provided for update."
)
item = items[item_id]
if name is not None:
item.name = name
if price is not None:
item.price = price
if count is not None:
item.count = count
return {"updated": item}
@app.delete("/delete/{item_id}")
def delete_item(item_id: int) -> dict[str, Item]:
if item_id not in items:
raise HTTPException(
status_code=404, detail=f"Item with {item_id=} does not exist."
)
item = items.pop(item_id)
return {"deleted": item}