-
Notifications
You must be signed in to change notification settings - Fork 1
/
main.py
70 lines (58 loc) · 1.76 KB
/
main.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
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from redis_om import get_redis_connection, HashModel, NotFoundError
from fastapi import HTTPException
from dotenv import load_dotenv
import os
load_dotenv()
app = FastAPI()
app.add_middleware(
CORSMiddleware,
allow_origins=os.getenv("ALLOW_ORIGINS", "*").split(","),
allow_methods=os.getenv("ALLOW_METHODS", "*").split(","),
allow_headers=os.getenv("ALLOW_HEADERS", "*").split(","),
)
redis = get_redis_connection(
host=os.getenv("REDIS_HOST", "localhost"),
port=int(os.getenv("REDIS_PORT", 6379)),
db=int(os.getenv("REDIS_DB", 0)),
decode_responses=True
)
class Product(HashModel):
name: str
price: float
stock: int
class Meta:
database = redis
model_key_prefix = "product"
def format_product(pk: str):
product = Product.get(pk)
return {
"id": product.pk,
"name": product.name,
"price": product.price,
"stock": product.stock
}
@app.get("/products")
def get_all_products():
return [format_product(pk) for pk in Product.all_pks()]
@app.get("/products/{pk}")
def get_product(pk: str):
try:
return Product.get(pk)
except NotFoundError:
raise HTTPException(status_code=404, detail=f"Product with pk {pk} not found")
@app.post("/products")
def create_product(product: Product):
return product.save()
@app.delete("/products/{pk}")
def delete_product(pk: str):
try:
product = Product.get(pk)
Product.delete(pk)
return {"message": f"Product with pk {pk} deleted"}
except NotFoundError:
raise HTTPException(status_code=404, detail=f"Product with pk {pk} not found")
@app.get("/")
async def root():
return {"message": "Hello World"}