......................... [100%] 25 passed in 0.43s score: 25/25 (100%)
import os
import uuid
import hashlib
from datetime import datetime, timedelta
from http.server import HTTPServer, BaseHTTPRequestHandler
from urllib.parse import urlparse, parse_qs
import json
# ─── In-memory data stores ────────────────────────────────────────────────────
USERS = {
"alice": {"username": "alice", "password": "password123"},
"bob": {"username": "bob", "password": "hunter2"},
}
PRODUCTS = [
{"id": 1, "name": "Wireless Mouse", "price": 29.99, "stock": 50},
{"id": 2, "name": "Mechanical Keyboard", "price": 89.99, "stock": 30},
{"id": 3, "name": "USB-C Hub", "price": 49.99, "stock": 45},
]
# Orders keyed by order id: { order_id: { id, product_id, quantity, total, username, created_at } }
ORDERS = {}
ORDER_ID_COUNTER = 0
# Token store: { token: username }
TOKEN_STORE = {}
# ─── Helpers ──────────────────────────────────────────────────────────────────
def _json_response(handler, status, body):
"""Send a JSON response with proper headers."""
handler.send_response(status)
handler.send_header("Content-Type", "application/json")
handler.end_headers()
handler.wfile.write(json.dumps(body).encode("utf-8"))
def _read_body(handler):
"""Read and parse the request body as JSON."""
content_length = int(handler.headers.get("Content-Length", 0))
if content_length == 0:
return None
raw = handler.rfile.read(content_length)
try:
return json.loads(raw)
except (json.JSONDecodeError, ValueError):
return None
def _authenticate(handler):
"""Extract and validate the Bearer token. Returns username or None."""
auth_header = handler.headers.get("Authorization", "")
if not auth_header.startswith("Bearer "):
return None
token = auth_header[7:]
username = TOKEN_STORE.get(token)
if username is None:
return None
return username
def _generate_token(username):
"""Create a new token for a user."""
token = hashlib.sha256(f"{uuid.uuid4()}{username}{datetime.utcnow().isoformat()}".encode()).hexdigest()
TOKEN_STORE[token] = username
return token
# ─── Request Handler ──────────────────────────────────────────────────────────
class StoreHandler(BaseHTTPRequestHandler):
"""HTTP request handler for the online store API."""
def log_message(self, format, *args):
"""Suppress default logging."""
pass
# ── GET ──────────────────────────────────────────────────────────────────
def do_GET(self):
parsed = urlparse(self.path)
path = parsed.path.rstrip("/") or "/"
query = parse_qs(parsed.query)
# GET /products (public)
if path == "/products":
return self._get_products(query)
# GET /products/<id> (public)
if path.startswith("/products/"):
parts = path.split("/")
if len(parts) == 3:
try:
product_id = int(parts[2])
except (ValueError, IndexError):
return _json_response(self, 404, {"error": "Product not found"})
return self._get_product(product_id)
# GET /orders (protected)
if path == "/orders":
return self._get_orders()
# GET /orders/<id> (protected)
if path.startswith("/orders/"):
parts = path.split("/")
if len(parts) == 3:
try:
order_id = int(parts[2])
except (ValueError, IndexError):
return _json_response(self, 404, {"error": "Order not found"})
return self._get_order(order_id)
_json_response(self, 404, {"error": "Not found"})
# ── POST ─────────────────────────────────────────────────────────────────
def do_POST(self):
parsed = urlparse(self.path)
path = parsed.path.rstrip("/") or "/"
# POST /login (public)
if path == "/login":
return self._login()
# POST /orders (protected)
if path == "/orders":
return self._create_order()
_json_response(self, 404, {"error": "Not found"})
# ── Product endpoints ────────────────────────────────────────────────────
def _get_products(self, query):
"""GET /products with optional ?max_price=X filter."""
products = PRODUCTS
max_price_param = query.get("max_price", [None])[0]
if max_price_param is not None:
try:
max_price = float(max_price_param)
products = [p for p in products if p["price"] <= max_price]
except (ValueError, TypeError):
pass # invalid max_price → return all
_json_response(self, 200, products)
def _get_product(self, product_id):
"""GET /products/<id>"""
for p in PRODUCTS:
if p["id"] == product_id:
return _json_response(self, 200, p)
_json_response(self, 404, {"error": "Product not found"})
# ── Authentication ───────────────────────────────────────────────────────
def _login(self):
"""POST /login"""
body = _read_body(self)
if body is None or not isinstance(body, dict):
return _json_response(self, 400, {"error": "Invalid JSON body"})
username = body.get("username")
password = body.get("password")
if not username or not password:
return _json_response(self, 400, {"error": "Missing username or password"})
user = USERS.get(username)
if user is None or user["password"] != password:
return _json_response(self, 401, {"error": "Invalid credentials"})
token = _generate_token(username)
_json_response(self, 200, {"token": token})
# ── Order endpoints (protected) ──────────────────────────────────────────
def _create_order(self):
"""POST /orders (protected)"""
username = _authenticate(self)
if username is None:
return _json_response(self, 401, {"error": "Invalid or missing token"})
body = _read_body(self)
if body is None or not isinstance(body, dict):
return _json_response(self, 400, {"error": "Invalid JSON body"})
product_id = body.get("product_id")
quantity = body.get("quantity")
# Validate presence and types
if product_id is None or quantity is None:
return _json_response(self, 400, {"error": "Missing product_id or quantity"})
if not isinstance(product_id, int) or not isinstance(quantity, int):
return _json_response(self, 400, {"error": "product_id and quantity must be integers"})
if quantity < 1:
return _json_response(self, 400, {"error": "Quantity must be at least 1"})
# Find product
product = None
for p in PRODUCTS:
if p["id"] == product_id:
product = p
break
if product is None:
return _json_response(self, 404, {"error": "Product not found"})
# Check stock
if quantity > product["stock"]:
return _json_response(self, 409, {"error": "Insufficient stock"})
# Create order
global ORDER_ID_COUNTER
ORDER_ID_COUNTER += 1
order_id = ORDER_ID_COUNTER
total = product["price"] * quantity
order = {
"id": order_id,
"product_id": product_id,
"quantity": quantity,
"total": total,
"username": username,
"created_at": datetime.utcnow().isoformat(),
}
ORDERS[order_id] = order
# Reduce stock
product["stock"] -= quantity
result = {
"id": order["id"],
"product_id": order["product_id"],
"quantity": order["quantity"],
"total": order["total"],
}
_json_response(self, 201, result)
def _get_orders(self):
"""GET /orders (protected) — return only the authenticated user's orders."""
username = _authenticate(self)
if username is None:
return _json_response(self, 401, {"error": "Invalid or missing token"})
user_orders = []
for o in ORDERS.values():
if o["username"] == username:
user_orders.append({
"id": o["id"],
"product_id": o["product_id"],
"quantity": o["quantity"],
"total": o["total"],
})
_json_response(self, 200, user_orders)
def _get_order(self, order_id):
"""GET /orders/<id> (protected)"""
username = _authenticate(self)
if username is None:
return _json_response(self, 401, {"error": "Invalid or missing token"})
order = ORDERS.get(order_id)
if order is None or order["username"] != username:
return _json_response(self, 404, {"error": "Order not found"})
result = {
"id": order["id"],
"product_id": order["product_id"],
"quantity": order["quantity"],
"total": order["total"],
}
_json_response(self, 200, result)
# ─── Main ─────────────────────────────────────────────────────────────────────
def main():
host = os.environ.get("HOST", "127.0.0.1")
port = int(os.environ.get("PORT", "8000"))
server = HTTPServer((host, port), StoreHandler)
print(f"Server running on http://{host}:{port}")
try:
server.serve_forever()
except KeyboardInterrupt:
pass
server.server_close()
print("Server stopped.")
if __name__ == "__main__":
main()
......................FF. [100%]
=================================== FAILURES ===================================
____________________________ test_get_missing_order ____________________________
base_url = 'http://127.0.0.1:48579'
alice_token = '6c00f77ee80c23aa9757655d54febee540c41840c6b4f75cf2a5b7b17d476359'
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 0x7fd77e252580>('error')
E + where <built-in method get of dict object at 0x7fd77e252580> = {'detail': 'Order not found'}.get
E + and None = <built-in method get of dict object at 0x7fd77e252580>('message')
E + where <built-in method get of dict object at 0x7fd77e252580> = {'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:48579'
alice_token = '6c00f77ee80c23aa9757655d54febee540c41840c6b4f75cf2a5b7b17d476359'
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 0x7fd77e547640>('error')
E + where <built-in method get of dict object at 0x7fd77e547640> = {'detail': 'Order not found'}.get
E + and None = <built-in method get of dict object at 0x7fd77e547640>('message')
E + where <built-in method get of dict object at 0x7fd77e547640> = {'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%)
"""Online Store REST API — built with FastAPI."""
import os
import secrets
import time
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional
from fastapi import FastAPI, Header, Query, Request, Response, HTTPException
from fastapi.responses import JSONResponse
# ---------------------------------------------------------------------------
# In-memory data stores
# ---------------------------------------------------------------------------
USERS: Dict[str, Dict[str, str]] = {
"alice": {"username": "alice", "password": "password123"},
"bob": {"username": "bob", "password": "hunter2"},
}
PRODUCTS: Dict[int, Dict[str, Any]] = {
1: {"id": 1, "name": "Wireless Mouse", "price": 29.99, "stock": 50},
2: {"id": 2, "name": "Mechanical Keyboard", "price": 89.50, "stock": 30},
3: {"id": 3, "name": "USB-C Hub", "price": 45.00, "stock": 25},
}
ORDERS: Dict[int, Dict[str, Any]] = {} # id -> order dict
order_counter: int = 0
# token -> {"username": str, ...}
TOKENS: Dict[str, Dict[str, Any]] = {}
# ---------------------------------------------------------------------------
# FastAPI app
# ---------------------------------------------------------------------------
app = FastAPI(title="Online Store API")
# ---------------------------------------------------------------------------
# Authentication helpers
# ---------------------------------------------------------------------------
def _authenticate(authorization: Optional[str]) -> Dict[str, Any]:
"""Return the user dict if the token is valid, else raise 401."""
if not authorization or not authorization.startswith("Bearer "):
raise HTTPException(status_code=401, detail="Missing or invalid Authorization header")
token = authorization[len("Bearer "):]
user_info = TOKENS.get(token)
if not user_info:
raise HTTPException(status_code=401, detail="Invalid or expired token")
return user_info
def _create_token(username: str) -> str:
token = secrets.token_hex(32)
TOKENS[token] = {"username": username, "created_at": time.time()}
return token
# ---------------------------------------------------------------------------
# Public endpoints
# ---------------------------------------------------------------------------
@app.get("/products")
def list_products(max_price: Optional[str] = Query(None)):
"""Return all products, optionally filtered by max_price."""
products = list(PRODUCTS.values())
if max_price is not None:
try:
threshold = float(max_price)
except (ValueError, TypeError):
pass # invalid number → return all
else:
products = [p for p in products if p["price"] <= threshold]
return products
@app.get("/products/{product_id}")
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
# ---------------------------------------------------------------------------
# Auth endpoint
# ---------------------------------------------------------------------------
@app.post("/login")
async def login(request: 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"})
username = body.get("username")
password = body.get("password")
if not username or not password:
return JSONResponse(status_code=400, content={"error": "Missing username or password"})
user = USERS.get(username)
if not user or user["password"] != password:
return JSONResponse(status_code=401, content={"error": "Invalid credentials"})
token = _create_token(username)
return {"token": token}
# ---------------------------------------------------------------------------
# Protected endpoints
# ---------------------------------------------------------------------------
@app.post("/orders")
async def create_order(request: Request, authorization: str = Header(None)):
user_info = _authenticate(authorization)
# Parse body
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
if product_id is None or quantity is None:
return JSONResponse(status_code=400, content={"error": "Missing product_id or quantity"})
# Validate integer types
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"})
# Check product exists
product = PRODUCTS.get(product_id)
if not product:
return JSONResponse(status_code=404, content={"error": "Product not found"})
# Check stock
if quantity > product["stock"]:
return JSONResponse(status_code=409, content={"error": "Insufficient stock"})
# Create order
global order_counter
order_counter += 1
total = product["price"] * quantity
order = {
"id": order_counter,
"product_id": product_id,
"quantity": quantity,
"total": total,
"username": user_info["username"],
"created_at": datetime.now(timezone.utc).isoformat(),
}
ORDERS[order["id"]] = order
# Reduce stock
product["stock"] -= quantity
# Return order without username/created_at
return JSONResponse(status_code=201, content={
"id": order["id"],
"product_id": order["product_id"],
"quantity": order["quantity"],
"total": order["total"],
})
@app.get("/orders")
def list_orders(authorization: str = Header(None)):
user_info = _authenticate(authorization)
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_info["username"]
]
return user_orders
@app.get("/orders/{order_id}")
def get_order(order_id: int, authorization: str = Header(None)):
user_info = _authenticate(authorization)
order = ORDERS.get(order_id)
if not order or order["username"] != user_info["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"],
}
# ---------------------------------------------------------------------------
# Global exception handler — catch anything that slips through
# ---------------------------------------------------------------------------
@app.exception_handler(Exception)
async def handle_unexpected(request: Request, exc: Exception):
return JSONResponse(
status_code=500,
content={"error": "Internal server error"},
)
# ---------------------------------------------------------------------------
# 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:app", host=host, port=port, log_level="info")
fastapi uvicorn
......................... [100%] 25 passed in 0.60s score: 25/25 (100%)
import os
import uuid
import hashlib
from datetime import datetime, timezone
from typing import Optional, Dict, Any
from fastapi import FastAPI, Request, HTTPException, Query
from fastapi.responses import JSONResponse
from pydantic import BaseModel
import uvicorn
# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------
HOST = os.environ.get("HOST", "127.0.0.1")
PORT = int(os.environ.get("PORT", "8000"))
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _hash(secret: str) -> str:
"""SHA-256 hash for demonstration."""
return hashlib.sha256(secret.encode()).hexdigest()
def _make_token(username: str) -> str:
"""Create a random token and record its owner."""
token = uuid.uuid4().hex
TOKENS[token] = username
return token
def _authenticate(request: Request) -> str:
"""Return the username from the Authorization header 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 not username:
raise HTTPException(status_code=401, detail="Invalid token")
return username
# ---------------------------------------------------------------------------
# In-memory stores
# ---------------------------------------------------------------------------
USERS: Dict[str, dict] = {
"alice": {"username": "alice", "password_hash": _hash("password123")},
"bob": {"username": "bob", "password_hash": _hash("hunter2")},
}
PRODUCTS: Dict[int, dict] = {
1: {"id": 1, "name": "Wireless Mouse", "price": 29.99, "stock": 20},
2: {"id": 2, "name": "Mechanical Keyboard", "price": 79.50, "stock": 15},
3: {"id": 3, "name": "USB-C Hub", "price": 49.99, "stock": 10},
}
ORDERS: Dict[str, dict] = {}
_next_order_id: int = 1
TOKENS: Dict[str, str] = {}
# ---------------------------------------------------------------------------
# Pydantic models
# ---------------------------------------------------------------------------
class LoginRequest(BaseModel):
username: str
password: str
class OrderRequest(BaseModel):
product_id: int
quantity: int
# ---------------------------------------------------------------------------
# FastAPI app
# ---------------------------------------------------------------------------
app = FastAPI(title="Online Store API")
# ---------- Public endpoints ------------------------------------------------
@app.get("/products")
async def list_products(max_price: Optional[str] = Query(None)):
"""Return all products, optionally filtered by max_price."""
products = list(PRODUCTS.values())
if max_price is not None:
try:
limit = float(max_price)
except (ValueError, TypeError):
pass # invalid value -> return all
else:
products = [p for p in products if p["price"] <= limit]
return products
@app.get("/products/{product_id}")
async def get_product(product_id: int):
"""Return a single product by id, or 404."""
product = PRODUCTS.get(product_id)
if product is None:
raise HTTPException(status_code=404, detail="Product not found")
return product
# ---------- Auth endpoint ---------------------------------------------------
@app.post("/login")
async def login(request: Request):
"""Authenticate and return a token."""
try:
body = await request.json()
except Exception:
raise HTTPException(status_code=400, detail="Invalid JSON body")
if not isinstance(body, dict):
raise HTTPException(status_code=400, detail="Invalid JSON body")
username = body.get("username")
password = body.get("password")
if not username or not password:
raise HTTPException(status_code=400, detail="Missing username or password")
user = USERS.get(username)
if user is None or user["password_hash"] != _hash(password):
raise HTTPException(status_code=401, detail="Invalid credentials")
token = _make_token(username)
return {"token": token}
# ---------- Protected endpoints ---------------------------------------------
@app.post("/orders", status_code=201)
async def create_order(request: Request):
"""Create a new order (authenticated)."""
username = _authenticate(request)
# Parse body
try:
body = await request.json()
except Exception:
raise HTTPException(status_code=400, detail="Invalid JSON body")
if not isinstance(body, dict):
raise HTTPException(status_code=400, detail="Invalid JSON body")
product_id = body.get("product_id")
quantity = body.get("quantity")
if product_id is None or quantity is None:
raise HTTPException(status_code=400, detail="Missing product_id or quantity")
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")
# Check product exists
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
_next_order_id += 1
order = {
"id": order_id,
"user": username,
"product_id": product_id,
"quantity": quantity,
"total": round(product["price"] * quantity, 2),
"created_at": datetime.now(timezone.utc).isoformat(),
}
ORDERS[str(order_id)] = order
# Reduce stock
product["stock"] -= quantity
return {
"id": order["id"],
"product_id": order["product_id"],
"quantity": order["quantity"],
"total": order["total"],
}
@app.get("/orders")
async def list_orders(request: Request):
"""Return orders for the authenticated user only."""
username = _authenticate(request)
user_orders = [
{
"id": o["id"],
"product_id": o["product_id"],
"quantity": o["quantity"],
"total": o["total"],
}
for o in ORDERS.values()
if o["user"] == username
]
return user_orders
@app.get("/orders/{order_id}")
async def get_order(order_id: int, request: Request):
"""Return a single order belonging to the authenticated user."""
username = _authenticate(request)
order = ORDERS.get(str(order_id))
if order is None or order["user"] != 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"],
}
# ---------- Global error handler for JSON 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},
)
# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
if __name__ == "__main__":
uvicorn.run(app, host=HOST, port=PORT)
fastapi uvicorn python-jose[cryptography] passlib[bcrypt]
......................... [100%] 25 passed in 0.61s score: 25/25 (100%)
import os
import uuid
import copy
import hashlib
from datetime import datetime, timedelta
from typing import Optional, Dict, List, Any
from fastapi import FastAPI, Request, Response, HTTPException, Query
from fastapi.responses import JSONResponse
from pydantic import BaseModel
# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------
HOST = os.environ.get("HOST", "127.0.0.1")
PORT = int(os.environ.get("PORT", 8000))
SECRET_KEY = os.environ.get("SECRET_KEY", "super-secret-key-for-jwt-signing")
TOKEN_EXPIRE_MINUTES = 60
# ---------------------------------------------------------------------------
# In-memory data stores
# ---------------------------------------------------------------------------
# Users: username -> {username, password_hash}
USERS: Dict[str, dict] = {}
# Products: id -> {id, name, price, stock}
PRODUCTS: Dict[int, dict] = {}
# Orders: order_id -> {id, user, product_id, quantity, total, created_at}
ORDERS: Dict[str, dict] = {}
# Tokens: token -> {username, expires_at}
TOKENS: Dict[str, dict] = {}
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _hash_password(password: str) -> str:
"""Simple SHA-256 hash for demo purposes."""
return hashlib.sha256(password.encode()).hexdigest()
def _create_token(username: str) -> str:
token = uuid.uuid4().hex
TOKENS[token] = {
"username": username,
"expires_at": datetime.utcnow() + timedelta(minutes=TOKEN_EXPIRE_MINUTES),
}
return token
def _authenticate(request: Request) -> dict:
"""Return the user dict if a valid Bearer token is present; else raise 401."""
auth_header = request.headers.get("Authorization", "")
if not auth_header.startswith("Bearer "):
raise HTTPException(status_code=401, detail={"error": "Missing or invalid Authorization header"})
token = auth_header[7:]
token_data = TOKENS.get(token)
if not token_data:
raise HTTPException(status_code=401, detail={"error": "Invalid or expired token"})
if token_data["expires_at"] < datetime.utcnow():
del TOKENS[token]
raise HTTPException(status_code=401, detail={"error": "Token expired"})
username = token_data["username"]
user = USERS.get(username)
if not user:
raise HTTPException(status_code=401, detail={"error": "User not found"})
return user
def _read_json_body(request: Request) -> dict:
"""Parse JSON body; raise 400 on failure."""
try:
body = request.json()
except Exception:
raise HTTPException(status_code=400, detail={"error": "Invalid JSON body"})
if not isinstance(body, dict):
raise HTTPException(status_code=400, detail={"error": "Request body must be a JSON object"})
return body
# ---------------------------------------------------------------------------
# Pydantic models (used for response serialization hints)
# ---------------------------------------------------------------------------
class LoginRequest(BaseModel):
username: str
password: str
class OrderRequest(BaseModel):
product_id: int
quantity: int
# ---------------------------------------------------------------------------
# FastAPI app
# ---------------------------------------------------------------------------
app = FastAPI(title="Online Store API")
# ---------------------------------------------------------------------------
# Seed data
# ---------------------------------------------------------------------------
def _seed():
USERS["alice"] = {"username": "alice", "password_hash": _hash_password("password123")}
USERS["bob"] = {"username": "bob", "password_hash": _hash_password("hunter2")}
PRODUCTS[1] = {"id": 1, "name": "Wireless Mouse", "price": 29.99, "stock": 50}
PRODUCTS[2] = {"id": 2, "name": "Mechanical Keyboard", "price": 89.95, "stock": 30}
PRODUCTS[3] = {"id": 3, "name": "USB-C Hub", "price": 49.50, "stock": 25}
_seed()
# ---------------------------------------------------------------------------
# Public endpoints
# ---------------------------------------------------------------------------
@app.get("/products")
async def api_get_products(max_price: Optional[str] = Query(None)):
"""Return all products, optionally filtered by max_price."""
result = list(PRODUCTS.values())
if max_price is not None:
try:
max_price_val = float(max_price)
except (ValueError, TypeError):
# If not a valid number, return all products
pass
else:
result = [p for p in result if p["price"] <= max_price_val]
return result
@app.get("/products/{product_id}")
async def api_get_product(product_id: int):
"""Return a single product by id, or 404."""
product = PRODUCTS.get(product_id)
if product is None:
raise HTTPException(status_code=404, detail={"error": f"Product with id {product_id} not found"})
return product
# ---------------------------------------------------------------------------
# Auth endpoint
# ---------------------------------------------------------------------------
@app.post("/login")
async def api_login(request: Request):
"""Authenticate and return a JWT-like token."""
# Try to parse as JSON
try:
body = await request.json()
except Exception:
raise HTTPException(status_code=400, detail={"error": "Invalid JSON body"})
if not isinstance(body, dict):
raise HTTPException(status_code=400, detail={"error": "Request body must be a JSON object"})
username = body.get("username")
password = body.get("password")
if not username or not password:
raise HTTPException(
status_code=400,
detail={"error": "Missing 'username' and/or 'password' in request body"},
)
user = USERS.get(username)
if not user or user["password_hash"] != _hash_password(password):
raise HTTPException(status_code=401, detail={"error": "Invalid username or password"})
token = _create_token(username)
return {"token": token}
# ---------------------------------------------------------------------------
# Protected endpoints
# ---------------------------------------------------------------------------
@app.post("/orders")
async def api_create_order(request: Request):
"""Create a new order (authenticated)."""
user = _authenticate(request)
# Parse body
try:
body = await request.json()
except Exception:
raise HTTPException(status_code=400, detail={"error": "Invalid JSON body"})
if not isinstance(body, dict):
raise HTTPException(status_code=400, detail={"error": "Request body must be a JSON object"})
# Validate required fields
if "product_id" not in body:
raise HTTPException(status_code=400, detail={"error": "Missing required field 'product_id'"})
if "quantity" not in body:
raise HTTPException(status_code=400, detail={"error": "Missing required field 'quantity'"})
product_id = body["product_id"]
quantity = body["quantity"]
# Validate types — must be integers
if not isinstance(product_id, int) or isinstance(product_id, bool):
raise HTTPException(status_code=400, detail={"error": "'product_id' must be an integer"})
if not isinstance(quantity, int) or isinstance(quantity, bool):
raise HTTPException(status_code=400, detail={"error": "'quantity' must be an integer"})
if quantity < 1:
raise HTTPException(status_code=400, detail={"error": "'quantity' must be at least 1"})
# Check product exists
product = PRODUCTS.get(product_id)
if product is None:
raise HTTPException(status_code=404, detail={"error": f"Product with id {product_id} not found"})
# Check stock
if product["stock"] < quantity:
raise HTTPException(
status_code=409,
detail={"error": f"Insufficient stock. Available: {product['stock']}, requested: {quantity}"},
)
# Create order
order_id = uuid.uuid4().hex
total = product["price"] * quantity
order = {
"id": order_id,
"user": user["username"],
"product_id": product_id,
"quantity": quantity,
"total": total,
"created_at": datetime.utcnow().isoformat(),
}
ORDERS[order_id] = order
# Reduce stock
product["stock"] -= quantity
return JSONResponse(
status_code=201,
content={
"id": order_id,
"user": user["username"],
"product_id": product_id,
"quantity": quantity,
"total": total,
"created_at": order["created_at"],
},
)
@app.get("/orders")
async def api_get_orders(request: Request):
"""Return orders for the authenticated user only."""
user = _authenticate(request)
user_orders = [
{
"id": o["id"],
"user": o["user"],
"product_id": o["product_id"],
"quantity": o["quantity"],
"total": o["total"],
"created_at": o["created_at"],
}
for o in ORDERS.values()
if o["user"] == user["username"]
]
return user_orders
@app.get("/orders/{order_id}")
async def api_get_order(order_id: str, request: Request):
"""Return a single order if it belongs to the authenticated user."""
user = _authenticate(request)
order = ORDERS.get(order_id)
if order is None:
raise HTTPException(status_code=404, detail={"error": f"Order with id {order_id} not found"})
if order["user"] != user["username"]:
raise HTTPException(status_code=404, detail={"error": f"Order with id {order_id} not found"})
return {
"id": order["id"],
"user": order["user"],
"product_id": order["product_id"],
"quantity": order["quantity"],
"total": order["total"],
"created_at": order["created_at"],
}
# ---------------------------------------------------------------------------
# Global exception handler — ensure every error is JSON with an 'error' field
# ---------------------------------------------------------------------------
@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
status = exc.status_code
detail = exc.detail
# If detail is already a dict with 'error', use it; otherwise wrap it
if isinstance(detail, dict) and "error" not in detail:
body = {"error": detail}
elif isinstance(detail, dict):
body = detail
else:
body = {"error": str(detail)}
return JSONResponse(status_code=status, content=body)
# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host=HOST, port=PORT)
fastapi uvicorn python-jose[cryptography] passlib[bcrypt] python-multipart
......................... [100%] 25 passed in 0.44s score: 25/25 (100%)
import json
import os
import re
import secrets
import threading
from http.server import HTTPServer, BaseHTTPRequestHandler
from urllib.parse import urlparse, parse_qs
# ── In-memory data stores ────────────────────────────────────────────────────
users = {
"alice": {"username": "alice", "password": "password123"},
"bob": {"username": "bob", "password": "hunter2"},
}
products = [
{"id": 1, "name": "Wireless Mouse", "price": 29.99, "stock": 20},
{"id": 2, "name": "Mechanical Keyboard", "price": 79.99, "stock": 15},
{"id": 3, "name": "USB-C Hub", "price": 49.99, "stock": 10},
]
orders = [] # list of order dicts
order_counter = 0 # auto-incrementing order id
# Thread-safe lock for order creation
lock = threading.Lock()
# ── Token management ─────────────────────────────────────────────────────────
active_tokens = {} # token_string -> username
# ── Helpers ──────────────────────────────────────────────────────────────────
def _find_product(product_id):
for p in products:
if p["id"] == product_id:
return p
return None
def _json_response(handler, status_code, data):
body = json.dumps(data).encode("utf-8")
handler.send_response(status_code)
handler.send_header("Content-Type", "application/json")
handler.send_header("Content-Length", str(len(body)))
handler.end_headers()
handler.wfile.write(body)
def _read_json_body(handler):
content_length = int(handler.headers.get("Content-Length", 0))
raw = handler.rfile.read(content_length)
return json.loads(raw)
def _get_username_from_token(handler):
auth_header = handler.headers.get("Authorization", "")
if auth_header.startswith("Bearer "):
token = auth_header[7:]
return active_tokens.get(token)
return None
# ── Request Handler ──────────────────────────────────────────────────────────
class StoreHandler(BaseHTTPRequestHandler):
# Silence default stderr logging
def log_message(self, format, *args):
pass
def do_GET(self):
parsed = urlparse(self.path)
path = parsed.path.rstrip("/") or "/"
qs = parse_qs(parsed.query)
# GET /products
if path == "/products":
self._handle_list_products(qs)
return
# GET /products/<id>
m = re.match(r"^/products/(\d+)$", path)
if m:
product_id = int(m.group(1))
self._handle_get_product(product_id)
return
# GET /orders (protected)
if path == "/orders":
self._handle_list_orders()
return
# GET /orders/<id> (protected)
m = re.match(r"^/orders/(\d+)$", path)
if m:
order_id = int(m.group(1))
self._handle_get_order(order_id)
return
_json_response(self, 404, {"error": "Not found"})
def do_POST(self):
parsed = urlparse(self.path)
path = parsed.path.rstrip("/") or "/"
# POST /login
if path == "/login":
self._handle_login()
return
# POST /orders (protected)
if path == "/orders":
self._handle_create_order()
return
_json_response(self, 404, {"error": "Not found"})
# ── Public endpoints ───────────────────────────────────────────────────
def _handle_list_products(self, qs):
max_price_param = qs.get("max_price", [None])[0]
result = products
if max_price_param is not None:
try:
max_price = float(max_price_param)
result = [p for p in result if p["price"] <= max_price]
except (ValueError, TypeError):
pass # invalid max_price → return all
_json_response(self, 200, result)
def _handle_get_product(self, product_id):
product = _find_product(product_id)
if product is None:
_json_response(self, 404, {"error": "Product not found"})
return
_json_response(self, 200, product)
# ── Auth endpoint ──────────────────────────────────────────────────────
def _handle_login(self):
try:
body = _read_json_body(self)
except Exception:
_json_response(self, 400, {"error": "Invalid JSON body"})
return
if not isinstance(body, dict):
_json_response(self, 400, {"error": "Invalid JSON body"})
return
username = body.get("username")
password = body.get("password")
if not username or not password:
_json_response(self, 400, {"error": "Missing username or password"})
return
user = users.get(username)
if user is None or user["password"] != password:
_json_response(self, 401, {"error": "Invalid credentials"})
return
token = secrets.token_hex(32)
active_tokens[token] = username
_json_response(self, 200, {"token": token})
# ── Protected endpoints ────────────────────────────────────────────────
def _handle_list_orders(self):
username = _get_username_from_token(self)
if username is None:
_json_response(self, 401, {"error": "Missing or invalid token"})
return
user_orders = [o for o in orders if o["username"] == username]
_json_response(self, 200, user_orders)
def _handle_get_order(self, order_id):
username = _get_username_from_token(self)
if username is None:
_json_response(self, 401, {"error": "Missing or invalid token"})
return
for o in orders:
if o["id"] == order_id:
if o["username"] != username:
_json_response(self, 404, {"error": "Order not found"})
return
_json_response(self, 200, o)
return
_json_response(self, 404, {"error": "Order not found"})
def _handle_create_order(self):
username = _get_username_from_token(self)
if username is None:
_json_response(self, 401, {"error": "Missing or invalid token"})
return
# Parse body
try:
body = _read_json_body(self)
except Exception:
_json_response(self, 400, {"error": "Invalid JSON body"})
return
if not isinstance(body, dict):
_json_response(self, 400, {"error": "Invalid JSON body"})
return
# Validate required fields
if "product_id" not in body or "quantity" not in body:
_json_response(self, 400, {"error": "Missing product_id or quantity"})
return
product_id = body["product_id"]
quantity = body["quantity"]
# Validate types (bool is subclass of int in Python, reject it)
if isinstance(product_id, bool) or not isinstance(product_id, int):
_json_response(self, 400, {"error": "product_id must be an integer"})
return
if isinstance(quantity, bool) or not isinstance(quantity, int):
_json_response(self, 400, {"error": "quantity must be an integer"})
return
if quantity < 1:
_json_response(self, 400, {"error": "quantity must be at least 1"})
return
# Find product
product = _find_product(product_id)
if product is None:
_json_response(self, 404, {"error": "Product not found"})
return
# Check stock and create order (thread-safe)
with lock:
if product["stock"] < quantity:
_json_response(self, 409, {"error": "Insufficient stock"})
return
product["stock"] -= quantity
global order_counter
order_counter += 1
order = {
"id": order_counter,
"username": username,
"product_id": product_id,
"quantity": quantity,
"total": round(product["price"] * quantity, 2),
}
orders.append(order)
_json_response(self, 201, order)
# ── Run ──────────────────────────────────────────────────────────────────────
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"Server running on http://{host}:{port}")
server.serve_forever()