Spaces:
Sleeping
Sleeping
| import time | |
| from fastapi import APIRouter, Header, HTTPException, Request, Query | |
| from pydantic import BaseModel | |
| from typing import Optional | |
| router = APIRouter(prefix="/mock_api", tags=["mock"]) | |
| _issued_tokens = set() | |
| _request_log = {} | |
| LOG_PAGES = { | |
| None: {"items": list(range(10)), "next_cursor": "cur_abc", "has_more": True}, | |
| "cur_abc": {"items": list(range(10, 20)), "next_cursor": "cur_def", "has_more": True}, | |
| "cur_def": {"items": list(range(20, 25)), "next_cursor": None, "has_more": False}, | |
| } | |
| class TokenRequest(BaseModel): | |
| client_id: str | |
| client_secret: str | |
| async def get_token(body: TokenRequest): | |
| if body.client_id == "abc" and body.client_secret == "xyz": | |
| token = f"tok_{int(time.time())}" | |
| _issued_tokens.add(token) | |
| return {"access_token": token, "expires_in": 60} | |
| raise HTTPException(status_code=401, detail="Invalid client credentials") | |
| async def get_users(authorization: Optional[str] = Header(default=None)): | |
| if not authorization or not authorization.startswith("Bearer "): | |
| raise HTTPException(status_code=401, detail="Authorization header missing or malformed") | |
| token = authorization.split(" ")[1] | |
| if token not in _issued_tokens and token != "demo_token_123": | |
| raise HTTPException(status_code=401, detail="Invalid token") | |
| return {"users": [{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}]} | |
| async def create_item(request: Request): | |
| ct = request.headers.get("content-type", "") | |
| if "application/json" not in ct: | |
| raise HTTPException(status_code=415, detail="Content-Type must be application/json") | |
| body = await request.json() | |
| if "name" not in body: | |
| raise HTTPException(status_code=422, detail=[{"loc": ["body", "name"], "msg": "field required"}]) | |
| return {"item_id": 42, "name": body["name"], "created": True} | |
| async def search(q: Optional[str] = Query(default=None)): | |
| if not q: | |
| raise HTTPException(status_code=422, detail=[{"loc": ["query", "q"], "msg": "field required"}]) | |
| return {"results": [{"title": f"Result for {q}", "score": 0.95}], "total": 1} | |
| class OrderBody(BaseModel): | |
| product_id: int | |
| qty: int | |
| async def create_order(order: OrderBody): | |
| return {"order_id": 999, "product_id": order.product_id, "qty": order.qty, "status": "confirmed"} | |
| class AddressModel(BaseModel): | |
| street: str | |
| city: str | |
| class ProfileBody(BaseModel): | |
| name: str | |
| address: AddressModel | |
| async def update_profile(profile: ProfileBody): | |
| return {"name": profile.name, "address": profile.address.model_dump(), "updated": True} | |
| async def protected(authorization: Optional[str] = Header(default=None)): | |
| if not authorization or not authorization.startswith("Bearer "): | |
| raise HTTPException(status_code=401, detail="Authorization header missing") | |
| token = authorization.split(" ")[1] | |
| if token not in _issued_tokens: | |
| raise HTTPException(status_code=401, detail="Invalid or expired token") | |
| return {"data": "secret_payload", "accessed_at": int(time.time())} | |
| async def rate_limited(request: Request, x_retry_after: Optional[str] = Header(default=None)): | |
| client_id = request.client.host if request.client else "default" | |
| now = time.time() | |
| window = [t for t in _request_log.get(client_id, []) if now - t < 5] | |
| if len(window) >= 3 and not x_retry_after: | |
| _request_log[client_id] = window | |
| raise HTTPException(status_code=429, detail="Rate limit exceeded", headers={"Retry-After": "2"}) | |
| window.append(now) | |
| _request_log[client_id] = window | |
| return {"data": "rate_limited_resource", "requests_in_window": len(window)} | |
| async def get_logs(cursor: Optional[str] = Query(default=None)): | |
| if cursor not in LOG_PAGES: | |
| raise HTTPException(status_code=400, detail=f"Invalid cursor: {cursor}") | |
| return LOG_PAGES[cursor] | |