......................... [100%] 25 passed in 0.60s score: 25/25 (100%)
from fastapi import FastAPI, Request, HTTPException, Header, status, Query
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field
import uvicorn
import os
import secrets
from typing import List, Optional, Dict
app = FastAPI()
# --- Data Models & Storage ---
# In-memory databases
users = {
"alice": "password123",
"bob": "hunter2",
}
# Token mapping: token -> username
tokens: Dict[str, str] = {}
products = {
1: {"id": 1, "name": "Laptop", "price": 999.99, "stock": 10},
2: {"id": 2, "name": "Mouse", "price": 25.50, "stock": 20},
3: {"id": 3, "name": "Keyboard", "price": 75.00, "stock": 15},
}
orders = []
order_id_counter = 1
class LoginRequest(BaseModel):
username: str
password: str
class OrderRequest(BaseModel):
product_id: int
quantity: int = Field(..., gt=0)
# --- Helpers ---
def get_user_from_token(token: str):
username = tokens.get(token)
if not username:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid or missing token"
)
return username
# Custom exception handler to ensure JSON responses for all errors
@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
return JSONResponse(
status_code=exc.status_code,
content={"error": exc.detail},
)
@app.exception_handler(Exception)
async def general_exception_handler(request: Request, exc: Exception):
return JSONResponse(
status_code=500,
content={"error": "Internal server error"},
)
# --- Endpoints ---
@app.post("/login")
async def login(request: Request):
try:
data = await request.json()
except Exception:
return JSONResponse(status_code=400, content={"error": "Invalid JSON body"})
username = data.get("username")
password = data.get("password")
if username is None or password is None:
return JSONResponse(status_code=400, content={"error": "Missing username or password"})
if users.get(username) == password:
token = secrets.token_hex(16)
tokens[token] = username
return {"token": token}
return JSONResponse(status_code=401, content={"error": "Invalid credentials"})
@app.get("/products")
async def get_products(max_price: Optional[str] = Query(None)):
product_list = list(products.values())
if max_price is not None:
try:
price_limit = float(max_price)
product_list = [p for p in product_list if p["price"] <= price_limit]
except ValueError:
# Requirement: "If max_price is absent or not a valid number, return all products."
pass
return product_list
@app.get("/products/{product_id}")
async def get_product(product_id: int):
product = products.get(product_id)
if not product:
raise HTTPException(status_code=404, detail="Product not found")
return product
@app.post("/orders", status_code=201)
async def create_order(request: Request, authorization: Optional[str] = Header(None)):
# Auth check
if not authorization or not authorization.startswith("Bearer "):
raise HTTPException(status_code=401, detail="Missing or invalid token")
token = authorization.split(" ")[1]
username = get_user_from_token(token)
# Body parsing
try:
data = await request.json()
except Exception:
return JSONResponse(status_code=400, content={"error": "Invalid JSON body"})
# Validate fields and types
product_id = data.get("product_id")
quantity = data.get("quantity")
if product_id is None or quantity is None:
return JSONResponse(status_code=400, content={"error": "Missing product_id or quantity"})
if not isinstance(product_id, int) or not isinstance(quantity, int):
return JSONResponse(status_code=400, content={"error": "product_id and quantity must be integers"})
if quantity < 1:
return JSONResponse(status_code=400, content={"error": "quantity must be at least 1"})
# Business Logic
product = products.get(product_id)
if not product:
return JSONResponse(status_code=404, content={"error": "Product not found"})
if product["stock"] < quantity:
return JSONResponse(status_code=409, content={"error": "Insufficient stock"})
# Create order and update stock
global order_id_counter
order = {
"id": order_id_counter,
"product_id": product_id,
"quantity": quantity,
"total": round(product["price"] * quantity, 2),
"username": username
}
orders.append(order)
order_id_counter += 1
product["stock"] -= quantity
# Return order without the username field
return {k: v for k, v in order.items() if k != "username"}
@app.get("/orders")
async def list_orders(authorization: Optional[str] = Header(None)):
if not authorization or not authorization.startswith("Bearer "):
raise HTTPException(status_code=401, detail="Missing or invalid token")
token = authorization.split(" ")[1]
username = get_user_from_token(token)
user_orders = [
{k: v for k, v in o.items() if k != "username"}
for o in orders if o["username"] == username
]
return user_orders
@app.get("/orders/{order_id}")
async def get_order(order_id: int, authorization: Optional[str] = Header(None)):
if not authorization or not authorization.startswith("Bearer "):
raise HTTPException(status_code=401, detail="Missing or invalid token")
token = authorization.split(" ")[1]
username = get_user_from_token(token)
order = next((o for o in orders if o["id"] == order_id), None)
if not order or order["username"] != username:
raise HTTPException(status_code=404, detail="Order not found")
return {k: v for k, v in order.items() if k != "username"}
if __name__ == "__main__":
host = os.environ.get("HOST", "127.0.0.1")
port = int(os.environ.get("PORT", "8000"))
uvicorn.run(app, host=host, port=port)
fastapi uvicorn pydantic
......................... [100%] 25 passed in 0.61s score: 25/25 (100%)
from fastapi import FastAPI, Request, HTTPException, Header, Depends, status
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field
from typing import List, Optional
import uvicorn
import os
import secrets
app = FastAPI()
# --- Data Stores ---
users = {
"alice": "password123",
"bob": "hunter2"
}
products = {
1: {"id": 1, "name": "Laptop", "price": 999.99, "stock": 10},
2: {"id": 2, "name": "Mouse", "price": 25.50, "stock": 20},
3: {"id": 3, "name": "Keyboard", "price": 75.00, "stock": 15},
}
orders = []
tokens = {} # token -> username
# --- Models ---
class LoginRequest(BaseModel):
username: str
password: str
class OrderRequest(BaseModel):
product_id: int
quantity: int = Field(..., ge=1)
# --- Helpers ---
def get_authenticated_user(authorization: Optional[str] = Header(None)):
if not authorization or not authorization.startswith("Bearer "):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Missing or invalid token"
)
token = authorization.split(" ")[1]
username = tokens.get(token)
if not username:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid token"
)
return username
# --- Exception Handlers ---
@app.exception_handler(HTTPException)
async def http_exception_handler(request, exc):
return JSONResponse(
status_code=exc.status_code,
content={"error": exc.detail},
)
@app.exception_handler(Exception)
async def general_exception_handler(request, exc):
return JSONResponse(
status_code=500,
content={"error": "Internal server error"},
)
# --- Routes ---
@app.post("/login")
async def login(request: Request):
try:
body = await request.json()
except Exception:
return JSONResponse(status_code=400, content={"error": "Invalid JSON"})
username = body.get("username")
password = body.get("password")
if username is None or password is None:
return JSONResponse(status_code=400, content={"error": "Missing username or password"})
if users.get(username) == password:
token = secrets.token_hex(16)
tokens[token] = username
return {"token": token}
return JSONResponse(status_code=401, content={"error": "Invalid credentials"})
@app.get("/products")
async def get_products(max_price: Optional[str] = None):
product_list = list(products.values())
if max_price:
try:
price_limit = float(max_price)
product_list = [p for p in product_list if p["price"] <= price_limit]
except ValueError:
pass
return product_list
@app.get("/products/{id}")
async def get_product(id: int):
product = products.get(id)
if not product:
raise HTTPException(status_code=404, detail="Product not found")
return product
@app.post("/orders", status_code=201)
async def create_order(
request: Request,
username: str = Depends(get_authenticated_user)
):
try:
body = await request.json()
# Manual validation to ensure exact error codes as requested
product_id = body.get("product_id")
quantity = body.get("quantity")
if product_id is None or quantity is None:
return JSONResponse(status_code=400, content={"error": "Missing product_id or quantity"})
if not isinstance(product_id, int) or not isinstance(quantity, int):
return JSONResponse(status_code=400, content={"error": "product_id and quantity must be integers"})
if quantity < 1:
return JSONResponse(status_code=400, content={"error": "quantity must be at least 1"})
except Exception:
return JSONResponse(status_code=400, content={"error": "Invalid JSON"})
product = products.get(product_id)
if not product:
return JSONResponse(status_code=404, content={"error": "Product not found"})
if product["stock"] < quantity:
return JSONResponse(status_code=409, content={"error": "Insufficient stock"})
# Process order
product["stock"] -= quantity
order_id = len(orders) + 1
total = product["price"] * quantity
order = {
"id": order_id,
"product_id": product_id,
"quantity": quantity,
"total": total,
"username": username
}
orders.append(order)
# Remove internal username before returning
return {k: v for k, v in order.items() if k != "username"}
@app.get("/orders")
async def get_orders(username: str = Depends(get_authenticated_user)):
user_orders = [
{k: v for k, v in o.items() if k != "username"}
for o in orders if o["username"] == username
]
return user_orders
@app.get("/orders/{id}")
async def get_order(id: int, username: str = Depends(get_authenticated_user)):
# Find order by id
order = next((o for o in orders if o["id"] == id), None)
if not order or order["username"] != username:
raise HTTPException(status_code=404, detail="Order not found")
return {k: v for k, v in order.items() if k != "username"}
if __name__ == "__main__":
host = os.getenv("HOST", "127.0.0.1")
port = int(os.getenv("PORT", "8000"))
uvicorn.run(app, host=host, port=port)
fastapi uvicorn pydantic
......................... [100%] 25 passed in 0.60s score: 25/25 (100%)
from fastapi import FastAPI, Request, HTTPException, Depends, status, Query
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field
from typing import List, Optional
import uvicorn
import os
import secrets
app = FastAPI()
# Data Store
users = {
"alice": "password123",
"bob": "hunter2"
}
products = {
1: {"id": 1, "name": "Laptop", "price": 999.99, "stock": 10},
2: {"id": 2, "name": "Mouse", "price": 25.50, "stock": 50},
3: {"id": 3, "name": "Keyboard", "price": 75.00, "stock": 20},
}
# Token store: {token: username}
tokens = {}
orders = []
order_id_counter = 1
# Models
class LoginRequest(BaseModel):
username: str
password: str
class OrderRequest(BaseModel):
product_id: int
quantity: int = Field(..., gt=0)
# Helpers
def get_current_user(request: Request):
auth_header = request.headers.get("Authorization")
if not auth_header or not auth_header.startswith("Bearer "):
raise HTTPException(status_code=401, detail="Missing or invalid token")
token = auth_header.split(" ")[1]
username = tokens.get(token)
if not username:
raise HTTPException(status_code=401, detail="Invalid token")
return username
# Exception Handlers to ensure JSON error responses
@app.exception_handler(HTTPException)
async def http_exception_handler(request, exc):
return JSONResponse(
status_code=exc.status_code,
content={"error": exc.detail},
)
@app.exception_handler(Exception)
async def general_exception_handler(request, exc):
return JSONResponse(
status_code=500,
content={"error": "Internal server error"},
)
# Endpoints
@app.post("/login")
async def login(request: Request):
try:
body = await request.json()
except Exception:
return JSONResponse(status_code=400, content={"error": "Invalid JSON"})
username = body.get("username")
password = body.get("password")
if username is None or password is None:
return JSONResponse(status_code=400, content={"error": "Missing username or password"})
if users.get(username) == password:
token = secrets.token_hex(16)
tokens[token] = username
return {"token": token}
return JSONResponse(status_code=401, content={"error": "Invalid credentials"})
@app.get("/products")
async def get_products(max_price: Optional[str] = Query(None)):
product_list = list(products.values())
if max_price:
try:
price_val = float(max_price)
product_list = [p for p in product_list if p["price"] <= price_val]
except ValueError:
pass # If not a valid number, return all products as per spec
return product_list
@app.get("/products/{id}")
async def get_product(id: int):
product = products.get(id)
if not product:
raise HTTPException(status_code=404, detail="Product not found")
return product
@app.post("/orders", status_code=201)
async def create_order(request: Request, username: str = Depends(get_current_user)):
try:
body = await request.json()
except Exception:
return JSONResponse(status_code=400, content={"error": "Invalid JSON"})
product_id = body.get("product_id")
quantity = body.get("quantity")
if product_id is None or quantity is None:
return JSONResponse(status_code=400, content={"error": "Missing product_id or quantity"})
if not isinstance(product_id, int) or not isinstance(quantity, int):
return JSONResponse(status_code=400, content={"error": "product_id and quantity must be integers"})
if quantity < 1:
return JSONResponse(status_code=400, content={"error": "quantity must be at least 1"})
product = products.get(product_id)
if not product:
return JSONResponse(status_code=404, content={"error": "Product not found"})
if product["stock"] < quantity:
return JSONResponse(status_code=409, content={"error": "Insufficient stock"})
# Process Order
global order_id_counter
order_id = order_id_counter
order_id_counter += 1
total = product["price"] * quantity
product["stock"] -= quantity
order = {
"id": order_id,
"product_id": product_id,
"quantity": quantity,
"total": total,
"username": username
}
orders.append(order)
# Return order without the username internal field
return {k: v for k, v in order.items() if k != "username"}
@app.get("/orders")
async def list_orders(username: str = Depends(get_current_user)):
user_orders = [o for o in orders if o["username"] == username]
return [{k: v for k, v in o.items() if k != "username"} for o in user_orders]
@app.get("/orders/{id}")
async def get_order(id: int, username: str = Depends(get_current_user)):
order = next((o for o in orders if o["id"] == id), None)
if not order or order["username"] != username:
raise HTTPException(status_code=404, detail="Order not found")
return {k: v for k, v in order.items() if k != "username"}
if __name__ == "__main__":
host = os.environ.get("HOST", "127.0.0.1")
port = int(os.environ.get("PORT", "8000"))
uvicorn.run(app, host=host, port=port)
fastapi uvicorn pydantic
......................... [100%] 25 passed in 0.60s score: 25/25 (100%)
from fastapi import FastAPI, Request, HTTPException, Header, Depends, status
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field
from typing import List, Optional, Dict
import os
import uvicorn
import secrets
app = FastAPI()
# --- Data Models ---
class User(BaseModel):
username: str
password: str
class LoginRequest(BaseModel):
username: str
password: str
class Product(BaseModel):
id: int
name: str
price: float
stock: int
class OrderRequest(BaseModel):
product_id: int
quantity: int = Field(..., ge=1)
class Order(BaseModel):
id: int
product_id: int
quantity: int
total: float
username: str
# --- In-Memory Storage ---
users_db = {
"alice": "password123",
"bob": "hunter2"
}
products_db = {
1: Product(id=1, name="Laptop", price=999.99, stock=10),
2: Product(id=2, name="Mouse", price=25.50, stock=20),
3: Product(id=3, name="Keyboard", price=50.00, stock=15),
}
orders_db: List[Order] = []
tokens_db: Dict[str, str] = {} # token -> username
order_id_counter = 1
# --- Helpers ---
def get_current_user(authorization: Optional[str] = Header(None)):
if not authorization or not authorization.startswith("Bearer "):
raise HTTPException(status_code=401, detail="Missing or invalid token")
token = authorization.split(" ")[1]
username = tokens_db.get(token)
if not username:
raise HTTPException(status_code=401, detail="Invalid token")
return username
def json_error(status_code: int, message: str):
return JSONResponse(
status_code=status_code,
content={"error": message}
)
# --- Endpoints ---
@app.post("/login")
async def login(request: Request):
try:
body = await request.json()
except Exception:
return json_error(400, "Malformed JSON body")
username = body.get("username")
password = body.get("password")
if username is None or password is None:
return json_error(400, "Missing username or password")
if users_db.get(username) == password:
token = secrets.token_hex(16)
tokens_db[token] = username
return {"token": token}
return json_error(401, "Invalid credentials")
@app.get("/products")
async def get_products(max_price: Optional[str] = None):
products = list(products_db.values())
if max_price:
try:
limit = float(max_price)
products = [p for p in products if p.price <= limit]
except ValueError:
pass # Ignore invalid max_price as per spec
return products
@app.get("/products/{product_id}")
async def get_product(product_id: int):
product = products_db.get(product_id)
if not product:
return json_error(404, "Product not found")
return product
@app.post("/orders", status_code=201)
async def create_order(
request: Request,
username: str = Depends(get_current_user)
):
try:
body = await request.json()
except Exception:
return json_error(400, "Malformed JSON body")
product_id = body.get("product_id")
quantity = body.get("quantity")
if product_id is None or quantity is None:
return json_error(400, "Missing product_id or quantity")
if not isinstance(product_id, int) or not isinstance(quantity, int):
return json_error(400, "product_id and quantity must be integers")
if quantity < 1:
return json_error(400, "quantity must be at least 1")
product = products_db.get(product_id)
if not product:
return json_error(404, "Product not found")
if product.stock < quantity:
return json_error(409, "Insufficient stock")
global order_id_counter
total = product.price * quantity
order = Order(
id=order_id_counter,
product_id=product_id,
quantity=quantity,
total=total,
username=username
)
# Update stock
product.stock -= quantity
orders_db.append(order)
order_id_counter += 1
return {
"id": order.id,
"product_id": order.product_id,
"quantity": order.quantity,
"total": order.total
}
@app.get("/orders")
async def get_orders(username: str = Depends(get_current_user)):
user_orders = [
{"id": o.id, "product_id": o.product_id, "quantity": o.quantity, "total": o.total}
for o in orders_db if o.username == username
]
return user_orders
@app.get("/orders/{order_id}")
async def get_order(order_id: int, username: str = Depends(get_current_user)):
order = next((o for o in orders_db if o.id == order_id), None)
if not order or order.username != username:
return json_error(404, "Order not found or access denied")
return {
"id": order.id,
"product_id": order.product_id,
"quantity": order.quantity,
"total": order.total
}
# Exception handler for FastAPI's default 401/404 etc to ensure JSON format
@app.exception_handler(HTTPException)
async def http_exception_handler(request, exc):
return json_error(exc.status_code, exc.detail)
if __name__ == "__main__":
host = os.environ.get("HOST", "127.0.0.1")
port = int(os.environ.get("PORT", "8000"))
uvicorn.run(app, host=host, port=port)
fastapi uvicorn pydantic
......................... [100%] 25 passed in 0.60s score: 25/25 (100%)
from fastapi import FastAPI, Request, HTTPException, Depends, status, Query
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field
from typing import List, Optional, Dict
import uvicorn
import os
import secrets
app = FastAPI()
# --- Data Models ---
class User(BaseModel):
username: str
password: str
class Product(BaseModel):
id: int
name: str
price: float
stock: int
class Order(BaseModel):
id: int
user_id: str
product_id: int
quantity: int
total: float
class OrderRequest(BaseModel):
product_id: int
quantity: int = Field(..., ge=1)
class LoginRequest(BaseModel):
username: str
password: str
# --- In-Memory Store ---
users_db: Dict[str, str] = {
"alice": "password123",
"bob": "hunter2"
}
products_db: Dict[int, Product] = {
1: Product(id=1, name="Laptop", price=999.99, stock=10),
2: Product(id=2, name="Mouse", price=25.50, stock=20),
3: Product(id=3, name="Keyboard", price=50.00, stock=15),
}
orders_db: Dict[int, Order] = {}
tokens_db: Dict[str, str] = {} # token -> username
order_id_counter = 1
# --- Helpers ---
async def get_current_user(request: Request):
auth_header = request.headers.get("Authorization")
if not auth_header or not auth_header.startswith("Bearer "):
raise HTTPException(status_code=401, detail="Missing or invalid token")
token = auth_header.split(" ")[1]
username = tokens_db.get(token)
if not username:
raise HTTPException(status_code=401, detail="Invalid token")
return username
@app.exception_handler(HTTPException)
async def http_exception_handler(request, exc):
return JSONResponse(
status_code=exc.status_code,
content={"error": exc.detail},
)
@app.exception_handler(Exception)
async def general_exception_handler(request, exc):
return JSONResponse(
status_code=400,
content={"error": "Malformed request or internal error"},
)
# --- Endpoints ---
@app.post("/login")
async def login(request: Request):
try:
body = await request.json()
username = body.get("username")
password = body.get("password")
if username is None or password is None:
return JSONResponse(status_code=400, content={"error": "Missing username or password"})
if users_db.get(username) == password:
token = secrets.token_hex(16)
tokens_db[token] = username
return {"token": token}
else:
return JSONResponse(status_code=401, content={"error": "Invalid credentials"})
except Exception:
return JSONResponse(status_code=400, content={"error": "Invalid JSON body"})
@app.get("/products")
async def get_products(max_price: Optional[str] = Query(None)):
product_list = list(products_db.values())
if max_price is not None:
try:
price_limit = float(max_price)
product_list = [p for p in product_list if p.price <= price_limit]
except ValueError:
pass # Ignore invalid max_price as per spec
return product_list
@app.get("/products/{id}")
async def get_product(id: int):
product = products_db.get(id)
if not product:
raise HTTPException(status_code=404, detail="Product not found")
return product
@app.post("/orders", status_code=201)
async def create_order(request: Request, username: str = Depends(get_current_user)):
try:
body = await request.json()
# Manual validation for 400 responses as requested
if "product_id" not in body or "quantity" not in body:
return JSONResponse(status_code=400, content={"error": "Missing product_id or quantity"})
product_id = body["product_id"]
quantity = body["quantity"]
if not isinstance(product_id, int) or not isinstance(quantity, int) or quantity < 1:
return JSONResponse(status_code=400, content={"error": "Invalid product_id or quantity"})
except Exception:
return JSONResponse(status_code=400, content={"error": "Invalid JSON body"})
product = products_db.get(product_id)
if not product:
raise HTTPException(status_code=404, detail="Product not found")
if product.stock < quantity:
return JSONResponse(status_code=409, content={"error": "Insufficient stock"})
global order_id_counter
total = product.price * quantity
order = Order(
id=order_id_counter,
user_id=username,
product_id=product_id,
quantity=quantity,
total=total
)
# Update stock
product.stock -= quantity
orders_db[order_id_counter] = order
order_id_counter += 1
return order
@app.get("/orders")
async def get_orders(username: str = Depends(get_current_user)):
user_orders = [o for o in orders_db.values() if o.user_id == username]
return user_orders
@app.get("/orders/{id}")
async def get_order(id: int, username: str = Depends(get_current_user)):
order = orders_db.get(id)
if not order or order.user_id != username:
raise HTTPException(status_code=404, detail="Order not found")
return order
if __name__ == "__main__":
host = os.environ.get("HOST", "127.0.0.1")
port = int(os.environ.get("PORT", "8000"))
uvicorn.run(app, host=host, port=port)
fastapi uvicorn pydantic