{"id":10021259,"date":"2026-02-19T12:38:50","date_gmt":"2026-02-19T04:38:50","guid":{"rendered":"https:\/\/itmax.cn\/?p=10021259"},"modified":"2026-02-19T12:38:50","modified_gmt":"2026-02-19T04:38:50","slug":"%e5%bc%82%e6%ad%a5-web-%e6%a1%86%e6%9e%b6%e5%ba%95%e5%b1%82%ef%bc%9afastapi-%e9%ab%98%e6%80%a7%e8%83%bd%e5%8e%9f%e7%90%86%e5%89%96%e6%9e%90","status":"publish","type":"post","link":"https:\/\/itmax.cn\/?p=10021259","title":{"rendered":"\u5f02\u6b65 Web \u6846\u67b6\u5e95\u5c42\uff1aFastAPI \u9ad8\u6027\u80fd\u539f\u7406\u5256\u6790"},"content":{"rendered":"

\u5f02\u6b65 Web \u6846\u67b6\u5e95\u5c42\uff1aFastAPI \u9ad8\u6027\u80fd\u539f\u7406\u5256\u6790<\/h2>\n

FastAPI \u662f\u73b0\u4ee3 Python Web \u5f00\u53d1\u7684\u70ed\u95e8\u9009\u62e9\uff0c\u5b83\u57fa\u4e8e Starlette \u548c Pydantic \u6784\u5efa\uff0c\u63d0\u4f9b\u4e86\u9ad8\u6027\u80fd\u548c\u5f00\u53d1\u6548\u7387\u7684\u5b8c\u7f8e\u5e73\u8861\u3002\u672c\u6587\u5c06\u6df1\u5165\u5256\u6790 FastAPI \u7684\u5e95\u5c42\u539f\u7406\uff0c\u4ece\u5f02\u6b65\u5904\u7406\u673a\u5236\u5230\u6027\u80fd\u4f18\u5316\u7b56\u7565\uff0c\u5e2e\u52a9\u8bfb\u8005\u5168\u9762\u7406\u89e3 FastAPI \u7684\u9ad8\u6027\u80fd\u5b9e\u73b0\u3002<\/p>\n

FastAPI \u67b6\u6784\u6982\u89c8<\/h3>\n

FastAPI \u7684\u67b6\u6784\u8bbe\u8ba1\u5145\u5206\u5229\u7528\u4e86 Python \u7684\u5f02\u6b65\u7279\u6027\uff0c\u901a\u8fc7 ASGI \u534f\u8bae\u5b9e\u73b0\u9ad8\u6027\u80fd\u7684 Web \u670d\u52a1\u3002<\/p>\n

from fastapi import FastAPI\nimport uvicorn\n\napp = FastAPI()\n\n@app.get("\/")\nasync def read_root():\n    return {"message": "Hello World"}\n\n@app.get("\/items\/{item_id}")\nasync def read_item(item_id: int, q: str = None):\n    return {"item_id": item_id, "q": q}\n\nif __name__ == "__main__":\n    uvicorn.run(app, host="0.0.0.0", port=8000)<\/code><\/pre>\n
ASGI \u534f\u8bae\u57fa\u7840<\/h3>\n
ASGI\uff08Asynchronous Server Gateway Interface\uff09\u662f FastAPI \u9ad8\u6027\u80fd\u7684\u57fa\u7840\uff0c\u5b83\u652f\u6301\u5f02\u6b65\u8bf7\u6c42\u5904\u7406\u3002<\/p>\n
import asyncio\n\nasync def asgi_app(scope, receive, send):\n    if scope['type'] == 'http':\n        await send({\n            'type': 'http.response.start',\n            'status': 200,\n            'headers': [[b'content-type', b'text\/plain']],\n        })\n        await send({\n            'type': 'http.response.body',\n            'body': b'Hello from ASGI!',\n        })\n\nasync def test_asgi():\n    scope = {\n        'type': 'http',\n        'method': 'GET',\n        'path': '\/',\n    }\n    \n    async def receive():\n        return {'type': 'http.request', 'body': b''}\n    \n    messages = []\n    async def send(message):\n        messages.append(message)\n    \n    await asgi_app(scope, receive, send)\n    print("ASGI \u54cd\u5e94:", messages)\n\nasyncio.run(test_asgi())<\/code><\/pre>\n
\u5f02\u6b65\u8def\u7531\u5904\u7406<\/h3>\n
FastAPI \u7684\u8def\u7531\u5904\u7406\u5b8c\u5168\u57fa\u4e8e\u5f02\u6b65\u673a\u5236\uff0c\u80fd\u591f\u9ad8\u6548\u5904\u7406\u5927\u91cf\u5e76\u53d1\u8bf7\u6c42\u3002<\/p>\n
from fastapi import FastAPI, BackgroundTasks\nimport asyncio\n\napp = FastAPI()\n\n@app.get("\/sync")\ndef sync_endpoint():\n    import time\n    time.sleep(1)\n    return {"message": "\u540c\u6b65\u54cd\u5e94"}\n\n@app.get("\/async")\nasync def async_endpoint():\n    await asyncio.sleep(1)\n    return {"message": "\u5f02\u6b65\u54cd\u5e94"}\n\n@app.post("\/background")\nasync def background_task(background_tasks: BackgroundTasks):\n    def process_data():\n        import time\n        time.sleep(2)\n        print("\u540e\u53f0\u4efb\u52a1\u5b8c\u6210")\n    \n    background_tasks.add_task(process_data)\n    return {"message": "\u4efb\u52a1\u5df2\u63d0\u4ea4"}\n\ndef demonstrate_async_routing():\n    print("\u5f02\u6b65\u8def\u7531\u5904\u7406\u6f14\u793a:")\n    print("1. \u540c\u6b65\u7aef\u70b9\u4f1a\u963b\u585e\u4e8b\u4ef6\u5faa\u73af")\n    print("2. \u5f02\u6b65\u7aef\u70b9\u53ef\u4ee5\u5e76\u53d1\u5904\u7406")\n    print("3. \u540e\u53f0\u4efb\u52a1\u4e0d\u963b\u585e\u54cd\u5e94")\n\ndemonstrate_async_routing()<\/code><\/pre>\n
FastAPI \u6027\u80fd\u67b6\u6784<\/h3>\n
graph TD\n    A[HTTP \u8bf7\u6c42] --> B[ASGI \u670d\u52a1\u5668]\n    B --> C[FastAPI \u5e94\u7528]\n    C --> D[\u8def\u7531\u5339\u914d]\n    D --> E[\u4f9d\u8d56\u6ce8\u5165]\n    E --> F[\u8bf7\u6c42\u9a8c\u8bc1]\n    F --> G[\u4e1a\u52a1\u903b\u8f91\u5904\u7406]\n    G --> H[\u54cd\u5e94\u5e8f\u5217\u5316]\n    H --> I[HTTP \u54cd\u5e94]\n    G --> J[\u540e\u53f0\u4efb\u52a1]\n    J --> K[\u4efb\u52a1\u961f\u5217]<\/code><\/pre>\n
\u4f9d\u8d56\u6ce8\u5165\u673a\u5236<\/h3>\n
FastAPI \u7684\u4f9d\u8d56\u6ce8\u5165\u7cfb\u7edf\u662f\u5176\u6838\u5fc3\u7279\u6027\u4e4b\u4e00\uff0c\u63d0\u4f9b\u4e86\u7075\u6d3b\u7684\u7ec4\u4ef6\u7ba1\u7406\u548c\u4ee3\u7801\u590d\u7528\u673a\u5236\u3002<\/p>\n
from fastapi import FastAPI, Depends, HTTPException\nfrom typing import Optional\n\napp = FastAPI()\n\nasync def get_db():\n    print("\u83b7\u53d6\u6570\u636e\u5e93\u8fde\u63a5")\n    yield {"connection": "db_connection"}\n    print("\u5173\u95ed\u6570\u636e\u5e93\u8fde\u63a5")\n\nasync def get_current_user(token: Optional[str] = None):\n    if not token:\n        raise HTTPException(status_code=401, detail="\u672a\u6388\u6743")\n    return {"user_id": 1, "username": "test_user"}\n\n@app.get("\/users\/me")\nasync def read_users_me(current_user: dict = Depends(get_current_user)):\n    return current_user\n\n@app.get("\/items\/")\nasync def read_items(db: dict = Depends(get_db)):\n    return {"db": db, "items": []}\n\ndef demonstrate_dependency_injection():\n    print("\u4f9d\u8d56\u6ce8\u5165\u6f14\u793a:")\n    print("1. \u4f9d\u8d56\u81ea\u52a8\u89e3\u6790\u548c\u6ce8\u5165")\n    print("2. \u652f\u6301\u5f02\u6b65\u4f9d\u8d56")\n    print("3. \u4f9d\u8d56\u7f13\u5b58\u548c\u590d\u7528")\n    print("4. \u5d4c\u5957\u4f9d\u8d56\u652f\u6301")\n\ndemonstrate_dependency_injection()<\/code><\/pre>\n
\u8bf7\u6c42\u9a8c\u8bc1\u4e0e\u5e8f\u5217\u5316<\/h3>\n
FastAPI \u4f7f\u7528 Pydantic \u8fdb\u884c\u8bf7\u6c42\u9a8c\u8bc1\u548c\u54cd\u5e94\u5e8f\u5217\u5316\uff0c\u786e\u4fdd\u6570\u636e\u7684\u5b89\u5168\u6027\u548c\u4e00\u81f4\u6027\u3002<\/p>\n
from fastapi import FastAPI, HTTPException\nfrom pydantic import BaseModel, Field\nfrom typing import List\n\napp = FastAPI()\n\nclass Item(BaseModel):\n    name: str = Field(..., min_length=1, max_length=50)\n    description: Optional[str] = None\n    price: float = Field(..., gt=0)\n    tax: Optional[float] = None\n\nclass User(BaseModel):\n    username: str\n    email: str\n\nitems_db = []\n\n@app.post("\/items\/", response_model=Item)\nasync def create_item(item: Item):\n    item_dict = item.dict()\n    item_dict["id"] = len(items_db) + 1\n    items_db.append(item_dict)\n    return item\n\n@app.get("\/items\/", response_model=List[Item])\nasync def read_items():\n    return items_db\n\ndef demonstrate_validation():\n    print("\u8bf7\u6c42\u9a8c\u8bc1\u6f14\u793a:")\n    \n    valid_item = Item(name="\u6d4b\u8bd5\u5546\u54c1", price=99.99)\n    print(f"\u6709\u6548\u5546\u54c1: {valid_item}")\n    \n    try:\n        invalid_item = Item(name="", price=-10)\n    except Exception as e:\n        print(f"\u65e0\u6548\u5546\u54c1\u9519\u8bef: {e}")\n\ndemonstrate_validation()<\/code><\/pre>\n
\u4e2d\u95f4\u4ef6\u673a\u5236<\/h3>\n
FastAPI \u7684\u4e2d\u95f4\u4ef6\u673a\u5236\u63d0\u4f9b\u4e86\u8bf7\u6c42\/\u54cd\u5e94\u5904\u7406\u7684\u6269\u5c55\u70b9\uff0c\u53ef\u4ee5\u7528\u4e8e\u65e5\u5fd7\u3001\u8ba4\u8bc1\u3001CORS \u7b49\u529f\u80fd\u3002<\/p>\n
from fastapi import FastAPI, Request\nimport time\n\napp = FastAPI()\n\n@app.middleware("http")\nasync def add_process_time_header(request: Request, call_next):\n    start_time = time.time()\n    response = await call_next(request)\n    process_time = time.time() - start_time\n    response.headers["X-Process-Time"] = str(process_time)\n    return response\n\n@app.get("\/")\nasync def read_root():\n    return {"message": "Hello World"}\n\ndef demonstrate_middleware():\n    print("\u4e2d\u95f4\u4ef6\u6f14\u793a:")\n    print("1. \u8bf7\u6c42\u524d\u5904\u7406")\n    print("2. \u8c03\u7528\u4e0b\u4e00\u4e2a\u5904\u7406\u5668")\n    print("3. \u54cd\u5e94\u540e\u5904\u7406")\n    print("4. \u652f\u6301\u591a\u4e2a\u4e2d\u95f4\u4ef6")\n\ndemonstrate_middleware()<\/code><\/pre>\n
\u5f02\u6b65\u6570\u636e\u5e93\u64cd\u4f5c<\/h3>\n
FastAPI \u53ef\u4ee5\u4e0e\u5f02\u6b65\u6570\u636e\u5e93\u9a71\u52a8\u914d\u5408\u4f7f\u7528\uff0c\u5b9e\u73b0\u9ad8\u6027\u80fd\u7684\u6570\u636e\u5e93\u64cd\u4f5c\u3002<\/p>\n
import asyncio\nfrom typing import List, Dict\n\nclass AsyncDatabase:\n    def __init__(self):\n        self.data: Dict[int, Dict] = {}\n        self.next_id = 1\n    \n    async def create(self, item: Dict) -> Dict:\n        await asyncio.sleep(0.01)\n        item_id = self.next_id\n        item["id"] = item_id\n        self.data[item_id] = item\n        self.next_id += 1\n        return item\n    \n    async def get(self, item_id: int) -> Dict:\n        await asyncio.sleep(0.01)\n        return self.data.get(item_id)\n    \n    async def get_all(self) -> List[Dict]:\n        await asyncio.sleep(0.01)\n        return list(self.data.values())\n\nasync def demonstrate_async_db():\n    print("\u5f02\u6b65\u6570\u636e\u5e93\u6f14\u793a:")\n    \n    db = AsyncDatabase()\n    \n    item1 = await db.create({"name": "\u5546\u54c11", "price": 100})\n    print(f"\u521b\u5efa\u5546\u54c1: {item1}")\n    \n    item2 = await db.create({"name": "\u5546\u54c12", "price": 200})\n    print(f"\u521b\u5efa\u5546\u54c1: {item2}")\n    \n    retrieved = await db.get(item1["id"])\n    print(f"\u83b7\u53d6\u5546\u54c1: {retrieved}")\n    \n    all_items = await db.get_all()\n    print(f"\u6240\u6709\u5546\u54c1: {all_items}")\n\nasyncio.run(demonstrate_async_db())<\/code><\/pre>\n
WebSocket \u652f\u6301<\/h3>\n
FastAPI \u539f\u751f\u652f\u6301 WebSocket\uff0c\u53ef\u4ee5\u5b9e\u73b0\u5b9e\u65f6\u7684\u53cc\u5411\u901a\u4fe1\u3002<\/p>\n
from fastapi import FastAPI, WebSocket\n\napp = FastAPI()\n\n@app.websocket("\/ws")\nasync def websocket_endpoint(websocket: WebSocket):\n    await websocket.accept()\n    try:\n        while True:\n            data = await websocket.receive_text()\n            await websocket.send_text(f"\u6536\u5230\u6d88\u606f: {data}")\n    except Exception as e:\n        print(f"WebSocket \u9519\u8bef: {e}")\n    finally:\n        await websocket.close()\n\ndef demonstrate_websocket():\n    print("WebSocket \u6f14\u793a:")\n    print("1. \u5efa\u7acb\u8fde\u63a5")\n    print("2. \u63a5\u6536\u6d88\u606f")\n    print("3. \u53d1\u9001\u54cd\u5e94")\n    print("4. \u4fdd\u6301\u8fde\u63a5")\n\ndemonstrate_websocket()<\/code><\/pre>\n
\u6027\u80fd\u4f18\u5316\u7b56\u7565<\/h3>\n
FastAPI \u63d0\u4f9b\u4e86\u591a\u79cd\u6027\u80fd\u4f18\u5316\u7b56\u7565\uff0c\u53ef\u4ee5\u8fdb\u4e00\u6b65\u63d0\u5347\u5e94\u7528\u6027\u80fd\u3002<\/p>\n
import asyncio\nfrom functools import lru_cache\n\nclass CacheManager:\n    def __init__(self):\n        self.cache = {}\n    \n    async def get(self, key: str):\n        return self.cache.get(key)\n    \n    async def set(self, key: str, value, ttl: int = 60):\n        self.cache[key] = value\n    \n    @lru_cache(maxsize=100)\n    def cached_computation(self, n: int):\n        return sum(i * i for i in range(n))\n\nasync def demonstrate_optimization():\n    print("\u6027\u80fd\u4f18\u5316\u6f14\u793a:")\n    \n    cache = CacheManager()\n    \n    await cache.set("key1", "value1")\n    value = await cache.get("key1")\n    print(f"\u7f13\u5b58\u503c: {value}")\n    \n    result = cache.cached_computation(1000)\n    print(f"\u8ba1\u7b97\u7ed3\u679c: {result}")\n    \n    result2 = cache.cached_computation(1000)\n    print(f"\u7f13\u5b58\u7ed3\u679c: {result2}")\n\nasyncio.run(demonstrate_optimization())<\/code><\/pre>\n
\u5e76\u53d1\u8bf7\u6c42\u5904\u7406<\/h3>\n
FastAPI \u53ef\u4ee5\u9ad8\u6548\u5904\u7406\u5927\u91cf\u5e76\u53d1\u8bf7\u6c42\uff0c\u8fd9\u662f\u5176\u9ad8\u6027\u80fd\u7684\u6838\u5fc3\u4f53\u73b0\u3002<\/p>\n
import asyncio\nimport time\n\nasync def handle_request(request_id: int):\n    await asyncio.sleep(0.1)\n    return f"\u54cd\u5e94 {request_id}"\n\nasync def demonstrate_concurrency():\n    print("\u5e76\u53d1\u5904\u7406\u6f14\u793a:")\n    \n    start = time.time()\n    \n    tasks = [handle_request(i) for i in range(100)]\n    results = await asyncio.gather(*tasks)\n    \n    elapsed = time.time() - start\n    \n    print(f"\u5904\u7406 100 \u4e2a\u8bf7\u6c42\u8017\u65f6: {elapsed:.2f}\u79d2")\n    print(f"\u5e73\u5747\u6bcf\u4e2a\u8bf7\u6c42: {elapsed\/100:.4f}\u79d2")\n    print(f"\u5e76\u53d1\u6027\u80fd: {100\/elapsed:.1f} \u8bf7\u6c42\/\u79d2")\n\nasyncio.run(demonstrate_concurrency())<\/code><\/pre>\n
\u9519\u8bef\u5904\u7406\u548c\u65e5\u5fd7<\/h3>\n
\u5b8c\u5584\u7684\u9519\u8bef\u5904\u7406\u548c\u65e5\u5fd7\u8bb0\u5f55\u662f\u6784\u5efa\u5065\u58ee Web \u5e94\u7528\u7684\u5173\u952e\u3002<\/p>\n
from fastapi import FastAPI, HTTPException, Request\nfrom fastapi.responses import JSONResponse\nimport logging\n\napp = FastAPI()\n\nlogging.basicConfig(level=logging.INFO)\nlogger = logging.getLogger(__name__)\n\n@app.exception_handler(Exception)\nasync def global_exception_handler(request: Request, exc: Exception):\n    logger.error(f"\u672a\u5904\u7406\u7684\u5f02\u5e38: {exc}")\n    return JSONResponse(\n        status_code=500,\n        content={"detail": "\u5185\u90e8\u670d\u52a1\u5668\u9519\u8bef"}\n    )\n\n@app.get("\/items\/{item_id}")\nasync def read_item(item_id: int):\n    if item_id < 0:\n        raise HTTPException(status_code=400, detail="\u65e0\u6548\u7684\u5546\u54c1 ID")\n    \n    if item_id > 1000:\n        raise ValueError("\u5546\u54c1\u4e0d\u5b58\u5728")\n    \n    return {"item_id": item_id, "name": f"\u5546\u54c1 {item_id}"}\n\ndef demonstrate_error_handling():\n    print("\u9519\u8bef\u5904\u7406\u6f14\u793a:")\n    print("1. \u5168\u5c40\u5f02\u5e38\u5904\u7406\u5668")\n    print("2. HTTP \u5f02\u5e38")\n    print("3. \u81ea\u5b9a\u4e49\u5f02\u5e38")\n    print("4. \u65e5\u5fd7\u8bb0\u5f55")\n\ndemonstrate_error_handling()<\/code><\/pre>\n
\u603b\u7ed3<\/h3>\n
FastAPI \u7684\u9ad8\u6027\u80fd\u6e90\u4e8e\u5176\u57fa\u4e8e ASGI \u7684\u5f02\u6b65\u67b6\u6784\u3001\u9ad8\u6548\u7684\u4f9d\u8d56\u6ce8\u5165\u7cfb\u7edf\u3001\u5f3a\u5927\u7684\u8bf7\u6c42\u9a8c\u8bc1\u673a\u5236\u4ee5\u53ca\u7075\u6d3b\u7684\u4e2d\u95f4\u4ef6\u652f\u6301\u3002\u901a\u8fc7\u6df1\u5165\u7406\u89e3\u8fd9\u4e9b\u5e95\u5c42\u539f\u7406\uff0c\u5f00\u53d1\u8005\u53ef\u4ee5\u66f4\u597d\u5730\u5229\u7528 FastAPI \u6784\u5efa\u9ad8\u6027\u80fd\u7684 Web \u5e94\u7528\u3002<\/p>\n
\u638c\u63e1 FastAPI \u7684\u5f02\u6b65\u5904\u7406\u3001\u6570\u636e\u5e93\u64cd\u4f5c\u3001WebSocket \u652f\u6301\u3001\u6027\u80fd\u4f18\u5316\u7b49\u6838\u5fc3\u6280\u672f\uff0c\u5bf9\u4e8e\u6784\u5efa\u73b0\u4ee3\u3001\u9ad8\u6548\u7684 Python Web \u5e94\u7528\u81f3\u5173\u91cd\u8981\u3002FastAPI \u4e0d\u4ec5\u662f\u5f00\u53d1\u5de5\u5177\uff0c\u66f4\u662f\u73b0\u4ee3 Python Web \u5f00\u53d1\u7684\u6700\u4f73\u5b9e\u8df5\u4f53\u73b0\u3002<\/p>\n
https:\/\/segmentfault.com\/a\/1190000047614278<\/a><\/p>\n","protected":false},"excerpt":{"rendered":"
\u5f02\u6b65 Web \u6846\u67b6\u5e95\u5c42\uff1aFastAPI \u9ad8\u6027\u80fd\u539f\u7406\u5256\u6790 FastAPI \u662f\u73b0\u4ee3 Python Web \u5f00\u53d1\u7684\u70ed […]<\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"_jetpack_memberships_contains_paid_content":false,"footnotes":""},"categories":[4],"tags":[],"class_list":["post-10021259","post","type-post","status-publish","format-standard","hentry","category-share"],"jetpack_sharing_enabled":true,"jetpack_featured_media_url":"","_links":{"self":[{"href":"https:\/\/itmax.cn\/index.php?rest_route=\/wp\/v2\/posts\/10021259","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/itmax.cn\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/itmax.cn\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/itmax.cn\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/itmax.cn\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=10021259"}],"version-history":[{"count":0,"href":"https:\/\/itmax.cn\/index.php?rest_route=\/wp\/v2\/posts\/10021259\/revisions"}],"wp:attachment":[{"href":"https:\/\/itmax.cn\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=10021259"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/itmax.cn\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=10021259"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/itmax.cn\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=10021259"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}