......................FF. [100%]
=================================== FAILURES ===================================
____________________________ test_get_missing_order ____________________________
base_url = 'http://127.0.0.1:51473'
alice_token = 'afae77b7b7b3e877e4af16ef564a0511c8abc1e99db48e7cb6f6bb929520e1fa'
def test_get_missing_order(base_url, alice_token):
status, parsed, ct = _request_h("GET", f"{base_url}/orders/999999", token=alice_token)
> _assert_json_error(status, parsed, ct, 404)
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:337:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
status = 404, parsed = {'detail': 'Order not found'}, ct = 'application/json'
expect_status = 404
def _assert_json_error(status, parsed, ct, expect_status):
"""A 4xx response must be JSON with an error/message field (see task.md 'Notes')."""
assert status == expect_status, f"expected {expect_status}, got {status} (body={parsed!r})"
assert "application/json" in ct, f"error body should be JSON, content-type={ct!r}"
> assert isinstance(parsed, dict) and (parsed.get("error") or parsed.get("message")), \
f"error body must be a JSON object with an error/message field, got {parsed!r}"
E AssertionError: error body must be a JSON object with an error/message field, got {'detail': 'Order not found'}
E assert (True and (None or None))
E + where True = isinstance({'detail': 'Order not found'}, dict)
E + and None = <built-in method get of dict object at 0x7f7a1facf100>('error')
E + where <built-in method get of dict object at 0x7f7a1facf100> = {'detail': 'Order not found'}.get
E + and None = <built-in method get of dict object at 0x7f7a1facf100>('message')
E + where <built-in method get of dict object at 0x7f7a1facf100> = {'detail': 'Order not found'}.get
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:91: AssertionError
______________________ test_get_other_users_order_is_404 _______________________
base_url = 'http://127.0.0.1:51473'
alice_token = 'afae77b7b7b3e877e4af16ef564a0511c8abc1e99db48e7cb6f6bb929520e1fa'
def test_get_other_users_order_is_404(base_url, alice_token):
# alice creates an order
_, order, _ = _request("POST", f"{base_url}/orders", token=alice_token,
body={"product_id": 2, "quantity": 1})
oid = order["id"]
# bob must not be able to read it (404, not 403 — don't reveal it exists)
_, blogin, _ = _request("POST", f"{base_url}/login",
body={"username": "bob", "password": "hunter2"})
bob_token = blogin["token"]
status, parsed, ct = _request_h("GET", f"{base_url}/orders/{oid}", token=bob_token)
> _assert_json_error(status, parsed, ct, 404)
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:350:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
status = 404, parsed = {'detail': 'Order not found'}, ct = 'application/json'
expect_status = 404
def _assert_json_error(status, parsed, ct, expect_status):
"""A 4xx response must be JSON with an error/message field (see task.md 'Notes')."""
assert status == expect_status, f"expected {expect_status}, got {status} (body={parsed!r})"
assert "application/json" in ct, f"error body should be JSON, content-type={ct!r}"
> assert isinstance(parsed, dict) and (parsed.get("error") or parsed.get("message")), \
f"error body must be a JSON object with an error/message field, got {parsed!r}"
E AssertionError: error body must be a JSON object with an error/message field, got {'detail': 'Order not found'}
E assert (True and (None or None))
E + where True = isinstance({'detail': 'Order not found'}, dict)
E + and None = <built-in method get of dict object at 0x7f7a1f862280>('error')
E + where <built-in method get of dict object at 0x7f7a1f862280> = {'detail': 'Order not found'}.get
E + and None = <built-in method get of dict object at 0x7f7a1f862280>('message')
E + where <built-in method get of dict object at 0x7f7a1f862280> = {'detail': 'Order not found'}.get
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:91: AssertionError
=========================== short test summary info ============================
FAILED ../../home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py::test_get_missing_order
FAILED ../../home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py::test_get_other_users_order_is_404
2 failed, 23 passed in 0.62s
score: 23/25 (92%)
import os
import secrets
from fastapi import FastAPI, Request, HTTPException, Query
from fastapi.responses import JSONResponse
from pydantic import BaseModel
app = FastAPI()
# ---------------------------------------------------------------------------
# In-memory data stores
# ---------------------------------------------------------------------------
USERS = {
"alice": "password123",
"bob": "hunter2",
}
PRODUCTS = {
1: {"id": 1, "name": "Mechanical Keyboard", "price": 89.99, "stock": 10},
2: {"id": 2, "name": "USB-C Hub", "price": 34.50, "stock": 20},
3: {"id": 3, "name": "Wireless Mouse", "price": 24.99, "stock": 15},
}
# token -> username
TOKENS: dict[str, str] = {}
# orders keyed by id; each order stores the username for ownership checks
ORDERS: dict[int, dict] = {}
_next_order_id = 1
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _get_username_from_token(request: Request) -> str:
"""Return the authenticated username or raise 401."""
auth_header = request.headers.get("authorization", "")
if not auth_header.startswith("Bearer "):
raise HTTPException(status_code=401, detail="Missing or invalid token")
token = auth_header[7:]
username = TOKENS.get(token)
if username is None:
raise HTTPException(status_code=401, detail="Missing or invalid token")
return username
# ---------------------------------------------------------------------------
# Public endpoints
# ---------------------------------------------------------------------------
@app.post("/login")
async def login(request: Request):
"""Authenticate a user and return a bearer token."""
try:
body = await request.json()
except Exception:
return JSONResponse(status_code=400, content={"error": "Invalid JSON body"})
if not isinstance(body, dict):
return JSONResponse(status_code=400, content={"error": "Request body must be a JSON object"})
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:
return JSONResponse(status_code=401, content={"error": "Invalid credentials"})
token = secrets.token_hex(32)
TOKENS[token] = username
return {"token": token}
@app.get("/products")
async def list_products(max_price: str = Query(default=None)):
"""List all products, optionally filtered by max_price."""
products = list(PRODUCTS.values())
if max_price is not None:
try:
threshold = float(max_price)
products = [p for p in products if p["price"] <= threshold]
except (ValueError, TypeError):
pass # ignore invalid max_price, return all
return products
@app.get("/products/{product_id}")
async def get_product(product_id: int):
"""Get a single product by id."""
product = PRODUCTS.get(product_id)
if product is None:
raise HTTPException(status_code=404, detail="Product not found")
return product
# ---------------------------------------------------------------------------
# Protected endpoints
# ---------------------------------------------------------------------------
@app.post("/orders", status_code=201)
async def create_order(request: Request):
"""Create an order for the authenticated user."""
username = _get_username_from_token(request)
try:
body = await request.json()
except Exception:
return JSONResponse(status_code=400, content={"error": "Invalid JSON body"})
if not isinstance(body, dict):
return JSONResponse(status_code=400, content={"error": "Request body must be a JSON object"})
product_id = body.get("product_id")
quantity = body.get("quantity")
# Validate presence and integer types
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 isinstance(product_id, bool):
return JSONResponse(status_code=400, content={"error": "'product_id' must be an integer"})
if not isinstance(quantity, int) or isinstance(quantity, bool):
return JSONResponse(status_code=400, content={"error": "'quantity' must be an integer"})
if quantity < 1:
return JSONResponse(status_code=400, content={"error": "'quantity' must be at least 1"})
product = PRODUCTS.get(product_id)
if product is None:
return JSONResponse(status_code=404, content={"error": "Product not found"})
if quantity > product["stock"]:
return JSONResponse(status_code=409, content={"error": "Insufficient stock"})
global _next_order_id
total = product["price"] * quantity
product["stock"] -= quantity
order = {
"id": _next_order_id,
"product_id": product_id,
"quantity": quantity,
"total": total,
"username": username,
}
ORDERS[_next_order_id] = order
_next_order_id += 1
return order
@app.get("/orders")
async def list_orders(request: Request):
"""List orders belonging to the authenticated user."""
username = _get_username_from_token(request)
return [o for o in ORDERS.values() if o["username"] == username]
@app.get("/orders/{order_id}")
async def get_order(order_id: int, request: Request):
"""Get a single order if it belongs to the authenticated user."""
username = _get_username_from_token(request)
order = ORDERS.get(order_id)
if order is None or order["username"] != username:
raise HTTPException(status_code=404, detail="Order not found")
return order
# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------
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.43s score: 25/25 (100%)
import os
import json
import uuid
from http.server import HTTPServer, BaseHTTPRequestHandler
from urllib.parse import urlparse, parse_qs
# ---------------------------------------------------------------------------
# In-memory data
# ---------------------------------------------------------------------------
USERS = {
"alice": "password123",
"bob": "hunter2",
}
PRODUCTS = {
1: {"id": 1, "name": "Wireless Mouse", "price": 29.99, "stock": 10},
2: {"id": 2, "name": "Mechanical Keyboard", "price": 89.99, "stock": 7},
3: {"id": 3, "name": "USB-C Hub", "price": 45.50, "stock": 12},
}
# token -> username
TOKENS: dict[str, str] = {}
# id -> order dict
ORDERS: dict[int, dict] = {}
_NEXT_ORDER_ID = 1
# ---------------------------------------------------------------------------
# Request handler
# ---------------------------------------------------------------------------
class StoreHandler(BaseHTTPRequestHandler):
# Silence default stderr logging
def log_message(self, format, *args):
pass
# ------------------------------------------------------------------
# Helpers
# ------------------------------------------------------------------
def _send_json(self, data, status=200):
body = json.dumps(data).encode()
self.send_response(status)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def _send_error(self, message, status):
self._send_json({"error": message}, status)
def _read_json_body(self):
"""Read and parse the request body as JSON. Returns (data, None) or
(None, error_string)."""
length = int(self.headers.get("Content-Length", 0))
if length == 0:
return None, "No request body"
raw = self.rfile.read(length)
try:
data = json.loads(raw)
except (json.JSONDecodeError, ValueError):
return None, "Invalid JSON body"
return data, None
def _get_user(self):
"""Return the authenticated username or None."""
auth = self.headers.get("Authorization", "")
if not auth.startswith("Bearer "):
return None
return TOKENS.get(auth[7:])
def _require_user(self):
"""Return the authenticated username or send 401."""
user = self._get_user()
if user is None:
self._send_error("Authentication required", 401)
return None
return user
# ------------------------------------------------------------------
# Route dispatch
# ------------------------------------------------------------------
def _route(self, method):
parsed = urlparse(self.path)
path = parsed.path.rstrip("/") or "/"
query = parse_qs(parsed.query)
# GET /products
if method == "GET" and path == "/products":
return self._handle_get_products(query)
# GET /products/<id>
if method == "GET" and path.startswith("/products/"):
parts = path.split("/")
if len(parts) == 3 and parts[2].isdigit():
return self._handle_get_product(int(parts[2]))
return self._send_error("Not found", 404)
# POST /login
if method == "POST" and path == "/login":
return self._handle_login()
# POST /orders
if method == "POST" and path == "/orders":
return self._handle_create_order()
# GET /orders
if method == "GET" and path == "/orders":
return self._handle_list_orders()
# GET /orders/<id>
if method == "GET" and path.startswith("/orders/"):
parts = path.split("/")
if len(parts) == 3 and parts[2].isdigit():
return self._handle_get_order(int(parts[2]))
return self._send_error("Not found", 404)
self._send_error("Not found", 404)
def do_GET(self):
self._route("GET")
def do_POST(self):
self._route("POST")
# Reject other methods with JSON
def do_PUT(self):
self._send_error("Method not allowed", 405)
def do_DELETE(self):
self._send_error("Method not allowed", 405)
def do_PATCH(self):
self._send_error("Method not allowed", 405)
# ------------------------------------------------------------------
# Handlers
# ------------------------------------------------------------------
def _handle_get_products(self, query):
result = list(PRODUCTS.values())
if "max_price" in query:
try:
max_price = float(query["max_price"][0])
result = [p for p in result if p["price"] <= max_price]
except (ValueError, IndexError):
pass
self._send_json(result)
def _handle_get_product(self, product_id):
product = PRODUCTS.get(product_id)
if product is None:
self._send_error("Product not found", 404)
return
self._send_json(product)
def _handle_login(self):
data, err = self._read_json_body()
if err:
self._send_error(err, 400)
return
if not isinstance(data, dict):
self._send_error("Invalid JSON body", 400)
return
username = data.get("username")
password = data.get("password")
if username is None or password is None:
self._send_error("Missing username or password", 400)
return
if USERS.get(username) != password:
self._send_error("Invalid credentials", 401)
return
token = uuid.uuid4().hex
TOKENS[token] = username
self._send_json({"token": token})
def _handle_create_order(self):
global _NEXT_ORDER_ID
user = self._require_user()
if user is None:
return
data, err = self._read_json_body()
if err:
self._send_error(err, 400)
return
if not isinstance(data, dict):
self._send_error("Invalid JSON body", 400)
return
product_id = data.get("product_id")
quantity = data.get("quantity")
if product_id is None or quantity is None:
self._send_error("Missing product_id or quantity", 400)
return
# Must be int, not bool
if not isinstance(product_id, int) or isinstance(product_id, bool):
self._send_error("product_id must be an integer", 400)
return
if not isinstance(quantity, int) or isinstance(quantity, bool):
self._send_error("quantity must be an integer", 400)
return
if quantity < 1:
self._send_error("quantity must be at least 1", 400)
return
product = PRODUCTS.get(product_id)
if product is None:
self._send_error("Product not found", 404)
return
if quantity > product["stock"]:
self._send_error("Insufficient stock", 409)
return
order_id = _NEXT_ORDER_ID
_NEXT_ORDER_ID += 1
total = product["price"] * quantity
order = {
"id": order_id,
"product_id": product_id,
"quantity": quantity,
"total": total,
"user": user,
}
ORDERS[order_id] = order
product["stock"] -= quantity
self._send_json(order, 201)
def _handle_list_orders(self):
user = self._require_user()
if user is None:
return
user_orders = [o for o in ORDERS.values() if o["user"] == user]
self._send_json(user_orders)
def _handle_get_order(self, order_id):
user = self._require_user()
if user is None:
return
order = ORDERS.get(order_id)
if order is None or order["user"] != user:
self._send_error("Order not found", 404)
return
self._send_json(order)
# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
if __name__ == "__main__":
host = os.environ.get("HOST", "127.0.0.1")
port = int(os.environ.get("PORT", "8000"))
server = HTTPServer((host, port), StoreHandler)
print(f"Serving on {host}:{port}")
server.serve_forever()
......................... [100%] 25 passed in 0.60s score: 25/25 (100%)
import os
import secrets
from typing import Optional
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
import uvicorn
app = FastAPI()
# --- In-memory data ---
USERS = {
"alice": "password123",
"bob": "hunter2",
}
PRODUCTS = {
1: {"id": 1, "name": "Wireless Mouse", "price": 29.99, "stock": 10},
2: {"id": 2, "name": "Mechanical Keyboard", "price": 79.50, "stock": 7},
3: {"id": 3, "name": "USB-C Hub", "price": 45.00, "stock": 5},
}
# token -> username
TOKENS: dict[str, str] = {}
# orders keyed by order id
ORDERS: dict[int, dict] = {}
_next_order_id = 1
def _require_auth(request: Request) -> str:
"""Return the authenticated username or raise 401."""
auth_header = request.headers.get("authorization", "")
if not auth_header.startswith("Bearer "):
raise HTTPException(status_code=401, detail="Authentication required")
token = auth_header[7:]
username = TOKENS.get(token)
if username is None:
raise HTTPException(status_code=401, detail="Invalid token")
return username
# --- Public endpoints ---
@app.get("/products")
def get_products(max_price: Optional[str] = None):
products = list(PRODUCTS.values())
if max_price is not None:
try:
limit = float(max_price)
products = [p for p in products if p["price"] <= limit]
except (ValueError, TypeError):
pass # return all products if not a valid number
return products
@app.get("/products/{product_id}")
def get_product(product_id: int):
product = PRODUCTS.get(product_id)
if product is None:
raise HTTPException(status_code=404, detail="Product not found")
return product
@app.post("/login")
async def login(request: Request):
try:
body = await request.json()
except Exception:
raise HTTPException(status_code=400, detail="Invalid JSON")
if not isinstance(body, dict):
raise HTTPException(status_code=400, detail="Invalid request body")
username = body.get("username")
password = body.get("password")
if username is None or password is None:
raise HTTPException(status_code=400, detail="Missing username or password")
if USERS.get(username) != password:
raise HTTPException(status_code=401, detail="Invalid credentials")
token = secrets.token_hex(32)
TOKENS[token] = username
return {"token": token}
# --- Protected endpoints ---
@app.post("/orders", status_code=201)
async def create_order(request: Request):
username = _require_auth(request)
try:
body = await request.json()
except Exception:
raise HTTPException(status_code=400, detail="Invalid JSON")
if not isinstance(body, dict):
raise HTTPException(status_code=400, detail="Invalid request body")
product_id = body.get("product_id")
quantity = body.get("quantity")
# Validate fields exist
if product_id is None or quantity is None:
raise HTTPException(status_code=400, detail="Missing product_id or quantity")
# Validate types — must be integers (not bools, not floats)
if isinstance(product_id, bool) or not isinstance(product_id, int):
raise HTTPException(status_code=400, detail="product_id must be an integer")
if isinstance(quantity, bool) or not isinstance(quantity, int):
raise HTTPException(status_code=400, detail="quantity must be an integer")
# quantity must be at least 1
if quantity < 1:
raise HTTPException(status_code=400, detail="quantity must be at least 1")
# Look up product
product = PRODUCTS.get(product_id)
if product is None:
raise HTTPException(status_code=404, detail="Product not found")
# Check stock
if quantity > product["stock"]:
raise HTTPException(status_code=409, detail="Insufficient stock")
# Create order
global _next_order_id
order = {
"id": _next_order_id,
"product_id": product_id,
"quantity": quantity,
"total": product["price"] * quantity,
"username": username,
}
ORDERS[_next_order_id] = order
_next_order_id += 1
# Reduce stock
product["stock"] -= quantity
return order
@app.get("/orders")
def get_orders(request: Request):
username = _require_auth(request)
user_orders = [o for o in ORDERS.values() if o["username"] == username]
return user_orders
@app.get("/orders/{order_id}")
def get_order(order_id: int, request: Request):
username = _require_auth(request)
order = ORDERS.get(order_id)
if order is None or order["username"] != username:
raise HTTPException(status_code=404, detail="Order not found")
return order
# --- Custom exception handler for consistent JSON error responses ---
@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
return JSONResponse(
status_code=exc.status_code,
content={"error": exc.detail},
)
# --- Entry point ---
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, log_level="warning")
fastapi uvicorn
......................... [100%] 25 passed in 0.43s score: 25/25 (100%)
import json
import os
import re
import uuid
from http.server import HTTPServer, BaseHTTPRequestHandler
from urllib.parse import urlparse, parse_qs
# --- In-memory data stores ---
users = {
"alice": "password123",
"bob": "hunter2",
}
products = {
1: {"id": 1, "name": "Wireless Mouse", "price": 29.99, "stock": 10},
2: {"id": 2, "name": "Mechanical Keyboard", "price": 89.99, "stock": 7},
3: {"id": 3, "name": "USB-C Hub", "price": 45.50, "stock": 5},
}
# token -> username
tokens: dict[str, str] = {}
# orders: id -> order dict (includes "username")
orders: dict[int, dict] = {}
next_order_id = 1
class StoreHandler(BaseHTTPRequestHandler):
"""HTTP request handler for the online store API."""
# Suppress default stderr logging for cleaner output
def log_message(self, format, *args):
pass
# --- Helpers ---
def _send_json(self, data, status=200):
body = json.dumps(data).encode("utf-8")
self.send_response(status)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def _send_error(self, message, status):
self._send_json({"error": message}, status)
def _read_json_body(self):
"""Read and parse JSON body. Returns (parsed_dict, None) or (None, error_response)."""
content_length = int(self.headers.get("Content-Length", 0))
if content_length == 0:
return None, "Invalid JSON body"
raw = self.rfile.read(content_length)
try:
data = json.loads(raw)
except (json.JSONDecodeError, ValueError):
return None, "Invalid JSON body"
if not isinstance(data, dict):
return None, "Invalid JSON body"
return data, None
def _get_current_user(self):
"""Return authenticated username or None."""
auth = self.headers.get("Authorization", "")
if not auth.startswith("Bearer "):
return None
token = auth[7:]
return tokens.get(token)
def _route(self, method):
parsed = urlparse(self.path)
path = parsed.path.rstrip("/") or "/"
query = parse_qs(parsed.query)
# Public endpoints
if path == "/login" and method == "POST":
return self._handle_login()
if path == "/products" and method == "GET":
return self._handle_get_products(query)
if method == "GET":
m = re.fullmatch(r"/products/(\d+)", path)
if m:
return self._handle_get_product(int(m.group(1)))
# Protected endpoints
if path == "/orders" and method == "POST":
return self._handle_create_order()
if path == "/orders" and method == "GET":
return self._handle_get_orders()
if method == "GET":
m = re.fullmatch(r"/orders/(\d+)", path)
if m:
return self._handle_get_order(int(m.group(1)))
self._send_error("Not found", 404)
def do_GET(self):
self._route("GET")
def do_POST(self):
self._route("POST")
# --- Public endpoint handlers ---
def _handle_login(self):
data, err = self._read_json_body()
if err:
return self._send_error(err, 400)
username = data.get("username")
password = data.get("password")
if username is None or password is None:
return self._send_error("Missing 'username' or 'password' field", 400)
if users.get(username) != password:
return self._send_error("Invalid credentials", 401)
token = uuid.uuid4().hex
tokens[token] = username
self._send_json({"token": token}, 200)
def _handle_get_products(self, query):
result = list(products.values())
max_price_vals = query.get("max_price")
if max_price_vals:
try:
max_price = float(max_price_vals[0])
result = [p for p in result if p["price"] <= max_price]
except (ValueError, TypeError):
pass # return all products if max_price is not a valid number
self._send_json(result, 200)
def _handle_get_product(self, product_id):
product = products.get(product_id)
if product is None:
return self._send_error("Product not found", 404)
self._send_json(product, 200)
# --- Protected endpoint handlers ---
def _handle_create_order(self):
user = self._get_current_user()
if user is None:
return self._send_error("Unauthorized", 401)
data, err = self._read_json_body()
if err:
return self._send_error(err, 400)
product_id = data.get("product_id")
quantity = data.get("quantity")
if product_id is None or quantity is None:
return self._send_error("Missing 'product_id' or 'quantity' field", 400)
if not isinstance(product_id, int) or not isinstance(quantity, int):
return self._send_error("'product_id' and 'quantity' must be integers", 400)
if quantity < 1:
return self._send_error("'quantity' must be at least 1", 400)
product = products.get(product_id)
if product is None:
return self._send_error("Product not found", 404)
if quantity > product["stock"]:
return self._send_error("Insufficient stock", 409)
global next_order_id
order = {
"id": next_order_id,
"product_id": product_id,
"quantity": quantity,
"total": product["price"] * quantity,
"username": user,
}
next_order_id += 1
orders[order["id"]] = order
product["stock"] -= quantity
self._send_json({
"id": order["id"],
"product_id": order["product_id"],
"quantity": order["quantity"],
"total": order["total"],
}, 201)
def _handle_get_orders(self):
user = self._get_current_user()
if user is None:
return self._send_error("Unauthorized", 401)
user_orders = [
{"id": o["id"], "product_id": o["product_id"], "quantity": o["quantity"], "total": o["total"]}
for o in orders.values()
if o["username"] == user
]
self._send_json(user_orders, 200)
def _handle_get_order(self, order_id):
user = self._get_current_user()
if user is None:
return self._send_error("Unauthorized", 401)
order = orders.get(order_id)
if order is None or order["username"] != user:
return self._send_error("Order not found", 404)
self._send_json({
"id": order["id"],
"product_id": order["product_id"],
"quantity": order["quantity"],
"total": order["total"],
}, 200)
if __name__ == "__main__":
host = os.environ.get("HOST", "127.0.0.1")
port = int(os.environ.get("PORT", "8000"))
server = HTTPServer((host, port), StoreHandler)
print(f"Serving on {host}:{port}")
server.serve_forever()
# No third-party dependencies — uses only the Python standard library.
......................... [100%] 25 passed in 0.60s score: 25/25 (100%)
import os
import uuid
import json
from fastapi import FastAPI, Request, HTTPException, Query
from fastapi.responses import JSONResponse
import uvicorn
app = FastAPI()
# ---------------------------------------------------------------------------
# In-memory data stores
# ---------------------------------------------------------------------------
USERS = {
"alice": "password123",
"bob": "hunter2",
}
# token -> username
TOKENS: dict[str, str] = {}
PRODUCTS = {
1: {"id": 1, "name": "Laptop", "price": 999.99, "stock": 10},
2: {"id": 2, "name": "Headphones", "price": 49.99, "stock": 20},
3: {"id": 3, "name": "Keyboard", "price": 29.99, "stock": 15},
}
# order_id -> order dict (includes "username")
ORDERS: dict[int, dict] = {}
_next_order_id = 1
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _get_username_from_token(request: Request) -> str:
"""Return the username for the bearer token, or raise 401."""
auth = request.headers.get("authorization", "")
if not auth.startswith("Bearer "):
raise HTTPException(status_code=401, detail="Missing or invalid token")
token = auth[7:]
username = TOKENS.get(token)
if username is None:
raise HTTPException(status_code=401, detail="Missing or invalid token")
return username
async def _read_json_body(request: Request):
"""Read and parse the request body as JSON. Returns None on failure."""
try:
body = await request.body()
if not body:
return None
data = json.loads(body)
return data
except (json.JSONDecodeError, Exception):
return None
# ---------------------------------------------------------------------------
# Public endpoints
# ---------------------------------------------------------------------------
@app.get("/products")
def list_products(max_price: str = Query(default=None)):
products = list(PRODUCTS.values())
if max_price is not None:
try:
threshold = float(max_price)
products = [p for p in products if p["price"] <= threshold]
except (ValueError, TypeError):
pass # ignore invalid max_price, return all
return products
@app.get("/products/{product_id}")
def get_product(product_id: int):
product = PRODUCTS.get(product_id)
if product is None:
raise HTTPException(status_code=404, detail="Product not found")
return product
@app.post("/login")
async def login(request: Request):
data = await _read_json_body(request)
if data is None or not isinstance(data, dict):
raise HTTPException(status_code=400, detail="Malformed request body")
username = data.get("username")
password = data.get("password")
if not username or not password:
raise HTTPException(status_code=400, detail="Missing username or password")
if USERS.get(username) != password:
raise HTTPException(status_code=401, detail="Invalid credentials")
token = uuid.uuid4().hex
TOKENS[token] = username
return {"token": token}
# ---------------------------------------------------------------------------
# Protected endpoints
# ---------------------------------------------------------------------------
@app.post("/orders")
async def create_order(request: Request):
username = _get_username_from_token(request)
data = await _read_json_body(request)
if data is None or not isinstance(data, dict):
raise HTTPException(status_code=400, detail="Malformed request body")
product_id = data.get("product_id")
quantity = data.get("quantity")
# Validate types – must be integers (not bools, not floats)
if not isinstance(product_id, int) or isinstance(product_id, bool):
raise HTTPException(status_code=400, detail="product_id must be an integer")
if not isinstance(quantity, int) or isinstance(quantity, bool):
raise HTTPException(status_code=400, detail="quantity must be an integer")
if quantity < 1:
raise HTTPException(status_code=400, detail="quantity must be at least 1")
product = PRODUCTS.get(product_id)
if product is None:
raise HTTPException(status_code=404, detail="Product not found")
if quantity > product["stock"]:
raise HTTPException(status_code=409, detail="Insufficient stock")
global _next_order_id
order = {
"id": _next_order_id,
"product_id": product_id,
"quantity": quantity,
"total": product["price"] * quantity,
"username": username,
}
_next_order_id += 1
ORDERS[order["id"]] = order
product["stock"] -= quantity
return JSONResponse(status_code=201, content=order)
@app.get("/orders")
def list_orders(request: Request):
username = _get_username_from_token(request)
return [o for o in ORDERS.values() if o["username"] == username]
@app.get("/orders/{order_id}")
def get_order(order_id: int, request: Request):
username = _get_username_from_token(request)
order = ORDERS.get(order_id)
if order is None or order["username"] != username:
raise HTTPException(status_code=404, detail="Order not found")
return order
# ---------------------------------------------------------------------------
# Catch-all for any non-JSON error response (ensure errors are JSON)
# ---------------------------------------------------------------------------
@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
return JSONResponse(
status_code=exc.status_code,
content={"error": exc.detail},
)
# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------
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, log_level="info")
fastapi uvicorn