当前位置:首页>python>Python Web开发:FastAPI基础

Python Web开发:FastAPI基础

  • 2026-10-11 07:31:30
Python Web开发:FastAPI基础
副标题

: 90%的人不知道,FastAPI的性能比Flask快300倍

痛点:为什么你的Web API开发总是很慢?

2025年某团队用Flask开发API,响应时间平均500ms。问题出在哪?工程师没有使用现代异步框架。

真相

:FastAPI基于Starlette和Pydantic,性能接近NodeJS和Go。

框架性能学习曲线推荐度
Flask1x低⭐⭐⭐
Django0.5x高⭐⭐⭐
FastAPI3-5x中⭐⭐⭐⭐⭐
Sanic4x中⭐⭐⭐⭐

一、FastAPI基础

1.1 第一个应用

from fastapi import FastAPI

app = FastAPI(

title="我的API",

description="一个FastAPI示例项目",

version="1.0.0"

)

@app.get("/")

async def root():

return {"message": "Hello, World!"}

@app.get("/items/{item_id}")

async def read_item(item_id: int):

return {"item_id": item_id}

# 运行

uvicorn main:app --reload

访问自动文档

http://localhost:8000/docs # Swagger UI

http://localhost:8000/redoc # ReDoc

1.2 路径操作

from fastapi import FastAPI

app = FastAPI()

GET

@app.get("/items/{item_id}")

async def get_item(item_id: int):

return {"item_id": item_id}

POST

@app.post("/items/")

async def create_item(name: str, price: float):

return {"name": name, "price": price}

PUT

@app.put("/items/{item_id}")

async def update_item(item_id: int, name: str):

return {"item_id": item_id, "name": name}

DELETE

@app.delete("/items/{item_id}")

async def delete_item(item_id: int):

return {"item_id": item_id, "message": "Deleted"}

带查询参数

@app.get("/search")

async def search(q: str, limit: int = 10, skip: int = 0):

return {"q": q, "limit": limit, "skip": skip}

1.3 请求体

from fastapi import FastAPI

from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):

name: str

price: float

description: str | None = None

tags: list[str] = []

@app.post("/items/")

async def create_item(item: Item):

return item

@app.put("/items/{item_id}")

async def update_item(item_id: int, item: Item):

return {"item_id": item_id, **item.model_dump()}

二、路径参数

2.1 基础路径参数

from fastapi import FastAPI

app = FastAPI()

@app.get("/users/{user_id}")

async def read_user(user_id: str):

return {"user_id": user_id}

@app.get("/items/{item_id}")

async def read_item(item_id: int):

return {"item_id": item_id}

2.2 路径参数验证

from fastapi import FastAPI, Path

from typing import Annotated

app = FastAPI()

@app.get("/items/{item_id}")

async def read_item(

item_id: Annotated[int, Path(title="物品ID", ge=0, le=1000)]

):

return {"item_id": item_id}

@app.get("/users/{user_id}")

async def read_user(

user_id: Annotated[str, Path(min_length=3, max_length=50, pattern=r"^[a-zA-Z0-9_-]+$")]

):

return {"user_id": user_id}

2.3 多个路径参数

@app.get("/users/{user_id}/items/{item_id}")

async def read_user_item(

user_id: str,

item_id: int,

q: str | None = None,

short: bool = False

):

item = {"item_id": item_id, "owner_id": user_id}

if q:

item.update({"q": q})

if short:

item = {"item_id": item_id, "owner_id": user_id}

return item

三、查询参数

3.1 基础查询参数

from fastapi import FastAPI

app = FastAPI()

@app.get("/items/")

async def read_items(

q: str | None = None,

skip: int = 0,

limit: int = 100

):

return {"q": q, "skip": skip, "limit": limit}

3.2 查询参数验证

from fastapi import FastAPI, Query

from typing import Annotated

app = FastAPI()

@app.get("/items/")

async def read_items(

q: Annotated[str | None, Query(min_length=3, max_length=50)] = None,

skip: Annotated[int, Query(ge=0)] = 0,

limit: Annotated[int, Query(ge=1, le=100)] = 100

):

return {"q": q, "skip": skip, "limit": limit}

3.3 列表查询参数

@app.get("/items/")

async def read_items(

q: list[str] | None = None,

skip: int = 0,

limit: int = 100

):

return {"q": q, "skip": skip, "limit": limit}

访问: /items/?q=foo&q=bar&q=baz

四、请求体与字段验证

4.1 Pydantic模型

from fastapi import FastAPI

from pydantic import BaseModel, Field, field_validator

from typing import Annotated

app = FastAPI()

class Item(BaseModel):

name: Annotated[str, Field(min_length=1, max_length=50)]

price: Annotated[float, Field(gt=0)]

description: str | None = Field(None, max_length=500)

tags: list[str] = Field(default_factory=list)

in_stock: bool = True

@field_validator('name')

@classmethod

def validate_name(cls, v: str) -> str:

if not v.strip():

raise ValueError('名称不能为空')

return v.strip()

@app.post("/items/")

async def create_item(item: Item):

return item

4.2 嵌套模型

class Address(BaseModel):

street: str

city: str

zip_code: str

class User(BaseModel):

username: str

email: str

address: Address | None = None

@app.post("/users/")

async def create_user(user: User):

return user

4.3 多个请求体

from fastapi import FastAPI, Body

from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):

name: str

price: float

@app.put("/items/{item_id}")

async def update_item(

item_id: int,

item: Item,

q: str | None = None,

importance: int = Body(1, ge=1, le=5),

include_in_memory: bool = Body(False)

):

return {

"item_id": item_id,

"item": item,

"q": q,

"importance": importance,

"include_in_memory": include_in_memory

}

五、响应模型

5.1 响应模型过滤

from fastapi import FastAPI

from pydantic import BaseModel, SecretStr

app = FastAPI()

class UserIn(BaseModel):

username: str

password: SecretStr

email: str

full_name: str | None = None

class UserOut(BaseModel):

username: str

email: str

full_name: str | None = None

@app.post("/users/", response_model=UserOut)

async def create_user(user: UserIn):

# password不会被返回

return user

5.2 响应状态码

from fastapi import FastAPI, status

app = FastAPI()

@app.post("/items/", status_code=status.HTTP_201_CREATED)

async def create_item(item: dict):

return item

@app.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)

async def delete_item(item_id: int):

return None

5.3 自定义响应

from fastapi import FastAPI

from fastapi.responses import JSONResponse, RedirectResponse

app = FastAPI()

@app.get("/redirect")

async def redirect_example():

return RedirectResponse(url="/items/")

@app.get("/custom")

async def custom_response():

return JSONResponse(

content={"message": "Custom response"},

status_code=200,

headers={"X-Custom-Header": "value"}

)

六、依赖注入

6.1 基础依赖

from fastapi import FastAPI, Depends

app = FastAPI()

async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):

return {"q": q, "skip": skip, "limit": limit}

@app.get("/items/")

async def read_items(commons: dict = Depends(common_parameters)):

return commons

@app.get("/users/")

async def read_users(commons: dict = Depends(common_parameters)):

return commons

6.2 类依赖

from fastapi import FastAPI, Depends, HTTPException

app = FastAPI()

class CommonQueryParams:

def __init__(self, q: str | None = None, skip: int = 0, limit: int = 100):

self.q = q

self.skip = skip

self.limit = limit

@app.get("/items/")

async def read_items(commons: CommonQueryParams = Depends()):

return {

"q": commons.q,

"skip": commons.skip,

"limit": commons.limit

}

6.3 认证依赖

from fastapi import FastAPI, Depends, HTTPException, status

from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

app = FastAPI()

security = HTTPBearer()

async def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)):

token = credentials.credentials

# 验证token

if token != "valid-token":

raise HTTPException(

status_code=status.HTTP_401_UNAUTHORIZED,

detail="Invalid token"

)

return {"user_id": "user123", "token": token}

@app.get("/users/me")

async def read_users_me(current_user: dict = Depends(get_current_user)):

return current_user

6.4 依赖树

async def query_dependency(q: str | None = None):

return {"q": q}

async def pagination_dependency(skip: int = 0, limit: int = 100):

return {"skip": skip, "limit": limit}

async def common_dependency(

q_data: dict = Depends(query_dependency),

page_data: dict = Depends(pagination_dependency)

):

return {q_data, page_data}

@app.get("/items/")

async def read_items(commons: dict = Depends(common_dependency)):

return commons

七、错误处理

7.1 HTTPException

from fastapi import FastAPI, HTTPException, status

app = FastAPI()

items = {"foo": "bar"}

@app.get("/items/{item_id}")

async def read_item(item_id: str):

if item_id not in items:

raise HTTPException(

status_code=status.HTTP_404_NOT_FOUND,

detail=f"Item {item_id} not found",

headers={"X-Error": "Item not found"}

)

return {"item": items[item_id]}

7.2 自定义异常处理器

from fastapi import FastAPI, Request

from fastapi.responses import JSONResponse

app = FastAPI()

class ItemNotFoundError(Exception):

def __init__(self, item_id: str):

self.item_id = item_id

@app.exception_handler(ItemNotFoundError)

async def item_not_found_handler(request: Request, exc: ItemNotFoundError):

return JSONResponse(

status_code=404,

content={"message": f"Item {exc.item_id} not found"}

)

@app.get("/items/{item_id}")

async def read_item(item_id: str):

if item_id not in items:

raise ItemNotFoundError(item_id)

return {"item": items[item_id]}

7.3 全局异常处理

@app.exception_handler(Exception)

async def global_exception_handler(request: Request, exc: Exception):

return JSONResponse(

status_code=500,

content={

"message": "Internal server error",

"detail": str(exc) if app.debug else None

}

)

八、中间件

8.1 基础中间件

from fastapi import FastAPI, Request

import time

app = FastAPI()

@app.middleware("http")

async def add_process_time_header(request: Request, call_next):

start_time = time.time()

response = await call_next(request)

process_time = time.time() - start_time

response.headers["X-Process-Time"] = str(process_time)

return response

8.2 CORS中间件

from fastapi import FastAPI

from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

app.add_middleware(

CORSMiddleware,

allow_origins=["*"],

allow_credentials=True,

allow_methods=["*"],

allow_headers=["*"],

)

8.3 日志中间件

import logging

from fastapi import FastAPI, Request

logger = logging.getLogger("uvicorn")

@app.middleware("http")

async def log_requests(request: Request, call_next):

start_time = time.time()

response = await call_next(request)

process_time = time.time() - start_time

logger.info(

f"{request.method} {request.url.path} - {response.status_code} - {process_time:.3f}s"

)

return response

九、实战案例

9.1 完整API项目结构

my_api/

├── app/

│ ├── __init__.py

│ ├── main.py

│ ├── config.py

│ ├── database.py

│ ├── models.py

│ ├── schemas.py

│ ├── api/

│ │ ├── __init__.py

│ │ ├── v1/

│ │ │ ├── __init__.py

│ │ │ ├── items.py

│ │ │ └── users.py

│ │ └── deps.py

│ └── core/

│ ├── __init__.py

│ ├── security.py

│ └── exceptions.py

├── tests/

│ ├── __init__.py

│ ├── test_items.py

│ └── test_users.py

├── .env

├── pyproject.toml

└── README.md

9.2 用户管理系统

# app/schemas.py

from pydantic import BaseModel, EmailStr, Field

from datetime import datetime

from typing import Annotated

class UserBase(BaseModel):

email: EmailStr

username: Annotated[str, Field(min_length=3, max_length=50)]

class UserCreate(UserBase):

password: Annotated[str, Field(min_length=8)]

class User(UserBase):

id: int

is_active: bool = True

created_at: datetime

class ItemBase(BaseModel):

name: str

description: str | None = None

price: float

class ItemCreate(ItemBase):

pass

class Item(ItemBase):

id: int

owner_id: int

created_at: datetime

# app/api/v1/items.py

from fastapi import APIRouter, Depends, HTTPException, status

from typing import Annotated

from app.schemas import Item, ItemCreate

from app.api import deps

router = APIRouter()

@router.post("/", response_model=Item, status_code=status.HTTP_201_CREATED)

async def create_item(

item_in: ItemCreate,

current_user: dict = Depends(deps.get_current_user)

):

return item_in

@router.get("/", response_model=list[Item])

async def read_items(

skip: int = 0,

limit: int = 100,

current_user: dict = Depends(deps.get_current_user)

):

return []

@router.get("/{item_id}", response_model=Item)

async def read_item(

item_id: int,

current_user: dict = Depends(deps.get_current_user)

):

return {"id": item_id, "name": "test"}

@router.delete("/{item_id}", status_code=status.HTTP_204_NO_CONTENT)

async def delete_item(

item_id: int,

current_user: dict = Depends(deps.get_current_user)

):

return None

# app/main.py

from fastapi import FastAPI

from app.api.v1 import items, users

from app.core.config import settings

app = FastAPI(

title=settings.APP_NAME,

version="1.0.0"

)

app.include_router(items.router, prefix="/api/v1/items", tags=["items"])

app.include_router(users.router, prefix="/api/v1/users", tags=["users"])

@app.get("/health")

async def health_check():

return {"status": "healthy"}

十、测试

10.1 单元测试

from fastapi.testclient import TestClient

from app.main import app

client = TestClient(app)

def test_read_main():

response = client.get("/")

assert response.status_code == 200

assert response.json() == {"message": "Hello, World!"}

def test_create_item():

response = client.post(

"/items/",

json={"name": "foo", "price": 10.5}

)

assert response.status_code == 201

data = response.json()

assert data["name"] == "foo"

assert data["price"] == 10.5

def test_read_item_not_found():

response = client.get("/items/999")

assert response.status_code == 404

10.2 依赖覆盖测试

from fastapi import Depends

from app.main import app

from app.api import deps

def override_get_current_user():

return {"user_id": "test_user", "is_admin": True}

app.dependency_overrides[deps.get_current_user] = override_get_current_user

def test_admin_endpoint():

response = client.get("/admin/users")

assert response.status_code == 200

app.dependency_overrides.clear()

常见坑自查清单

坑现象自查方法修复方案
类型错误422错误检查Pydantic模型添加类型注解
依赖未注入None值检查Depends使用正确使用Depends
异步阻塞性能差检查async/await使用async函数
循环依赖导入错误检查模块结构重构代码结构

结语

关键洞察

:

  • ●FastAPI性能优秀,基于异步
  • ●Pydantic提供强大的数据验证
  • ●依赖注入系统灵活强大
  • ●自动生成API文档

互动

  1. 1.你用FastAPI开发过什么项目?
  2. 2.依赖注入好用吗?
  3. 3.遇到过性能问题吗?
版本: V1.0 | 2026-05-26 | Python Web开发系列

📚 推荐阅读

📝 摘要:今天深入学习静态代码分析技术,这是安全审计的核心技能。从 Python AST 模块到检测模式设计,收获满满!

发布于 202603

01-Python 环境搭建与第一个脚本

发布于 202603

【优化】Python代码优化与调试技巧

发布于 202603

KEYWORDS

IL, Python, AI, 函数, 循环

💡 如果你觉得这篇文章有帮助,请点个在看,分享给更多需要的人!

📝 关注我,获取更多实用干货~

🤝 有问题欢迎评论区留言交流!

最新文章

随机文章