掘金 人工智能 22小时前
AI 大模型应用进阶系列(五):FastAPI 入门
index_new5.html
../../../zaker_core/zaker_tpl_static/wap/tpl_guoji1.html

 

FastAPI是一款基于Python类型提示构建API的现代、高性能Web框架。它以其极高的性能、快速的开发效率和更少的错误率而著称,能够与NodeJS和Go相媲美。FastAPI的核心组件包括Uvicorn(高性能ASGI服务器)、Starlette(轻量级ASGI框架)和Pydantic(数据验证库)。它深度集成了这些组件,提供了自动文档生成、依赖注入、请求体处理、响应模型、文件上传、异常处理、中间件和安全性等丰富功能,极大地简化了API开发流程,并支持与数据库(如MySQL)的集成,使其成为构建健壮API的理想选择。

🚀 **高性能与快速开发**:FastAPI基于Starlette和Pydantic,提供与NodeJS和Go相当的性能,并将开发速度提升2-3倍,同时减少约40%的开发者错误,显著提升开发效率和代码质量。

⚙️ **核心技术栈**:FastAPI的底层依赖于Uvicorn(高性能ASGI服务器)提供运行环境,Starlette(轻量级ASGI框架)提供路由、请求处理等基础功能,而Pydantic则负责数据验证和类型提示,确保API的数据准确性和健壮性。

📚 **丰富的功能支持**:FastAPI内置了自动生成的交互式API文档(Swagger UI和ReDoc)、请求体处理、响应模型、文件上传、异常处理、中间件和依赖注入等功能,极大地简化了API的构建和管理。

🔗 **数据库集成**:文章展示了如何利用FastAPI结合SQLAlchemy和Pydantic实现与MySQL数据库的交互,包括用户信息的CRUD(创建、读取、更新、删除)操作,为构建数据驱动的API提供了清晰的示例。

🛡️ **安全性与标准化**:FastAPI支持OAuth2PasswordBearer等安全机制,确保API的安全性,并遵循API开放标准,保证了API的兼容性和互操作性。

FastAPI 是一个用于构建 API 的现代、快速(高性能)的 web 框架,使用 Python 并基于标准的 Python 类型提示。

官网: fastapi.tiangolo.com/

官网中文:fastapi.tiangolo.com/zh/

特性

FastAPI 的核心

Uvicorn

官方文档:www.uvicorn.org/

Starlette

官方文档:www.starlette.io/

Starlette 是一个轻量级的 ASGI(Asynchronous Server Gateway Interface)框架,专为构建异步 Web 应用设计。它是 FastAPI 的底层框架,提供了路由、请求处理、中间件、WebSocket 支持等核心功能。Starlette 的设计目标是简单、高效,同时保持足够的灵活性,适用于各种 Web 开发场景。拥有以下特性:

FastAPI 直接继承了 Starlette 的所有功能,例如路由、请求处理和中间件。FastAPI 的 @app.get()、@app.post() 等装饰器实际上是对 Starlette 路由系统的封装。换句话说,FastAPI 在 Starlette 的基础上增加了类型检查、自动文档生成等高级特性

Pydantic

Pydantic 是一个基于 Python 类型提示(type hints)的数据验证和解析库,它能自动校验数据是否符合预定义的类型和约束,广泛用于 API 开发、配置管理、数据序列化等场景。其核心优势是结合了 Python 类型系统和运行时数据验证,既提升了代码可读性,又确保了数据合法性

FastAPI 与 Uvicorn、 Starlette、Pydantic的关系

FastAPI、Starlette、Uvicorn 和 Pydantic 是 Python 异步 Web 开发生态中紧密协作的四个核心组件,它们分工明确又相互配合,共同构成了高效的 API 开发体系。它们的关系可以概括为:

具体关系解析

    FastAPI 与 Uvicorn

      Uvicorn 是一个基于 asyncio 的 ASGI 服务器(异步服务器网关接口),负责处理底层网络通信FastAPI 和 Starlette 都是 ASGI 应用框架,必须依赖 Uvicorn(或其他 ASGI 服务器,如 Hypercorn)才能运行关系类比:Uvicorn 像 "服务器主机",FastAPI/Starlette 像运行在主机上的 "应用程序"

    FastAPI 与 Starlette

      FastAPI 直接继承自 Starlette,是 Starlette 的增强版Starlette 提供基础 Web 功能(路由、中间件、WebSocket、异步支持等)FastAPI 在 Starlette 基础上增加了 API 专用功能(自动文档、依赖注入等)关系类比:Starlette 是 "基础模板",FastAPI 是基于模板定制的 "API 专用版本"

    FastAPI 与 Pydantic

      Pydantic 是独立的数据验证库,基于类型提示实现数据校验FastAPI 深度集成 Pydantic,将其作为数据验证的核心工具所有请求参数、响应模型都通过 Pydantic 模型定义和验证关系类比:Pydantic 是 "数据安检仪",确保进入 FastAPI 的数据符合规范

基础示例

from fastapi import FastAPI # 创建 FastAPI 应用实例 app = FastAPI() # 定义根路径的 GET 请求接口@app.get("/") # 处理根路径请求的函数 # 返回一个字典,内容为{"Hello": "World"} def read_root():     return {"Hello": "World"}     # 定义带有路径参数的 GET 请求接口 @app.get("/items/{item_id}") # 处理带参数请求的函数,item_id 为路径参数,q 为可选查询参数 def read_item(item_id: int, q: str = None):     return {"item_id": item_id, "q": q}    # 定义带有查询参数的 GET 请求接口@app.get("/items/")# 处理带查询参数请求的函数,skip 和 limit 为查询参数def read_items(skip: int = 0, limit: int = 10):    return {"skip": skip, "limit": limit}    # 仅在直接运行本文件时启动 uvicorn 服务器 if __name__ == "__main__":     # 导入 uvicorn 服务器     import uvicorn     # 启动应用,host 为 127.0.0.1,端口为 8000,自动重载    uvicorn.run(app, host="127.0.0.1", port=8000, reload=True)

运行以上 main.py 文件,启动服务访问: http://127.0.0.1:8000/docs ,可看到由Swagger UI生成的交互式API文档

请求体

# 导入 FastAPI 和 Pydantic 的 BaseModelfrom fastapi import FastAPIfrom pydantic import BaseModel# 定义请求体的数据模型class Item(BaseModel):    # 名称,字符串类型    name: str    # 描述,字符串类型,可选    description: str = None    # 价格,浮点数类型    price: float    # 税,浮点数类型,可选    tax: float = None# 创建 FastAPI 应用实例app = FastAPI()# 定义 POST 请求接口,接收 Item 类型的请求体@app.post("/items/")def create_item(item: Item):    return itemif __name__ == "__main__":    import uvicorn    uvicorn.run(app, host="127.0.0.1", port=8000)

响应模型

# 导入 FastAPI 和 Pydantic 的 BaseModelfrom fastapi import FastAPIfrom pydantic import BaseModel# 定义响应模型class Item(BaseModel):    name: str    description: str = None    price: float    tax: float = Noneapp = FastAPI()# 指定 response_model,返回的数据会自动校验和序列化@app.post("/items/", response_model=Item)def create_item(item: Item):    return itemif __name__ == "__main__":    import uvicorn    uvicorn.run(app, host="127.0.0.1", port=8000)

表单

# 导入 FastAPI 和 Formfrom fastapi import FastAPI, Formapp = FastAPI()# 定义 POST 接口,接收表单数据@app.post("/login/")def login(username: str = Form(...), password: str = Form(...)):    return {"username": username}if __name__ == "__main__":    import uvicorn    uvicorn.run(app, host="127.0.0.1", port=8000)

文件上传

# 导入 FastAPI、UploadFile 和 Filefrom fastapi import FastAPI, UploadFile, Fileapp = FastAPI()# 上传原始字节文件@app.post("/files/")def create_file(file: bytes = File(...)):    return {"file_size": len(file)}# 上传文件对象@app.post("/uploadfile/")def create_upload_file(file: UploadFile = File(...)):    return {"filename": file.filename}if __name__ == "__main__":    import uvicorn    uvicorn.run(app, host="127.0.0.1", port=8000)

异常处理

# 导入 FastAPI 和 HTTPExceptionfrom fastapi import FastAPI, HTTPExceptionapp = FastAPI()# 假设有一个 items 字典存储数据items = {"foo": "The Foo item", "bar": "The Bar item"}# 定义带有异常处理的接口@app.get("/items/{item_id}")def read_item(item_id: str):    # 如果 item_id 不在 items 中,抛出 404 异常    if item_id not in items:        raise HTTPException(status_code=404, detail="Item not found")    # 返回对应的 item    return {"item": items[item_id]}if __name__ == "__main__":    import uvicorn    uvicorn.run(app, host="127.0.0.1", port=8000)

中间件

# 导入 FastAPI 和 Requestfrom fastapi import FastAPI, Requestimport timeapp = 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# 示例接口@app.get("/")async def root():    return {"message": "Hello World"}if __name__ == "__main__":    import uvicorn    uvicorn.run(app, host="127.0.0.1", port=8000)

依赖注入

# 导入 FastAPI 和 Dependsfrom fastapi import FastAPI, Dependsapp = FastAPI()# 定义依赖函数,返回查询参数def common_parameters(q: str = None, skip: int = 0, limit: int = 100):    return {"q": q, "skip": skip, "limit": limit}# 使用 Depends 注入依赖@app.get("/items/")def read_items(commons: dict = Depends(common_parameters)):    return commonsif __name__ == "__main__":    import uvicorn    uvicorn.run(app, host="127.0.0.1", port=8000)

安全性

# 导入 FastAPI 和 OAuth2PasswordBearerfrom fastapi import FastAPI, Dependsfrom fastapi.security import OAuth2PasswordBearerapp = FastAPI()# 创建 OAuth2PasswordBearer 实例,指定 tokenUrloauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")# 依赖注入 token@app.get("/items/")def read_items(token: str = Depends(oauth2_scheme)):    return {"token": token}if __name__ == "__main__":    import uvicorn    uvicorn.run(app, host="127.0.0.1", port=8000)

数据库链接

from fastapi import FastAPI, APIRouter, HTTPException, Dependsfrom pydantic import BaseModelfrom typing import Listfrom sqlalchemy import create_engine, Column, String, Integerfrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy.orm import sessionmaker, Session# 初始化FastAPI应用app = FastAPI(title="用户管理API", description="一个简单的用户表CRUD示例")application = APIRouter()# 数据库配置 - MySQL# 请替换为你的MySQL数据库信息SQLALCHEMY_DATABASE_URL = (    "mysql+pymysql://username:password@localhost/fastapi?charset=utf8mb4")engine = create_engine(SQLALCHEMY_DATABASE_URL)SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)# 数据库模型基类Base = declarative_base()# 用户模型class User(Base):    __tablename__ = "users"    id = Column(Integer, primary_key=True, index=True)    user_name = Column(String(100), index=True)  # 为MySQL指定VARCHAR长度    user_age = Column(Integer)# 创建数据库表Base.metadata.create_all(bind=engine)# Pydantic模型 - 用于数据验证和序列化class UserBase(BaseModel):    user_name: str    user_age: intclass UserCreate(UserBase):    passclass UserUpdate(UserBase):    passclass UserInDB(UserBase):    id: int    class Config:        orm_mode = True# 依赖项 - 获取数据库会话(修正的部分)def get_db() -> Session:    db = SessionLocal()    try:        yield db    finally:        db.close()# CRUD操作实现# 创建用户 - 使用Depends(get_db)而非next(get_db)@application.post("/users/", response_model=UserInDB, summary="创建新用户")def create_user(user: UserCreate, db: Session = Depends(get_db)):    # 检查用户名是否已存在    db_user = db.query(User).filter(User.user_name == user.user_name).first()    if db_user:        raise HTTPException(status_code=400, detail="用户名已存在")    # 创建新用户    db_user = User(user_name=user.user_name, user_age=user.user_age)    db.add(db_user)    db.commit()    db.refresh(db_user)    return db_user# 获取所有用户@application.get("/users/", response_model=List[UserInDB], summary="获取所有用户")def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):    users = db.query(User).offset(skip).limit(limit).all()    return users# 根据ID获取用户@application.get("/users/{user_id}", response_model=UserInDB, summary="根据ID获取用户")def read_user(user_id: int, db: Session = Depends(get_db)):    db_user = db.query(User).filter(User.id == user_id).first()    if db_user is None:        raise HTTPException(status_code=404, detail="用户不存在")    return db_user# 更新用户@application.put("/users/{user_id}", response_model=UserInDB, summary="更新用户信息")def update_user(user_id: int, user: UserUpdate, db: Session = Depends(get_db)):    db_user = db.query(User).filter(User.id == user_id).first()    if db_user is None:        raise HTTPException(status_code=404, detail="用户不存在")    # 检查新用户名是否已被其他用户使用    if user.user_name != db_user.user_name:        existing_user = db.query(User).filter(User.user_name == user.user_name).first()        if existing_user:            raise HTTPException(status_code=400, detail="用户名已存在")    # 更新用户信息    db_user.user_name = user.user_name    db_user.user_age = user.user_age    db.commit()    db.refresh(db_user)    return db_user# 删除用户@application.delete("/users/{user_id}", summary="删除用户")def delete_user(user_id: int, db: Session = Depends(get_db)):    db_user = db.query(User).filter(User.id == user_id).first()    if db_user is None:        raise HTTPException(status_code=404, detail="用户不存在")    db.delete(db_user)    db.commit()    return {"message": f"用户 {user_id} 已成功删除"}# 添加路由app.include_router(application, prefix="/demo", tags=["用户操作"])if __name__ == "__main__":    import uvicorn    uvicorn.run(app, host="127.0.0.1", port=8000)

运行上述main.py文件,访问:http://127.0.0.1:8000/docs ,可以看到如下交互API页面,就可以对fastapi数据库的user表进行增删改查

创建用户

数据库

查询用户

更新用户信息

Fish AI Reader

Fish AI Reader

AI辅助创作,多种专业模板,深度分析,高质量内容生成。从观点提取到深度思考,FishAI为您提供全方位的创作支持。新版本引入自定义参数,让您的创作更加个性化和精准。

FishAI

FishAI

鱼阅,AI 时代的下一个智能信息助手,助你摆脱信息焦虑

联系邮箱 441953276@qq.com

相关标签

FastAPI Python Web框架 API开发 ASGI
相关文章