......................... [100%] 25 passed in 0.66s score: 25/25 (100%)
import os
import uuid
import secrets
from typing import List, Optional, Dict
from fastapi import FastAPI, Depends, HTTPException, Request, status
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel, Field, ValidationError
# --- Models ---
class User(BaseModel):
username: str
password: str
class LoginRequest(BaseModel):
username: str
password: str
class TokenResponse(BaseModel):
token: str
class Product(BaseModel):
id: int
name: str
price: float
stock: int
class OrderRequest(BaseModel):
product_id: int
quantity: int
class Order(BaseModel):
id: int
product_id: int
quantity: int
total: float
username: str # internal use
class OrderResponse(BaseModel):
id: int
product_id: int
quantity: int
total: float
# --- In-memory Database ---
USERS = {
"alice": "password123",
"bob": "hunter2"
}
PRODUCTS = [
{"id": 1, "name": "Laptop", "price": 999.99, "stock": 10},
{"id": 2, "name": "Mouse", "price": 25.50, "stock": 50},
{"id": 3, "name": "Keyboard", "price": 75.00, "stock": 20},
]
ORDERS = []
ORDER_ID_COUNTER = 1
# Token mapping: token -> username
TOKENS: Dict[str, str] = {}
# --- App Setup ---
app = FastAPI()
security = HTTPBearer(auto_error=False)
# --- Helpers ---
def get_current_user(auth: Optional[HTTPAuthorizationCredentials] = Depends(security)) -> str:
if auth is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail={"error": "Missing authorization header"}
)
token = auth.credentials
username = TOKENS.get(token)
if not username:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail={"error": "Invalid or expired token"}
)
return username
# Custom exception handler for FastAPI/Pydantic validation errors to match requirements
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
return JSONResponse(
status_code=422,
content={"error": "Invalid request parameters"},
)
@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
# Ensure error responses are JSON with an 'error' field
content = {"error": exc.detail} if isinstance(exc.detail, str) else exc.detail
return JSONResponse(
status_code=exc.status_code,
content=content,
)
# --- Endpoints ---
@app.post("/login")
async def login(request: Request):
try:
data = await request.json()
except Exception:
return JSONResponse(status_code=400, content={"error": "Malformed JSON"})
if "username" not in data or "password" not in data:
return JSONResponse(status_code=400, content={"error": "Missing username or password"})
username = data["username"]
password = data["password"]
if USERS.get(username) == password:
token = secrets.token_hex(16)
TOKENS[token] = username
return {"token": token}
else:
return JSONResponse(status_code=401, content={"error": "Invalid credentials"})
@app.get("/products")
async def get_products(max_price: Optional[str] = None):
results = PRODUCTS[:]
if max_price is not None:
try:
m_price = float(max_price)
results = [p for p in results if p["price"] <= m_price]
except ValueError:
pass # If not a valid number, return all products
return results
@app.get("/products/{product_id}")
async def get_product(product_id: int):
product = next((p for p in PRODUCTS if p["id"] == product_id), None)
if product:
return product
raise HTTPException(status_code=404, detail="Product not found")
@app.post("/orders", status_code=201)
async def create_order(request: Request, username: str = Depends(get_current_user)):
try:
data = await request.json()
except Exception:
raise HTTPException(status_code=400, detail="Malformed JSON")
if "product_id" not in data or "quantity" not in data:
raise HTTPException(status_code=400, detail="Missing product_id or quantity")
if not isinstance(data["product_id"], int) or not isinstance(data["quantity"], int):
raise HTTPException(status_code=400, detail="product_id and quantity must be integers")
product_id = data["product_id"]
quantity = data["quantity"]
if quantity < 1:
raise HTTPException(status_code=400, detail="Quantity must be at least 1")
product = next((p for p in PRODUCTS if p["id"] == product_id), None)
if not product:
raise HTTPException(status_code=404, detail="Product not found")
if product["stock"] < quantity:
raise HTTPException(status_code=409, detail="Insufficient stock")
# Create order
global ORDER_ID_COUNTER
total = product["price"] * quantity
new_order = {
"id": ORDER_ID_COUNTER,
"product_id": product_id,
"quantity": quantity,
"total": total,
"username": username
}
# Update stock
product["stock"] -= quantity
ORDERS.append(new_order)
ORDER_ID_COUNTER += 1
# Return order without username
return {
"id": new_order["id"],
"product_id": new_order["product_id"],
"quantity": new_order["quantity"],
"total": new_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 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 if o["id"] == order_id), None)
if not order or order["username"] != username:
raise HTTPException(status_code=404, detail="Order not found")
return {
"id": order["id"],
"product_id": order["product_id"],
"quantity": order["quantity"],
"total": order["total"]
}
if __name__ == "__main__":
import uvicorn
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
......................... [100%] 25 passed in 0.61s score: 25/25 (100%)
import os
import uuid
from typing import List, Optional, Dict
from fastapi import FastAPI, Depends, HTTPException, Request, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field, validator
# --- Models ---
class User(BaseModel):
username: str
password: str
class LoginRequest(BaseModel):
username: str
password: str
class LoginResponse(BaseModel):
token: str
class Product(BaseModel):
id: int
name: str
price: float
stock: int
class OrderRequest(BaseModel):
product_id: int
quantity: int
@validator('quantity')
def quantity_must_be_positive(cls, v):
if v < 1:
raise ValueError('quantity must be at least 1')
return v
class Order(BaseModel):
id: int
product_id: int
quantity: int
total: float
username: str # Internal use to track owner
class OrderResponse(BaseModel):
id: int
product_id: int
quantity: int
total: float
# --- In-Memory Data ---
users_db = {
"alice": "password123",
"bob": "hunter2"
}
# tokens mapping: token -> username
tokens_db: Dict[str, str] = {}
products_db = [
{"id": 1, "name": "Laptop", "price": 999.99, "stock": 10},
{"id": 2, "name": "Mouse", "price": 25.50, "stock": 50},
{"id": 3, "name": "Keyboard", "price": 75.00, "stock": 15},
]
orders_db = []
order_id_counter = 1
# --- App Setup ---
app = FastAPI()
security = HTTPBearer(auto_error=False)
# --- Helpers ---
def get_current_user(auth: Optional[HTTPAuthorizationCredentials] = Depends(security)) -> str:
if auth is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail={"error": "Missing authorization header"}
)
token = auth.credentials
if token not in tokens_db:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail={"error": "Invalid or expired token"}
)
return tokens_db[token]
# --- Custom Exception Handlers for consistent JSON error responses ---
@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
# If detail is already a dict (from our custom raises), return it.
# Otherwise, wrap it in an error object.
content = exc.detail
if isinstance(content, dict):
return JSONResponse(
status_code=exc.status_code,
content=content
)
return JSONResponse(
status_code=exc.status_code,
content={"error": content}
)
@app.exception_handler(Exception)
async def general_exception_handler(request: Request, exc: Exception):
return JSONResponse(
status_code=500,
content={"error": str(exc)}
)
# --- Endpoints ---
@app.post("/login", response_model=LoginResponse, status_code=200)
async def login(request: Request):
# We handle manual parsing to match the "Malformed body" requirement exactly
try:
body = await request.json()
except Exception:
raise HTTPException(status_code=400, detail={"error": "Malformed JSON body"})
if "username" not in body or "password" not in body:
raise HTTPException(status_code=400, detail={"error": "Missing username or password"})
username = body["username"]
password = body["password"]
if username in users_db and users_db[username] == password:
token = str(uuid.uuid4())
tokens_db[token] = username
return {"token": token}
else:
raise HTTPException(status_code=401, detail={"error": "Invalid credentials"})
@app.get("/products", response_model=List[Product])
async def get_products(max_price: Optional[str] = None):
filtered_products = products_db
if max_price is not None:
try:
mp = float(max_price)
filtered_products = [p for p in products_db if p["price"] <= mp]
except ValueError:
# If max_price is not a valid number, return all products.
pass
return filtered_products
@app.get("/products/{product_id}", response_model=Product)
async def get_product(product_id: int):
product = next((p for p in products_db if p["id"] == product_id), None)
if not product:
raise HTTPException(status_code=404, detail={"error": "Product not found"})
return product
@app.post("/orders", response_model=OrderResponse, status_code=201)
async def create_order(request: Request, username: str = Depends(get_current_user)):
global order_id_counter
# Parse body manually to satisfy specific error requirements (malformed JSON, missing fields, non-int)
try:
body = await request.json()
except Exception:
raise HTTPException(status_code=400, detail={"error": "Malformed JSON body"})
if "product_id" not in body or "quantity" not in body:
raise HTTPException(status_code=400, detail={"error": "Missing product_id or quantity"})
try:
product_id = int(body["product_id"])
quantity = int(body["quantity"])
except (ValueError, TypeError):
raise HTTPException(status_code=400, detail={"error": "product_id and quantity must be integers"})
if quantity < 1:
raise HTTPException(status_code=400, detail={"error": "quantity must be at least 1"})
# Find product
product = next((p for p in products_db if p["id"] == product_id), None)
if not product:
raise HTTPException(status_code=404, detail={"error": "Product not found"})
# Check stock
if product["stock"] < quantity:
raise HTTPException(status_code=409, detail={"error": "Insufficient stock"})
# Create order
total = product["price"] * quantity
new_order = {
"id": order_id_counter,
"product_id": product_id,
"quantity": quantity,
"total": total,
"username": username
}
# Update stock and state
product["stock"] -= quantity
orders_db.append(new_order)
order_id_counter += 1
return {
"id": new_order["id"],
"product_id": new_order["product_id"],
"quantity": new_order["quantity"],
"total": new_order["total"]
}
@app.get("/orders", response_model=List[OrderResponse])
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}", response_model=OrderResponse)
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:
raise HTTPException(status_code=404, detail={"error": "Order not found"})
return {
"id": order["id"],
"product_id": order["product_id"],
"quantity": order["quantity"],
"total": order["total"]
}
if __name__ == "__main__":
import uvicorn
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
......................... [100%] 25 passed in 0.60s score: 25/25 (100%)
import os
import uuid
from typing import List, Optional, Dict
from fastapi import FastAPI, HTTPException, Depends, Request, status
from fastapi.responses import JSONResponse
from pydantic import BaseModel, ValidationError
import uvicorn
# --- Data Models ---
class User:
def __init__(self, username: str, password: str):
self.username = username
self.password = password
class Product:
def __init__(self, id: int, name: str, price: float, stock: int):
self.id = id
self.name = name
self.price = price
self.stock = stock
class Order:
def __init__(self, id: int, username: str, product_id: int, quantity: int, total: float):
self.id = id
self.username = username
self.product_id = product_id
self.quantity = quantity
self.total = total
# --- In-Memory Database ---
USERS = {
"alice": User("alice", "password123"),
"bob": User("bob", "hunter2")
}
PRODUCTS = [
Product(1, "Laptop", 999.99, 10),
Product(2, "Mouse", 25.50, 50),
Product(3, "Keyboard", 75.00, 15),
]
ORDERS: List[Order] = []
TOKENS: Dict[str, str] = {} # token -> username
ORDER_ID_COUNTER = 1
# --- Schemas for API ---
class LoginRequest(BaseModel):
username: str
password: str
class OrderCreateRequest(BaseModel):
product_id: int
quantity: int
# --- FastAPI App Setup ---
app = FastAPI()
# --- Helper Functions & Dependencies ---
async def get_current_user(request: Request) -> str:
auth_header = request.headers.get("Authorization")
if not auth_header or not auth_header.startswith("Bearer "):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail={"error": "Missing or invalid token"}
)
token = auth_header.split(" ")[1]
username = TOKENS.get(token)
if not username:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail={"error": "Invalid token"}
)
return username
@app.exception_handler(HTTPException)
async def custom_http_exception_handler(request: Request, exc: HTTPException):
# Ensure error responses are JSON with an 'error' field
content = {"error": exc.detail}
if isinstance(exc.detail, dict) and "error" in exc.detail:
content = exc.detail
return JSONResponse(
status_code=exc.status_code,
content=content
)
# --- Endpoints ---
@app.post("/login")
async def login(request: Request):
# We use request.json() manually to handle malformed JSON as requested
try:
body = await request.json()
except Exception:
return JSONResponse(status_code=400, content={"error": "Malformed JSON body"})
if "username" not in body or "password" not in body:
return JSONResponse(status_code=400, content={"error": "Missing username or password"})
username = body["username"]
password = body["password"]
user = USERS.get(username)
if user and user.password == password:
token = str(uuid.uuid4())
TOKENS[token] = username
return {"token": token}
else:
return JSONResponse(status_code=401, content={"error": "Invalid credentials"})
@app.get("/products")
async def get_products(max_price: Optional[str] = None):
results = []
try:
limit = None
if max_price is not None:
limit = float(max_price)
except ValueError:
# If max_price is not a valid number, return all products as per requirements
pass
for p in PRODUCTS:
if limit is None or p.price <= limit:
results.append({
"id": p.id,
"name": p.name,
"price": p.price,
"stock": p.stock
})
return results
@app.get("/products/{product_id}")
async def get_product(product_id: int):
product = next((p for p in PRODUCTS if p.id == product_id), None)
if not product:
raise HTTPException(status_code=404, detail={"error": "Product not found"})
return {
"id": product.id,
"name": product.name,
"price": product.price,
"stock": product.stock
}
@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": "Malformed JSON body"})
# Validate schema manually for more precise control over error messages as per requirements
if "product_id" not in body or "quantity" not in body:
return JSONResponse(status_code=400, content={"error": "Missing product_id or quantity"})
if not isinstance(body["product_id"], int) or not isinstance(body["quantity"], int):
return JSONResponse(status_code=400, content={"error": "product_id and quantity must be integers"})
if body["quantity"] < 1:
return JSONResponse(status_code=400, content={"error": "quantity must be at least 1"})
product_id = body["product_id"]
quantity = body["quantity"]
product = next((p for p in PRODUCTS if p.id == product_id), None)
if not product:
raise HTTPException(status_code=404, detail={"error": "Product not found"})
if product.stock < quantity:
raise HTTPException(status_code=409, detail={"error": "Insufficient stock"})
# Process order
global ORDER_ID_COUNTER
total = product.price * quantity
order = Order(ORDER_ID_COUNTER, username, product_id, quantity, total)
# Update stock and save order
product.stock -= quantity
ORDERS.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 list_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 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 if o.id == order_id), None)
# If order doesn't exist OR belongs to another user, return 404 (to not reveal existence)
if not order or order.username != username:
raise HTTPException(status_code=404, detail={"error": "Order not found"})
return {
"id": order.id,
"product_id": order.product_id,
"quantity": order.quantity,
"total": order.total
}
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
.........F.F.......F..... [100%]
=================================== FAILURES ===================================
_________________________ test_create_order_and_total __________________________
base_url = 'http://127.0.0.1:48185'
alice_token = '58ec2c52-6f90-46c4-bf27-d101b36a3ce5'
def test_create_order_and_total(base_url, alice_token):
# discover product 1's price to validate total
_, prod, _ = _request("GET", f"{base_url}/products/1")
price = prod["price"]
status, order, raw = _request("POST", f"{base_url}/orders", token=alice_token,
body={"product_id": 1, "quantity": 3})
> assert status == 201, f"POST /orders -> {status} ({raw})"
E AssertionError: POST /orders -> 200 ({"id":1,"product_id":1,"quantity":3,"total":2999.9700000000003})
E assert 200 == 201
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:223: AssertionError
___________________________ test_orders_are_per_user ___________________________
base_url = 'http://127.0.0.1:48185'
alice_token = '58ec2c52-6f90-46c4-bf27-d101b36a3ce5'
def test_orders_are_per_user(base_url, alice_token):
# alice creates an order
status, _, raw = _request("POST", f"{base_url}/orders", token=alice_token,
body={"product_id": 2, "quantity": 1})
> assert status == 201, f"alice POST /orders -> {status} ({raw})"
E AssertionError: alice POST /orders -> 200 ({"id":2,"product_id":2,"quantity":1,"total":25.5})
E assert 200 == 201
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:240: AssertionError
_________________________ test_order_decrements_stock __________________________
base_url = 'http://127.0.0.1:48185'
alice_token = '58ec2c52-6f90-46c4-bf27-d101b36a3ce5'
def test_order_decrements_stock(base_url, alice_token):
_, before, _ = _request("GET", f"{base_url}/products/3")
start = before["stock"]
status, _, raw = _request("POST", f"{base_url}/orders", token=alice_token,
body={"product_id": 3, "quantity": 2})
> assert status == 201, f"POST /orders -> {status} ({raw})"
E AssertionError: POST /orders -> 200 ({"id":3,"product_id":3,"quantity":2,"total":150.0})
E assert 200 == 201
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:307: AssertionError
=========================== short test summary info ============================
FAILED ../../home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py::test_create_order_and_total
FAILED ../../home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py::test_orders_are_per_user
FAILED ../../home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py::test_order_decrements_stock
3 failed, 22 passed in 0.62s
score: 22/25 (88%)
import os
import uuid
from typing import List, Optional, Dict
from fastapi import FastAPI, Depends, HTTPException, status, Request, Query
from fastapi.responses import JSONResponse
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel
# --- Models ---
class User(BaseModel):
username: str
password: str
class LoginRequest(BaseModel):
username: str
password: str
class TokenResponse(BaseModel):
token: str
class Product(BaseModel):
id: int
name: str
price: float
stock: int
class OrderRequest(BaseModel):
product_id: int
quantity: int
class Order(BaseModel):
id: int
product_id: int
quantity: int
total: float
username: str # Internal field to track ownership
# --- In-Memory Data Store ---
class DataStore:
def __init__(self):
self.users = {
"alice": "password123",
"bob": "hunter2"
}
self.products = [
Product(id=1, name="Laptop", price=999.99, stock=10),
Product(id=2, name="Mouse", price=25.50, stock=50),
Product(id=3, name="Keyboard", price=75.00, stock=15),
]
self.orders: List[Order] = []
self.tokens: Dict[str, str] = {} # token -> username
self.order_id_counter = 1
def create_token(self, username: str) -> str:
token = str(uuid.uuid4())
self.tokens[token] = username
return token
def get_user_from_token(self, token: str) -> Optional[str]:
return self.tokens.get(token)
store = DataStore()
# --- FastAPI Setup ---
app = FastAPI()
security = HTTPBearer(auto_error=False)
# --- Error Handling Helpers ---
def error_response(status_code: int, message: str):
return JSONResponse(
status_code=status_code,
content={"error": message}
)
# --- Authentication Dependency ---
async def get_current_user(auth: Optional[HTTPAuthorizationCredentials] = Depends(security)) -> str:
if not auth:
raise HTTPException(status_code=401, detail="Missing token")
username = store.get_user_from_token(auth.credentials)
if not username:
# Instead of raising HTTPException (which returns 422/403 in some configs),
# we'll return a custom error response to ensure 401 with JSON body.
# However, FastAPI Depends handles exceptions. Let's check if we can use a custom exception.
# To keep it simple, we will raise a custom exception and use an exception handler.
raise AuthenticationError("Invalid token")
return username
class AuthenticationError(Exception):
def __init__(self, message: str):
self.message = message
@app.exception_handler(AuthenticationError)
async def auth_exception_handler(request: Request, exc: AuthenticationError):
return error_response(401, exc.message)
# --- Routes ---
@app.post("/login")
async def login(request: Request):
try:
body = await request.json()
except Exception:
return error_response(400, "Malformed JSON body")
if not isinstance(body, dict) or "username" not in body or "password" not in body:
return error_response(400, "Missing username or password")
username = body.get("username")
password = body.get("password")
if store.users.get(username) == password:
token = store.create_token(username)
return {"token": token}
else:
return error_response(401, "Invalid credentials")
@app.get("/products")
async def get_products(request: Request):
max_price_raw = request.query_params.get("max_price")
results = store.products
if max_price_raw is not None:
try:
max_price = float(max_price_raw)
results = [p for p in store.products if p.price <= max_price]
except ValueError:
# If not a valid number, return all products
results = store.products
return results
@app.get("/products/{product_id}")
async def get_product(product_id: int):
for p in store.products:
if p.id == product_id:
return p
return error_response(404, "Product not found")
@app.post("/orders")
async def create_order(request: Request, username: str = Depends(get_current_user)):
try:
body = await request.json()
except Exception:
return error_response(400, "Malformed JSON body")
if not isinstance(body, dict) or "product_id" not in body or "quantity" not in body:
return error_response(400, "Missing product_id or quantity")
try:
product_id = int(body["product_id"])
quantity = int(body["quantity"])
except (ValueError, TypeError):
return error_response(400, "product_id and quantity must be integers")
if quantity < 1:
return error_response(400, "quantity must be at least 1")
# Find product
product = next((p for p in store.products if p.id == product_id), None)
if not product:
return error_response(404, "Product not found")
if product.stock < quantity:
return error_response(409, "Insufficient stock")
# Process order
product.stock -= quantity
new_order = Order(
id=store.order_id_counter,
product_id=product_id,
quantity=quantity,
total=product.price * quantity,
username=username
)
store.orders.append(new_order)
store.order_id_counter += 1
return {
"id": new_order.id,
"product_id": new_order.product_id,
"quantity": new_order.quantity,
"total": new_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 store.orders 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 store.orders if o.id == order_id), None)
if not order or order.username != username:
return error_response(404, "Order not found")
return {
"id": order.id,
"product_id": order.product_id,
"quantity": order.quantity,
"total": order.total
}
# --- Main ---
if __name__ == "__main__":
import uvicorn
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
......................... [100%] 25 passed in 0.61s score: 25/25 (100%)
import os
import uuid
from typing import List, Optional
from fastapi import FastAPI, Depends, HTTPException, status, Request
from fastapi.responses import JSONResponse
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel, Field, ValidationError
# --- Models ---
class User:
def __init__(self, username, password):
self.username = username
self.password = password
class Product:
def __init__(self, id: int, name: str, price: float, stock: int):
self.id = id
self.name = name
self.price = price
self.stock = stock
class Order:
def __init__(self, id: int, username: str, product_id: int, quantity: int, total: float):
self.id = id
self.username = username
self.product_id = product_id
self.quantity = quantity
self.total = total
class LoginRequest(BaseModel):
username: str
password: str
class OrderRequest(BaseModel):
product_id: int
quantity: int = Field(gt=0)
# --- In-Memory Data ---
users = {
"alice": User("alice", "password123"),
"bob": User("bob", "hunter2")
}
products = [
Product(1, "Laptop", 999.99, 10),
Product(2, "Mouse", 25.50, 50),
Product(3, "Keyboard", 75.00, 15),
]
orders = []
order_id_counter = 1
# Simple token management: token -> username
tokens = {}
# --- App Setup ---
app = FastAPI()
security = HTTPBearer(auto_error=False)
# --- Helper functions ---
def get_current_user(auth: Optional[HTTPAuthorizationCredentials] = Depends(security)):
if not auth:
raise HTTPException(status_code=401, detail="Missing token")
token = auth.credentials
username = tokens.get(token)
if not username:
raise HTTPException(status_code=401, detail="Invalid token")
return username
# --- Custom Exception Handlers ---
@app.exception_handler(ValidationError)
async def validation_exception_handler(request: Request, exc: ValidationError):
return JSONResponse(
status_code=400,
content={"error": "Malformed body"},
)
@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
return JSONResponse(
status_code=exc.status_code,
content={"error": exc.detail},
)
# --- Endpoints ---
@app.post("/login")
async def login(request: Request):
try:
data = await request.json()
except Exception:
return JSONResponse(status_code=400, content={"error": "Malformed JSON"})
if "username" not in data or "password" not in data:
return JSONResponse(status_code=400, content={"error": "Missing username or password"})
username = data["username"]
password = data["password"]
user = users.get(username)
if user and user.password == password:
token = str(uuid.uuid4())
tokens[token] = username
return {"token": token}
else:
return JSONResponse(status_code=401, content={"error": "Invalid credentials"})
@app.get("/products")
async def get_products(max_price: Optional[str] = None):
result = []
try:
price_limit = float(max_price) if max_price is not None else None
except ValueError:
price_limit = None
for p in products:
if price_limit is None or p.price <= price_limit:
result.append({
"id": p.id,
"name": p.name,
"price": p.price,
"stock": p.stock
})
return result
@app.get("/products/{product_id}")
async def get_product(product_id: int):
product = next((p for p in products if p.id == product_id), None)
if not product:
raise HTTPException(status_code=404, detail="Product not found")
return {
"id": product.id,
"name": product.name,
"price": product.price,
"stock": product.stock
}
@app.post("/orders", status_code=201)
async def create_order(request: Request, username: str = Depends(get_current_user)):
try:
data = await request.json()
except Exception:
raise HTTPException(status_code=400, detail="Malformed JSON")
# Manual validation to match the specific error requirements if necessary,
# but Pydantic is better. Let's use Pydantic for the core logic.
try:
order_req = OrderRequest(**data)
except Exception:
raise HTTPException(status_code=400, detail="Malformed body")
product = next((p for p in products if p.id == order_req.product_id), None)
if not product:
raise HTTPException(status_code=404, detail="Product not found")
if order_req.quantity > product.stock:
raise HTTPException(status_code=409, detail="Insufficient stock")
global order_id_counter
total = product.price * order_req.quantity
new_order = Order(order_id_counter, username, product.id, order_req.quantity, total)
# Update stock
product.stock -= order_req.quantity
orders.append(new_order)
order_id_counter += 1
return {
"id": new_order.id,
"product_id": new_order.product_id,
"quantity": new_order.quantity,
"total": new_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 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 if o.id == order_id), None)
if not order or order.username != username:
raise HTTPException(status_code=404, detail="Order not found")
return {
"id": order.id,
"product_id": order.product_id,
"quantity": order.quantity,
"total": order.total
}
if __name__ == "__main__":
import uvicorn
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