......................... [100%] 25 passed in 0.60s score: 25/25 (100%)
from fastapi import FastAPI, Header, HTTPException, Request, status, Depends
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field
from typing import List, Optional, Dict
import uvicorn
import os
import secrets
app = FastAPI()
# --- Data Models ---
class User(BaseModel):
username: str
password: str
class Product(BaseModel):
id: int
name: str
price: float
stock: int
class Order(BaseModel):
id: int
product_id: int
quantity: int
total: float
username: str
class OrderRequest(BaseModel):
product_id: int
quantity: int = Field(..., gt=0)
# --- In-Memory Store ---
users_db = {
"alice": "password123",
"bob": "hunter2"
}
products_db = {
1: Product(id=1, name="Laptop", price=999.99, stock=10),
2: Product(id=2, name="Mouse", price=25.50, stock=20),
3: Product(id=3, name="Keyboard", price=75.00, stock=15),
}
orders_db: Dict[int, Order] = {}
tokens_db: Dict[str, str] = {} # token -> username
order_id_counter = 1
# --- Helpers ---
def get_authenticated_user(authorization: Optional[str] = Header(None)):
if not authorization or not authorization.startswith("Bearer "):
raise HTTPException(status_code=401, detail="Missing or invalid token")
token = authorization.split(" ")[1]
username = tokens_db.get(token)
if not username:
raise HTTPException(status_code=401, detail="Invalid token")
return username
# --- Endpoints ---
@app.post("/login")
async def login(request: Request):
try:
body = await request.json()
except Exception:
raise HTTPException(status_code=400, detail="Invalid JSON 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_db.get(username) == password:
token = secrets.token_hex(16)
tokens_db[token] = username
return {"token": token}
raise HTTPException(status_code=401, detail="Invalid credentials")
@app.get("/products")
async def list_products(max_price: Optional[str] = None):
products = list(products_db.values())
if max_price:
try:
price_limit = float(max_price)
products = [p for p in products if p.price <= price_limit]
except ValueError:
pass # Return all products if max_price is not a valid number
return products
@app.get("/products/{product_id}")
async def get_product(product_id: int):
product = products_db.get(product_id)
if not product:
raise HTTPException(status_code=404, detail="Product not found")
return product
@app.post("/orders", status_code=status.HTTP_201_CREATED)
async def create_order(
request: Request,
username: str = Depends(get_authenticated_user)
):
try:
body = await request.json()
# Manual validation to match specific requirements (non-int check etc)
product_id = body.get("product_id")
quantity = body.get("quantity")
if product_id is None or quantity is None:
raise ValueError("Missing product_id or quantity")
if not isinstance(product_id, int) or not isinstance(quantity, int):
raise ValueError("product_id and quantity must be integers")
if quantity < 1:
raise ValueError("quantity must be at least 1")
except Exception as e:
detail = str(e) if isinstance(e, ValueError) else "Invalid JSON body"
raise HTTPException(status_code=400, detail=detail)
product = products_db.get(product_id)
if not product:
raise HTTPException(status_code=404, detail="Product not found")
if product.stock < quantity:
raise HTTPException(status_code=409, detail="Insufficient stock")
# Update stock
product.stock -= quantity
global order_id_counter
order = Order(
id=order_id_counter,
product_id=product_id,
quantity=quantity,
total=product.price * quantity,
username=username
)
orders_db[order_id_counter] = 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_authenticated_user)):
user_orders = [
{"id": o.id, "product_id": o.product_id, "quantity": o.quantity, "total": o.total}
for o in orders_db.values() if o.username == username
]
return user_orders
@app.get("/orders/{order_id}")
async def get_order(order_id: int, username: str = Depends(get_authenticated_user)):
order = orders_db.get(order_id)
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
}
# Custom error handler to ensure JSON response with "error" field
@app.exception_handler(HTTPException)
async def http_exception_handler(request, exc):
return JSONResponse(
status_code=exc.status_code,
content={"error": exc.detail},
)
if __name__ == "__main__":
host = os.environ.get("HOST", "127.0.0.1")
port = int(os.environ.get("PORT", "8000"))
uvicorn.run(app, host=host, port=port)
fastapi\nuvicorn
......................... [100%] 25 passed in 0.65s score: 25/25 (100%)
from fastapi import FastAPI, Request, HTTPException, Depends, Header
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field
from typing import List, Optional
import uvicorn
import os
import secrets
app = FastAPI()
# --- Data Store ---
users = {
"alice": "password123",
"bob": "hunter2"
}
products = {
1: {"id": 1, "name": "Laptop", "price": 999.99, "stock": 10},
2: {"id": 2, "name": "Mouse", "price": 25.50, "stock": 20},
3: {"id": 3, "name": "Keyboard", "price": 75.00, "stock": 15},
}
# Mapping of token -> username
sessions = {}
orders = []
order_id_counter = 1
# --- Models ---
class LoginRequest(BaseModel):
username: str
password: str
class OrderRequest(BaseModel):
product_id: int
quantity: int = Field(..., gt=0)
# --- Auth Helper ---
async def get_current_user(authorization: Optional[str] = Header(None)):
if not authorization or not authorization.startswith("Bearer "):
raise HTTPException(status_code=401, detail="Missing or invalid token")
token = authorization.split(" ")[1]
username = sessions.get(token)
if not username:
raise HTTPException(status_code=401, detail="Invalid token")
return username
# --- Exception Handlers ---
@app.exception_handler(HTTPException)
async def http_exception_handler(request, exc):
return JSONResponse(
status_code=exc.status_code,
content={"error": exc.detail},
)
@app.exception_handler(Exception)
async def general_exception_handler(request, exc):
return JSONResponse(
status_code=400,
content={"error": "Malformed request or internal server error"},
)
# --- Routes ---
@app.post("/login")
async def login(request: Request):
try:
body = await request.json()
except Exception:
return JSONResponse(status_code=400, content={"error": "Invalid JSON"})
username = body.get("username")
password = body.get("password")
if username is None or password is None:
return JSONResponse(status_code=400, content={"error": "Missing username or password"})
if users.get(username) == password:
token = secrets.token_hex(16)
sessions[token] = username
return {"token": token}
return JSONResponse(status_code=401, content={"error": "Invalid credentials"})
@app.get("/products")
async def get_products(max_price: Optional[str] = None):
product_list = list(products.values())
if max_price:
try:
price_limit = float(max_price)
product_list = [p for p in product_list if p["price"] <= price_limit]
except ValueError:
pass
return product_list
@app.get("/products/{product_id}")
async def get_product(product_id: int):
product = products.get(product_id)
if not product:
raise HTTPException(status_code=404, detail="Product not found")
return product
@app.post("/orders", status_code=201)
async def create_order(request: Request, username: str = Depends(get_current_user)):
try:
body = await request.json()
# Manual validation to ensure strict 400 for non-ints or missing fields as per reqs
pid = body.get("product_id")
qty = body.get("quantity")
if not isinstance(pid, int) or not isinstance(qty, int) or qty < 1:
raise ValueError("Invalid product_id or quantity")
except Exception:
return JSONResponse(status_code=400, content={"error": "Malformed request body"})
product = products.get(pid)
if not product:
raise HTTPException(status_code=404, detail="Product not found")
if product["stock"] < qty:
return JSONResponse(status_code=409, content={"error": "Insufficient stock"})
# Process order
global order_id_counter
total = product["price"] * qty
order = {
"id": order_id_counter,
"product_id": pid,
"quantity": qty,
"total": total,
"username": username
}
orders.append(order)
order_id_counter += 1
# Update stock
product["stock"] -= qty
# Return order without the internal 'username' field
return {k: v for k, v in order.items() if k != "username"}
@app.get("/orders")
async def get_orders(username: str = Depends(get_current_user)):
user_orders = [
{k: v for k, v in o.items() if k != "username"}
for o in orders if o["username"] == username
]
return user_orders
@app.get("/orders/{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 {k: v for k, v in order.items() if k != "username"}
if __name__ == "__main__":
host = os.environ.get("HOST", "127.0.0.1")
port = int(os.environ.get("PORT", "8000"))
uvicorn.run(app, host=host, port=port)
fastapi uvicorn pydantic
......................... [100%] 25 passed in 0.65s score: 25/25 (100%)
from fastapi import FastAPI, Request, HTTPException, Header, Depends, status
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field
from typing import List, Optional
import uvicorn
import os
import secrets
app = FastAPI()
# --- Data Models ---
class User(BaseModel):
username: str
password: str
class LoginRequest(BaseModel):
username: str
password: str
class Product(BaseModel):
id: int
name: str
price: float
stock: int
class OrderRequest(BaseModel):
product_id: int
quantity: int = Field(..., gt=0)
class Order(BaseModel):
id: int
user_id: str
product_id: int
quantity: int
total: float
# --- In-Memory Storage ---
users = {
"alice": "password123",
"bob": "hunter2"
}
products = {
1: Product(id=1, name="Laptop", price=999.99, stock=10),
2: Product(id=2, name="Mouse", price=25.50, stock=20),
3: Product(id=3, name="Keyboard", price=50.00, stock=15),
}
orders = {}
tokens = {} # token -> username
order_counter = 1
# --- Helpers ---
async def get_current_user(authorization: Optional[str] = Header(None)):
if not authorization or not authorization.startswith("Bearer "):
raise HTTPException(status_code=401, detail="Missing or invalid token")
token = authorization.split(" ")[1]
username = tokens.get(token)
if not username:
raise HTTPException(status_code=401, detail="Invalid token")
return username
# Custom error handler to ensure all responses are JSON
@app.exception_handler(HTTPException)
async def http_exception_handler(request, exc):
return JSONResponse(
status_code=exc.status_code,
content={"error": exc.detail},
)
@app.exception_handler(Exception)
async def general_exception_handler(request, exc):
return JSONResponse(
status_code=500,
content={"error": "Internal server error"},
)
# --- Endpoints ---
@app.post("/login")
async def login(request: Request):
try:
body = await request.json()
except Exception:
return JSONResponse(status_code=400, content={"error": "Invalid JSON"})
username = body.get("username")
password = body.get("password")
if username is None or password is None:
return JSONResponse(status_code=400, content={"error": "Missing username or password"})
if users.get(username) == password:
token = secrets.token_hex(16)
tokens[token] = username
return {"token": token}
return JSONResponse(status_code=401, content={"error": "Invalid credentials"})
@app.get("/products")
async def get_products(max_price: Optional[str] = None):
product_list = list(products.values())
if max_price is not None:
try:
limit = float(max_price)
product_list = [p for p in product_list if p.price <= limit]
except ValueError:
pass # Return all if not a valid number
return product_list
@app.get("/products/{id}")
async def get_product(id: int):
product = products.get(id)
if not product:
raise HTTPException(status_code=404, detail="Product not found")
return product
@app.post("/orders", status_code=201)
async def create_order(request: Request, username: str = Depends(get_current_user)):
try:
body = await request.json()
except Exception:
return JSONResponse(status_code=400, content={"error": "Invalid JSON"})
product_id = body.get("product_id")
quantity = body.get("quantity")
if product_id is None or quantity is None:
return JSONResponse(status_code=400, content={"error": "Missing product_id or quantity"})
if not isinstance(product_id, int) or not isinstance(quantity, int) or quantity < 1:
return JSONResponse(status_code=400, content={"error": "product_id and quantity must be positive integers"})
product = products.get(product_id)
if not product:
return JSONResponse(status_code=404, content={"error": "Product not found"})
if product.stock < quantity:
return JSONResponse(status_code=409, content={"error": "Insufficient stock"})
global order_counter
order_id = order_counter
order_counter += 1
total = product.price * quantity
product.stock -= quantity
order = Order(id=order_id, user_id=username, product_id=product_id, quantity=quantity, total=total)
orders[order_id] = order
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.values() if o.user_id == username
]
return user_orders
@app.get("/orders/{id}")
async def get_order(id: int, username: str = Depends(get_current_user)):
order = orders.get(id)
if not order or order.user_id != 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__":
host = os.environ.get("HOST", "127.0.0.1")
port = int(os.environ.get("PORT", "8000"))
uvicorn.run(app, host=host, port=port)
fastapi uvicorn pydantic
......................... [100%] 25 passed in 0.60s score: 25/25 (100%)
from fastapi import FastAPI, Request, HTTPException, Depends, Header, status
from pydantic import BaseModel, Field
from typing import List, Optional, Dict
import os
import uvicorn
import secrets
app = FastAPI()
# --- Data Models ---
class User(BaseModel):
username: str
password: str
class Product(BaseModel):
id: int
name: str
price: float
stock: int
class Order(BaseModel):
id: int
user_id: str
product_id: int
quantity: int
total: float
class LoginRequest(BaseModel):
username: str
password: str
class OrderRequest(BaseModel):
product_id: int
quantity: int = Field(..., gt=0)
# --- In-Memory Storage ---
users = {
"alice": {"password": "password123", "id": "u1"},
"bob": {"password": "hunter2", "id": "u2"},
}
products = {
1: Product(id=1, name="Laptop", price=999.99, stock=10),
2: Product(id=2, name="Mouse", price=25.50, stock=20),
3: Product(id=3, name="Keyboard", price=45.00, stock=15),
}
orders = {}
order_id_counter = 1
tokens = {} # token -> user_id
# --- Helpers ---
def get_current_user(authorization: Optional[str] = Header(None)):
if not authorization or not authorization.startswith("Bearer "):
raise HTTPException(status_code=401, detail="Missing or invalid token")
token = authorization.split(" ")[1]
user_id = tokens.get(token)
if not user_id:
raise HTTPException(status_code=401, detail="Invalid token")
return user_id
# --- Endpoints ---
@app.post("/login")
async def login(request: Request):
try:
body = await request.json()
username = body.get("username")
password = body.get("password")
if not username or not password:
raise HTTPException(status_code=400, detail="Missing username or password")
except Exception:
raise HTTPException(status_code=400, detail="Invalid JSON body")
user = users.get(username)
if not user or user["password"] != password:
raise HTTPException(status_code=401, detail="Invalid credentials")
token = secrets.token_hex(16)
tokens[token] = user["id"]
return {"token": token}
@app.get("/products")
async def get_products(max_price: Optional[str] = None):
product_list = list(products.values())
if max_price:
try:
limit = float(max_price)
product_list = [p for p in product_list if p.price <= limit]
except ValueError:
pass
return product_list
@app.get("/products/{id}")
async def get_product(id: int):
product = products.get(id)
if not product:
raise HTTPException(status_code=404, detail="Product not found")
return product
@app.post("/orders", status_code=201)
async def create_order(request: Request, user_id: str = Depends(get_current_user)):
try:
body = await request.json()
# Manual validation to ensure strict 400 for non-integers etc as per spec
product_id = body.get("product_id")
quantity = body.get("quantity")
if product_id is None or quantity is None:
raise ValueError("Missing product_id or quantity")
if not isinstance(product_id, int) or not isinstance(quantity, int):
raise ValueError("product_id and quantity must be integers")
if quantity < 1:
raise ValueError("quantity must be at least 1")
except Exception as e:
detail = str(e) if isinstance(e, ValueError) else "Invalid JSON body"
raise HTTPException(status_code=400, detail=detail)
product = products.get(product_id)
if not product:
raise HTTPException(status_code=404, detail="Product not found")
if product.stock < quantity:
raise HTTPException(status_code=409, detail="Insufficient stock")
# Process order
global order_id_counter
total = product.price * quantity
order = Order(id=order_id_counter, user_id=user_id, product_id=product_id, quantity=quantity, total=total)
orders[order_id_counter] = order
order_id_counter += 1
# 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 get_orders(user_id: 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.values() if o.user_id == user_id
]
return user_orders
@app.get("/orders/{id}")
async def get_order(id: int, user_id: str = Depends(get_current_user)):
order = orders.get(id)
if not order or order.user_id != user_id:
raise HTTPException(status_code=404, detail="Order not found")
return {
"id": order.id,
"product_id": order.product_id,
"quantity": order.quantity,
"total": order.total
}
# Custom exception handler to ensure JSON response for all HTTPExceptions
from fastapi.responses import JSONResponse
from fastapi import Request
@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
return JSONResponse(
status_code=exc.status_code,
content={"error": exc.detail},
)
if __name__ == "__main__":
host = os.environ.get("HOST", "127.0.0.1")
port = int(os.environ.get("PORT", "8000"))
uvicorn.run(app, host=host, port=port)
fastapi uvicorn pydantic
....................F.FF. [100%]
=================================== FAILURES ===================================
_____________________ test_order_exceeding_stock_conflict ______________________
base_url = 'http://127.0.0.1:45091'
alice_token = '7323b17785a1f96d28f93964a8d748d1'
def test_order_exceeding_stock_conflict(base_url, alice_token):
_, before, _ = _request("GET", f"{base_url}/products/1")
start = before["stock"]
status, parsed, ct = _request_h("POST", f"{base_url}/orders", token=alice_token,
body={"product_id": 1, "quantity": start + 1000})
> _assert_json_error(status, parsed, ct, 409)
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:318:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
status = 409, parsed = {'detail': 'Insufficient stock'}, ct = 'application/json'
expect_status = 409
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': 'Insufficient stock'}
E assert (True and (None or None))
E + where True = isinstance({'detail': 'Insufficient stock'}, dict)
E + and None = <built-in method get of dict object at 0x7fa50aed74c0>('error')
E + where <built-in method get of dict object at 0x7fa50aed74c0> = {'detail': 'Insufficient stock'}.get
E + and None = <built-in method get of dict object at 0x7fa50aed74c0>('message')
E + where <built-in method get of dict object at 0x7fa50aed74c0> = {'detail': 'Insufficient stock'}.get
/home/lzieniew/Documents/my_local_llm_benchmarks/tasks/py-store-api/tests/test_store_api.py:91: AssertionError
____________________________ test_get_missing_order ____________________________
base_url = 'http://127.0.0.1:45091'
alice_token = '7323b17785a1f96d28f93964a8d748d1'
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 or access denied'}
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 or access denied'}
E assert (True and (None or None))
E + where True = isinstance({'detail': 'Order not found or access denied'}, dict)
E + and None = <built-in method get of dict object at 0x7fa50b01fc40>('error')
E + where <built-in method get of dict object at 0x7fa50b01fc40> = {'detail': 'Order not found or access denied'}.get
E + and None = <built-in method get of dict object at 0x7fa50b01fc40>('message')
E + where <built-in method get of dict object at 0x7fa50b01fc40> = {'detail': 'Order not found or access denied'}.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:45091'
alice_token = '7323b17785a1f96d28f93964a8d748d1'
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 or access denied'}
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 or access denied'}
E assert (True and (None or None))
E + where True = isinstance({'detail': 'Order not found or access denied'}, dict)
E + and None = <built-in method get of dict object at 0x7fa50b18a140>('error')
E + where <built-in method get of dict object at 0x7fa50b18a140> = {'detail': 'Order not found or access denied'}.get
E + and None = <built-in method get of dict object at 0x7fa50b18a140>('message')
E + where <built-in method get of dict object at 0x7fa50b18a140> = {'detail': 'Order not found or access denied'}.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_order_exceeding_stock_conflict
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
3 failed, 22 passed in 0.63s
score: 22/25 (88%)
from fastapi import FastAPI, Header, HTTPException, Query, status
from pydantic import BaseModel, Field
from typing import List, Optional
import uvicorn
import os
import secrets
app = FastAPI()
# --- Data Models ---
class User(BaseModel):
username: str
password: str
class Product(BaseModel):
id: int
name: str
price: float
stock: int
class Order(BaseModel):
id: int
product_id: int
quantity: int
total: float
user_token: str
class LoginRequest(BaseModel):
username: str
password: str
class OrderRequest(BaseModel):
product_id: int
quantity: int = Field(..., gt=0)
# --- In-Memory Store ---
users_db = {
"alice": "password123",
"bob": "hunter2"
}
products_db = {
1: Product(id=1, name="Laptop", price=999.99, stock=10),
2: Product(id=2, name="Mouse", price=25.50, stock=20),
3: Product(id=3, name="Keyboard", price=49.99, stock=15),
}
# token -> username
tokens_db = {}
orders_db = {}
order_counter = 1
# --- Helpers ---
def get_user_from_token(token: str):
if not token or not token.startswith("Bearer "):
raise HTTPException(status_code=401, detail="Invalid or missing token")
actual_token = token[7:]
username = tokens_db.get(actual_token)
if not username:
raise HTTPException(status_code=401, detail="Invalid or expired token")
return username
# --- Endpoints ---
@app.post("/login")
async def login(payload: LoginRequest):
# FastAPI handles 400 for malformed JSON or missing fields automatically via Pydantic
password = users_db.get(payload.username)
if password and password == payload.password:
token = secrets.token_hex(16)
tokens_db[token] = payload.username
return {"token": token}
raise HTTPException(status_code=401, detail="Invalid credentials")
@app.get("/products")
async def get_products(max_price: Optional[str] = Query(None)):
products = list(products_db.values())
if max_price:
try:
limit = float(max_price)
products = [p for p in products if p.price <= limit]
except ValueError:
pass # Return all products if max_price is not a valid number
return products
@app.get("/products/{id}")
async def get_product(id: int):
product = products_db.get(id)
if not product:
raise HTTPException(status_code=404, detail="Product not found")
return product
@app.post("/orders", status_code=status.HTTP_201_CREATED)
async def create_order(payload: OrderRequest, authorization: Optional[str] = Header(None)):
username = get_user_from_token(authorization)
product = products_db.get(payload.product_id)
if not product:
raise HTTPException(status_code=404, detail="Product not found")
if payload.quantity > product.stock:
raise HTTPException(status_code=409, detail="Insufficient stock")
# Process order
global order_counter
order_id = order_counter
order_counter += 1
total = product.price * payload.quantity
product.stock -= payload.quantity
# Store token in order to link to user (simplest way for this exercise)
# We need the token for ownership check
token = authorization[7:]
new_order = Order(
id=order_id,
product_id=payload.product_id,
quantity=payload.quantity,
total=total,
user_token=token
)
orders_db[order_id] = new_order
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(authorization: Optional[str] = Header(None)):
token = get_user_from_token(authorization) # This returns username, but we need the token for lookup
# Redefining helper or just getting token here
actual_token = authorization[7:]
user_orders = [
{"id": o.id, "product_id": o.product_id, "quantity": o.quantity, "total": o.total}
for o in orders_db.values() if o.user_token == actual_token
]
return user_orders
@app.get("/orders/{id}")
async def get_order(id: int, authorization: Optional[str] = Header(None)):
actual_token = authorization[7:] if authorization else ""
get_user_from_token(authorization) # Validate token
order = orders_db.get(id)
if not order or order.user_token != actual_token:
raise HTTPException(status_code=404, detail="Order not found or access denied")
return {
"id": order.id,
"product_id": order.product_id,
"quantity": order.quantity,
"total": order.total
}
# Custom error handler to ensure 400/422 responses are JSON with "detail" or "error"
# FastAPI's default RequestValidationError returns 422, but requirements ask for 400
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
return JSONResponse(
status_code=400,
content={"error": "Malformed request body"},
)
if __name__ == "__main__":
host = os.getenv("HOST", "127.0.0.1")
port = int(os.getenv("PORT", "8000"))
uvicorn.run(app, host=host, port=port)
fastapi uvicorn pydantic