-
Notifications
You must be signed in to change notification settings - Fork 0
/
main.py
131 lines (116 loc) · 4.46 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
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
121
122
123
124
125
126
127
128
129
130
131
from typing import Tuple
from fastapi import FastAPI, HTTPException, Depends
from passlib.hash import bcrypt
from fastapi.middleware.cors import CORSMiddleware
from models import *
from auth import get_current_user, authenticate_user, create_token
from db import cursor, mysql_conn
app = FastAPI()
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@app.post("/v1/signup")
async def signup(user: User):
hashed_password = bcrypt.hash(user.password)
cursor.execute(
"INSERT INTO users (email, name, password) VALUES (%s, %s, %s)",
(user.email, user.name, hashed_password),
)
mysql_conn.commit()
return {"message": "User created successfully"}
@app.post("/v1/login")
async def login(user: User):
user = authenticate_user(user.email, user.password)
if not user:
raise HTTPException(status_code=401, detail="Invalid email or password")
token = create_token(user[0])
return {"access_token": token}
@app.post("/v1/order")
async def create_order(order: Order, user: User = Depends(get_current_user)):
cursor.execute(
"INSERT INTO orders (user_id, total_price) VALUES (%s, %s)",
(user[0], order.total_price),
)
order_id = cursor.lastrowid
for food_item in order.food_items:
cursor.execute(
"INSERT INTO order_items (order_id, name, quantity, unit_price) VALUES (%s, %s, %s, %s)",
(order_id, food_item.name, food_item.quantity, food_item.unit_price),
)
mysql_conn.commit()
return {"message": "Order created successfully"}
@app.get("/v1/orders")
async def get_orders(user: User = Depends(get_current_user)):
cursor.execute("SELECT * FROM orders WHERE user_id = %s", (user[0],))
orders = cursor.fetchall()
response = []
for order in orders:
cursor.execute("SELECT * FROM order_items WHERE order_id = %s", (order[0],))
items = cursor.fetchall()
food_items = [
{
"name": item[2],
"quantity": item[3],
"unit_price": item[4]
}
for item in items
]
response.append({
"order_id": order[0],
"food_items": food_items,
"total_price": order[2]
})
return response
@app.get("/v1/order/{order_id}")
async def get_order(order_id: int, user: User = Depends(get_current_user)):
cursor.execute("SELECT * FROM orders WHERE id = %s AND user_id = %s", (order_id, user[0]))
order = cursor.fetchone()
if not order:
raise HTTPException(status_code=404, detail="Order not found")
cursor.execute("SELECT * FROM order_items WHERE order_id = %s", (order[0],))
items = cursor.fetchall()
food_items = [
{
"name": item[2],
"quantity": item[3],
"unit_price": item[4]
}
for item in items
]
return {
"order_id": order[0],
"food_items": food_items,
"total_price": order[2]
}
@app.put("/v1/order/{order_id}")
async def update_order(order_id: int, order: Order, user: Tuple[int] = Depends(get_current_user)):
cursor.execute("SELECT * FROM orders WHERE id = %s AND user_id = %s", (order_id, user[0]))
existing_order = cursor.fetchone()
if not existing_order:
raise HTTPException(status_code=404, detail="Order not found")
cursor.execute("DELETE FROM order_items WHERE order_id = %s", (order_id,))
for food_item in order.food_items:
cursor.execute(
"INSERT INTO order_items (order_id, name, quantity, unit_price) VALUES (%s, %s, %s, %s)",
(order_id, food_item.name, food_item.quantity, food_item.unit_price),
)
cursor.execute(
"UPDATE orders SET total_price = %s WHERE id = %s",
(order.total_price, order_id),
)
mysql_conn.commit()
return {"message": "Order updated successfully"}
@app.delete("/v1/order/{order_id}")
async def delete_order(order_id: int, user: User = Depends(get_current_user)):
cursor.execute("SELECT * FROM orders WHERE id = %s AND user_id = %s", (order_id, user[0]))
existing_order = cursor.fetchone()
if not existing_order:
raise HTTPException(status_code=404, detail="Order not found")
cursor.execute("DELETE FROM order_items WHERE order_id = %s", (order_id,))
cursor.execute("DELETE FROM orders WHERE id = %s", (order_id,))
mysql_conn.commit()
return {"message": "Order deleted successfully"}