boxmoe_header_banner_img

Hello! 欢迎来到悠悠畅享网!

文章导读

Python如何构建REST API?FastAPI框架快速入门


avatar
站长 2025年8月15日 3

fastapi是构建高性能rest api的首选python框架,1.它基于类型提示和依赖注入实现代码清晰与自动文档生成;2.通过pydantic模型验证请求体数据;3.利用依赖注入系统复用公共逻辑;4.支持api key、oauth2等身份验证机制;5.可集成sqlalchemy等orm进行数据库操作;6.使用testclient配合pytest完成单元测试;7.可通过docker容器化并部署到云平台。该框架兼具高性能与开发效率,适用于现代api开发全流程,从定义路由到部署均提供完整解决方案。

Python如何构建REST API?FastAPI框架快速入门

构建REST API,Python提供了多种选择,但FastAPI无疑是近年来最受欢迎的框架之一。它以其高性能、易用性和自动化的文档生成能力脱颖而出。

解决方案

FastAPI的核心在于类型提示和依赖注入,这使得代码更加清晰、易于维护,并且能够自动生成OpenAPI和Swagger文档。以下是一个快速入门的示例:

立即学习Python免费学习笔记(深入)”;

  1. 安装FastAPI和Uvicorn:

    pip install fastapi uvicorn

    Uvicorn是一个ASGI服务器,用于运行FastAPI应用。

  2. 创建

    main.py

    文件:

    from fastapi import FastAPI  app = FastAPI()  @app.get("/") async def read_root():     return {"Hello": "World"}  @app.get("/items/{item_id}") async def read_item(item_id: int, q: str = None):     return {"item_id": item_id, "q": q}
    • FastAPI()

      创建一个FastAPI应用实例。

    • @app.get("/")

      定义一个GET请求的路由,路径为根目录

      /

    • async def read_root()

      定义一个异步函数,处理根目录的请求,返回一个JSON响应。

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

      定义一个GET请求的路由,路径为

      /items/{item_id}

      ,其中

      {item_id}

      是一个路径参数。

    • item_id: int

      使用类型提示,将

      item_id

      声明为整数类型。FastAPI会自动进行数据验证。

    • q: str = None

      定义一个查询参数

      q

      ,类型为字符串,默认值为

      None

  3. 运行应用:

    uvicorn main:app --reload
    • main:app

      指定

      main.py

      文件中的

      app

      对象作为FastAPI应用。

    • --reload

      启用自动重载,当代码发生更改时,服务器会自动重启

  4. 访问API:

    • 浏览器中访问
      http://127.0.0.1:8000/

      ,你将看到

      {"Hello": "World"}

    • 访问
      http://127.0.0.1:8000/items/123?q=test

      ,你将看到

      {"item_id": 123, "q": "test"}

  5. 查看自动生成的文档:

    • 访问
      http://127.0.0.1:8000/docs

      ,你将看到Swagger UI,它会根据你的代码自动生成API文档。

    • 访问
      http://127.0.0.1:8000/redoc

      ,你将看到ReDoc文档。

如何处理请求体?

FastAPI使用Pydantic模型来定义请求体。Pydantic是一个数据验证和设置管理库,它可以将Python类型转换为JSON模式,并自动验证请求数据。

from fastapi import FastAPI from pydantic import BaseModel  app = FastAPI()  class Item(BaseModel):     name: str     description: str = None     price: float     tax: float = None  @app.post("/items/") async def create_item(item: Item):     item_dict = item.dict()     if item.tax:         price_with_tax = item.price + item.tax         item_dict.update({"price_with_tax": price_with_tax})     return item_dict
  • class Item(BaseModel):

    定义一个Pydantic模型,用于描述请求体的数据结构。

  • name: str

    ,

    description: str = None

    ,

    price: float

    ,

    tax: float = None

    定义模型的字段,并使用类型提示。

  • @app.post("/items/")

    定义一个POST请求的路由,路径为

    /items/

  • async def create_item(item: Item):

    接收一个

    Item

    类型的参数,FastAPI会自动将请求体的数据转换为

    Item

    对象。

  • item.dict()

    Item

    对象转换为字典。

FastAPI的依赖注入如何工作?

FastAPI的依赖注入系统允许你将依赖项声明为函数参数。FastAPI会自动解析这些依赖项,并将它们传递给你的函数。这使得代码更加模块化、可测试和可重用。

from fastapi import FastAPI, Depends  app = FastAPI()  async def common_parameters(q: str = 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
  • async def common_parameters(q: str = None, skip: int = 0, limit: int = 100):

    定义一个依赖项函数,它接收查询参数

    q

    skip

    limit

  • commons: dict = Depends(common_parameters)

    声明

    commons

    参数的依赖项为

    common_parameters

    函数。FastAPI会自动调用

    common_parameters

    函数,并将返回值传递给

    read_items

    函数。

  • read_items

    read_users

    函数都使用了相同的依赖项

    common_parameters

    ,这避免了代码重复。

如何进行身份验证?

身份验证是REST API开发中的一个重要方面。FastAPI提供了多种身份验证方案,例如基于OAuth2的身份验证、基于JWT的身份验证等。

一个简单的API Key示例:

from fastapi import FastAPI, Depends, HTTPException, status from fastapi.security import APIKeyHeader  app = FastAPI()  API_KEY = "your_secret_api_key" API_KEY_NAME = "X-API-Key"  api_key_header = APIKeyHeader(name=API_KEY_NAME, auto_error=False)  async def get_api_key(api_key_header: str = Depends(api_key_header)):     if api_key_header == API_KEY:         return api_key_header     else:         raise HTTPException(             status_code=status.HTTP_401_UNAUTHORIZED,             detail="Invalid API Key",         )  @app.get("/items/", dependencies=[Depends(get_api_key)]) async def read_items():     return [{"item": "Foo"}, {"item": "Bar"}]
  • APIKeyHeader

    定义了一个API Key头。

  • get_api_key

    函数验证API Key是否正确。

  • dependencies=[Depends(get_api_key)]

    get_api_key

    函数作为

    read_items

    函数的依赖项。只有当API Key验证成功时,才能访问

    read_items

    函数。

如何处理数据库操作?

FastAPI可以与各种数据库集成,例如PostgreSQL、MySQL、MongoDB等。通常,可以使用ORM(对象关系映射)库来简化数据库操作,例如SQLAlchemy、Tortoise ORM等。

例如,使用SQLAlchemy:

from fastapi import FastAPI, Depends from sqlalchemy import create_engine, Column, Integer, String from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker, Session  SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"  # 使用SQLite,方便演示  engine = create_engine(     SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}  # 生产环境不推荐 ) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)  Base = declarative_base()  class Item(Base):     __tablename__ = "items"      id = Column(Integer, primary_key=True, index=True)     name = Column(String, index=True)     description = Column(String, nullable=True)     price = Column(Integer)  Base.metadata.create_all(bind=engine)  app = FastAPI()  def get_db():     db = SessionLocal()     try:         yield db     finally:         db.close()  @app.post("/items/") async def create_item(name: str, description: str, price: int, db: Session = Depends(get_db)):     db_item = Item(name=name, description=description, price=price)     db.add(db_item)     db.commit()     db.refresh(db_item)     return db_item
  • 定义数据库连接和模型。
  • 使用依赖注入获取数据库会话。
  • 进行数据库操作。

如何进行单元测试?

FastAPI的测试非常简单,可以使用

pytest

httpx

库进行单元测试。

from fastapi.testclient import TestClient from .main import app  # 假设你的FastAPI应用在main.py文件中  client = TestClient(app)  def test_read_main():     response = client.get("/")     assert response.status_code == 200     assert response.json() == {"Hello": "World"}  def test_create_item():     response = client.post(         "/items/",         json={"name": "Test Item", "description": "A test item", "price": 10}     )     assert response.status_code == 200     assert response.json()["name"] == "Test Item"
  • TestClient

    是一个用于测试FastAPI应用的客户端。

  • 使用
    client.get()

    client.post()

    发送HTTP请求。

  • 使用
    assert

    语句验证响应状态码和内容。

如何部署FastAPI应用?

FastAPI应用可以部署到各种云平台和服务器上,例如Heroku、AWS、Google Cloud Platform等。通常,可以使用Docker容器化应用,然后部署到容器编排平台,例如Kubernetes。

一个简单的Dockerfile:

FROM python:3.9  WORKDIR /app  COPY requirements.txt ./ RUN pip install --no-cache-dir -r requirements.txt  COPY . .  CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
  • 指定基础镜像。
  • 设置工作目录。
  • 复制依赖项文件并安装。
  • 复制应用代码。
  • 运行Uvicorn服务器。



评论(已关闭)

评论已关闭