Spaces:
Sleeping
Sleeping
File size: 3,757 Bytes
8067a56 73821b2 8067a56 73821b2 8067a56 73821b2 8067a56 73821b2 8067a56 73821b2 8067a56 73821b2 8067a56 73821b2 8067a56 |
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 |
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from datasets import load_dataset
app = FastAPI()
# Load the datasets from the Hugging Face Hub
customers_dataset = load_dataset("dwb2023/blackbird-customers", split="train")
orders_dataset = load_dataset("dwb2023/blackbird-orders", split="train")
class GetUserInput(BaseModel):
key: str
value: str
class UpdateUserInput(BaseModel):
user_id: str
email: str = None
phone: str = None
class GetOrderByIdInput(BaseModel):
order_id: str
class GetCustomerOrdersInput(BaseModel):
customer_id: str
class CancelOrderInput(BaseModel):
order_id: str
class GetUserInfoInput(BaseModel):
key: str
value: str
@app.post("/get_user")
def get_user(input: GetUserInput):
customer = [c for c in customers_dataset if c[input.key] == input.value]
if customer:
return customer[0]
else:
raise HTTPException(status_code=404, detail=f"Couldn't find a user with {input.key} of {input.value}")
@app.post("/update_user")
def update_user(input: UpdateUserInput):
global customers_dataset
customers_data = customers_dataset.to_list()
user_found = False
for customer in customers_data:
if customer["id"] == input.user_id:
if input.email:
customer["email"] = input.email
if input.phone:
customer["phone"] = input.phone
user_found = True
break
if user_found:
updated_dataset = Dataset.from_list(customers_data)
updated_dataset.push_to_hub("dwb2023/blackbird-customers", split="train")
customers_dataset = updated_dataset # Update the global dataset
return {"message": "User information updated successfully"}
else:
raise HTTPException(status_code=404, detail=f"Couldn't find a user with ID {input.user_id}")
@app.post("/get_order_by_id")
def get_order_by_id(input: GetOrderByIdInput):
order = [o for o in orders_dataset if o["id"] == input.order_id]
if order:
return order[0]
else:
raise HTTPException(status_code=404, detail="Order not found")
@app.post("/get_customer_orders")
def get_customer_orders(input: GetCustomerOrdersInput):
customer_orders = [o for o in orders_dataset if o["customer_id"] == input.customer_id]
return customer_orders
@app.post("/cancel_order")
def cancel_order(input: CancelOrderInput):
global orders_dataset
orders_data = orders_dataset.to_list()
order_found = False
for order in orders_data:
if order["id"] == input.order_id:
if order["status"] == "Processing":
order["status"] = "Cancelled"
order_found = True
break
else:
raise HTTPException(status_code=400, detail="Order has already shipped. Can't cancel it.")
if order_found:
updated_orders_dataset = Dataset.from_list(orders_data)
updated_orders_dataset.push_to_hub("dwb2023/blackbird-orders", split="train")
orders_dataset = updated_orders_dataset # Update the global dataset
return {"status": "Cancelled"}
else:
raise HTTPException(status_code=404, detail="Order not found")
@app.post("/get_user_info")
def get_user_info(input: GetUserInfoInput):
customer = [c for c in customers_dataset if c[input.key] == input.value]
if customer:
customer_id = customer[0]["id"]
customer_orders = [o for o in orders_dataset if o["customer_id"] == customer_id]
return {
"user_info": customer[0],
"orders": customer_orders
}
else:
raise HTTPException(status_code=404, detail=f"Couldn't find a user with {input.key} of {input.value}")
|