跳至主要內容

FastAPI笔记

棋大约 13 分钟约 3978 字

认识FastAPI

FastAPI是一个快速、高性能的Web框架，基于标准Python类星提示，使用Python3.6+构建API。FastAPI具有以下特点：

  • 高性能：与NodeJS和Go相当，拥有高性能，是现在速度最快的Python开发框架之一。
  • 快速编码：将功能开发速度提升2~3倍。
  • 更少的bug：减少约40%的人为编程错误。
  • 直观：更好的编程支持，可智能感知并补全多处代码（IntelliJSense），减少调试时间。
  • 简单：方便使用和学习，减少阅读文档的时间。
  • 简洁：最小化代码重复，每个参数都可以声明多个要素，更少的错误。
  • Robust：获取便于生产的代码，带自动交互式文档。
  • 基于标准：基于API开放标准OpenAPI和JSON Schema，并完全兼容。

FastAPI要求Python版本高于3.6，此外，它依赖于Starlette和Pydantic两个包，前者用于Web相关部分，后者用于数据库部分。

安装FastAPI非常简单，使用pip install fastapi即可。此外，FastAPI依赖于uvicorn，因此还需要安装Uvicorn。

第一个FastAPI程序

下面通过一个简单的例子输出Hello World。

from fastapi import FastAPI  # 导入FastAPI类

app = FastAPI()  # 实例化应用类

@app.get('/')  # 路由
def index():
    return {'Hello': 'World'}

@app.get('/items/{item_id}')  # 从URL中接收参数
def read_item(item_id: int, q: str = None):
    return {'item_id': item_id, 'q': q}

接下来，使用Uvicorn启动服务，命令如下：

uvicorn main:app --reload
启动FastAPI程序
启动FastAPI程序

上述代码中，main表示启动main.py文件，app表示在main.py内部创建的对象，通过app=FastAPI()创建。--reload表示使服务在更该代码后重新启动，仅在开发时使用。

启动服务后，已经创建了一个API。

  • 在路由“/”和“item/{item_id}”中接收HTTP请求。
  • 两个路径都执行GET操作。
  • 路径“item/{item_id}”有一个路径参数item_id，一般是一个int。
  • 路径“item/{item_id}”有一个可选的str类型的查询参数q。

在浏览器中输入上图中提供的URL（http://127.0.0.1:8000/），匹配index()函数，返回一个JSON响应，运行结果如下图。

运行结果
运行结果

在浏览器中输入“http://127.0.0.1:8000/items/5?q=somequery”，运行结果如下图。

运行结果
运行结果

如果想使用异步async/await功能，应使用async def，示例如下。

@app.get('/')
async def index():
	return {'Hello': 'World'}

API文档

FastAPI提供了两种文档：交互式API文档和备用API文档。

交互式API文档

在浏览器中输入127.0.0.1:8000/docs，将自动显示交互式API文档，如下图。

运行结果
运行结果

点击“/items/{item_id}”选项，会展开URL信息，单击Try It Out按钮，可以填写item_id和q参数。填写完成后点击Execute按钮，运行结果如下图。

运行结果
运行结果

备用API文档

在浏览器中输入“127.0.0.1:8000/redoc”，将显示备用文档，如下图。

备用文档
备用文档

Path路径参数

声明路径参数

可以使用与Python格式化字符串相同的语法来声明路径参数或者变量。如：

from fastapi import FastAPI

app = FastAPI()

@app.get('/items/{item_id}')
async def read_item(item_id):
	return {'item_id': item_id}

路径参数的类型与转换

路径参数有很多类型，如int、float、str。可以使用标准的Python类型注释来声明路径参数的类型。如：

from fastAPI import FastAPI

app = FastAPI()

@app.get('/items/{item_id}')
async def read_item(item_id: int):
	return {'item_id': item_id}

上述示例中，设定了参数item_id的类型为int。此外，可以转换路径参数类型。例如在浏览器中输入127.0.0.1:8000/items/3，item_id就是整型3而不是字符串'3'。因此只要有了类型检查，FastAPI就会自动提供请求解析。

数据类型转换

运行上方的代码时，如果输入了一个非int型的路由，如127.0.0.1:8000/items/foo，将会显示一个友好的HTTP错误页，如下图。

数据类型转换
数据类型转换

错误的原因就是foo不是int类型的数据，由于不存在对应的路由。而尝试输入float类型的数据也会报错。

在浏览器中访问127.0.0.1:8000/docs，将会在交互式API文档中显示参数类型，如下图。

参数类型展示
参数类型展示

指定路径顺序

在创建路径时，可能会遇到一个路由匹配两个路由模式串的情况，这时候就需要指定路径顺序，来决定该路由应该由哪个路由处理程序处理。

在下方的例子中，我们设置两个路由：/user/me和/user/{user_id}，且不为user_id设置类型。这样输入路由/user/me时，它就会同时匹配两个路由。当设定路由顺序时，它们的匹配就有优先级，我们设定/user/me优先于/user/{user_id}。

指定路径顺序的方式就是优先级高的先声明。如：

from fastapi import FastAPI

app = FastAPI()

@app.get('/user/me')
def read_user_me():
    return {'user': 'the current user'}

@app.get('/user/{user_id}')
def read_user_id(user_id):
    return {'user': user_id}
运行结果
运行结果

Query查询参数

Query参数

当声明不属于路径参数的其他函数参数时，FastAPI会将其当作查询参数处理，即“?”后面的参数。如：

from fastapi import FastAPI

app = FastAPI()

@app.get('/')
async def index(a: int = 0, b: int = 0):
    return f'a = {a}, b = {b}'
运行结果
运行结果

由于它们都是URL的一部分，所以本质上都是字符串。当需要使用Python类型来声明Query参数时，它们就会被转换成对应的类型。

设置Query参数

Query参数并不是path中固定的一部分。它们是可选的，并且可以有默认值。

上面的例子中a和b的类型都为整型，它们在输出的时候没有带引号。

如果访问127.0.0.1:8000，由于a和b存在默认值，因此它就相当于访问127.0.0.1:8000?a=0&b=0。

如果URL中没有对应的查询参数，它会赋给该参数默认值。

同样地，也可以声明可选的Query参数，将其默认值设置为None即可。如：

from fastapi import FastAPI

app = FastAPI()

@app.get('/')
async def index(a: int = None, b: int = None):
    return f'a = {a if a else None}, b = {b if b else None}'

在这个例子中，a和b都为可选参数。

Query参数类型转换

若将查询参数设为bool类型，那么传入的参数会被转换。如：

from fastapi import FastAPI

app = FastAPI()

@app.get('/')
async def index(b: bool = None):
    return {'b': b}
运行结果
运行结果

查询参数为1、on、true、True等值时，b参数值为True。查询参数为0、off、false、False等值时，b参数值为False。

同时使用Path和Query参数

当Path和Query参数交替出现时，FastAPI会区分参数。因此，并不需要特意指定它们的顺序，因为这些参数可以被参数名检测到。如：

from fastapi import FastAPI

app = FastAPI()

@app.get('/users/{user_id}/items/{item_id}')
async def index(user_id: int, item_id: int, short: bool):
    return {'user_id': user_id, 'item_id': item_id, 'short': short}
运行结果
运行结果

使用交互式文档添加参数的页面如下图。

运行结果
运行结果

必需的查询参数

当一个查询参数没有默认值时，它就是一个必需参数。当查询参数不包含该参数时，FastAPI就会返回一个友好的错误页面。

from fastapi import FastAPI

app = FastAPI()

@app.get('/')
async def index(short: bool):
    return {'short': short}

上方的例子中，Query参数short没有设置默认值，在大窗口中的URL没有此参数，就会返回错误信息。小窗口中包含了short参数，它就会返回正确的结果。

运行结果
运行结果

Request Body请求体

什么是请求体

将数据从客户端发送到服务器时，可以将其作为请求体发送。请求体是客户端发送到API的数据，响应体是API发送到客户端的数据。API几乎总是必须发送一个响应体，但是客户端并不需要一直发送请求体。

定义请求体需要使用pydantic模型，不能通过GET请求发送请求体。发送请求体数据，必须使用POST、PUT、DELETE、PATCH等请求。

创建数据模型

  1. 定义模型。需要从pydantic中导入BaseModel，代码如下：
from pydantic import BaseModel
  1. 创建模型。声明数据模型为一个类，且继承自BaseModel类，代码如下：
class Item(BaseModel):
    name: str
    desc: str = None
    price: float
    tax: float = None

和Query参数一样，数据类型如果不是必须的话，可以拥有一个默认值。否则，该属性就是必须的。使用None可以让该数据变为可选的。

上面的Item模型可声明为一个JSON对象或Pythondict，如：

{
    'name': 'Foo',
    'desc': 'An optional description',
    'price': 12.3,
    'tax': 3.5
}

desc属性和tax属性是可选的，所以下方的JSON值也是有效的：

{
    'name': 'Foo',
    'price': 12.3
}
  1. 将模型定义为参数。将上方定义的模型添加到路径操作中，和定义Path和Query参数的方法一样，代码如下：
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    desc: str = None
    price: float
    tax: float = None

@app.post('/items')
async def create_item(item: Item):
    return item

声明参数的类型为创建的Item类型。

使用Request Body有以下好处：

  • 将请求的正文读取为JSON类型。
  • 转换相应的类型。
  • 验证数据。若数据无效，它会指出错误所在。
  • 在参数item中提供接收的数据。
  • 这些Schemas将是生成的OpenAPI Schema的一部分，并由自动文档UI使用。

模型的JSON Schenma将成为OpenAPI Schema的一部分，显示在交互式API文档中，并可在需要的路径操作的API文档中使用，如下图。

交互式API文档
交互式API文档

同时定义Path参数、Query参数和Request Body参数

下面通过一个实例介绍如何同时定义这三个参数。

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    desc: str = None
    price: float
    tax: float = None

@app.post('/items/{item_id}')
async def create_item(item: Item, item_id: int, short: bool):
    result = {'item_id': item_id, **item.model_dump()}
    if short:
        return result.update({'short': short})
    return result

FastAPI识别参数的步骤如下：

  • 若在Path中声明了该参数，它将用作path参数。
  • 若参数为单一类型（如int类型），它将被解释为query参数。
  • 若参数声明为pydantic模型的类型，它将被解释为请求体。

在交互式API文档中添加了参数，点击Execute按钮，运行结果如下图。

添加参数
添加参数

Header请求头参数

FastAPI使用Header类可以获取到Header参数。首先导入Header类，代码如下：

from fastapi import FastAPI

app = FastAPI()

@app.get('/items/')
async def read_items(*, user_agent:str = Header(None))
	return {'User-Agent': user_agent}

Header是Path、Query、Cookie的姐妹类，它同样继承自相同的Param类。

声明cookie时需要使用Cookie方法，否则参数会被解释为查询参数。

Header除了提供Path、Query和Cookie之外，还具有一些额外的功能。大多数标准标头都由“连字符”分隔。但是像user-agent这样的变量在Python中是无效的。因此，默认情况下，标头将参数名称字符从下画线转换为连字符以提取并记录标题。

另外，HTTP标头不区分大小写，因此，可以使用Python标准样式（snake_case）声明它们。因此可以像通常在Python代码中那样使用user_agent，而无需将首字符大写。

若出于某种原因需要禁用下画线到连字符的转化，请将Header参数convert_underscores设置为False。需要注意，需要提前记住一些HTTP代理和服务器禁止使用下画线的标头。

Form表单数据

当需要接收的字段为表单而不是JSON时，可以使用Form类。使用Form类需要先安装python-multipart，可以使用pip安装。安装命令如下：

pip install python-multipart

首先，从FastAPI导入Form，代码如下：

from fastapi import FastAPI, Form

接下来创建Form参数，和之前创建Body和Query参数的方法一样，代码如下:

from fastapi import FastAPI, Form

app = FastAPI()

@app.post('/login/')
async def login(*, 
                username: str=Form(...), 
                password: str=Form(...)
):
    return {'username': username, 'password': password}

例如，以一种可以使用OAuth2规范的方式（成为“密码流”）发送“用户名”和“密码”作为表的字段。规范要求这些字段必须准确命名为“用户名”和“密码”，并作为表单字段而不是JSON发送。

使用Form可以声明与Body相同的元数据，验证以及Query、Path和Cookie。

Form是一个类，直接继承自Body。要声明表单主体时，需要明确地使用Form。因为没有它，参数会被解释为查询参数或主体JSON参数。

HTML表单将数据发送至服务器时，通常会对数据添加“特殊”编码，这与JSON不同。FastAPi将确保从正确的位置（而不是JSON）读取数据。

表单的数据通常使用“媒体类型”application/x-www-form-urlencoded进行编码。当表单包含文件时，将被编码为multipart/form-data形式。

操作MySQL数据库

FastAPI不需要安装SQL数据库，但是可以操作任何关系型数据库。下面通过一个简单的示例介绍一下如何使用FastAPI结合SQLAlchemy操作MySQL数据库。

为提高代码的可读性，规定项目结构如下：

项目结构
项目结构

执行如下步骤实现该功能。

  1. 安装SQLAlchemy，命令如下：
pip install sqlalchemy
  1. 创建MySQL数据库，命名为fastapi，命令如下：
CREATE DATABASE fastapi DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
  1. 使用SQLAlchemy连接数据库，在database.py中编写代码如下：
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
SQLALCHEMY_DATABASE_URL = "mysql+pymysql://root:andy123456@localhost/fastapi?charset=utf8"

engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()
  1. 创建数据模型。在models.py中编写如下代码：
from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship

from .database import Base


class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String(255), unique=True, index=True)
    hashed_password = Column(String(255))
    is_active = Column(Boolean, default=True)

    items = relationship("Item", back_populates="owner")


class Item(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String(255), index=True)
    description = Column(String(255), index=True)
    owner_id = Column(Integer, ForeignKey("users.id"))

    owner = relationship("User", back_populates="items")
  1. 创建pydantic模型，在schemas.py中编写如下代码：
from typing import List

from pydantic import BaseModel

class ItemBase(BaseModel):
    title: str
    description: str = None

class ItemCreate(ItemBase):
    pass

class Item(ItemBase):
    id: int
    owner_id: int

    class Config:
        orm_mode = True

class UserBase(BaseModel):
    email: str

class UserCreate(UserBase):
    password: str

class User(UserBase):
    id: int
    is_active: bool
    items: List[Item] = []

    class Config:
        orm_mode = True

设置orm_mode=True是为了实现表关联。

  1. 创建CURD工具集。在curd.py中编写如下代码：
from sqlalchemy.orm import Session

from . import models, schemas


def get_user(db: Session, user_id: int):
    return db.query(models.User).filter(models.User.id == user_id).first()


def get_user_by_email(db: Session, email: str):
    return db.query(models.User).filter(models.User.email == email).first()


def get_users(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.User).offset(skip).limit(limit).all()


def create_user(db: Session, user: schemas.UserCreate):
    fake_hashed_password = user.password + "notreallyhashed"
    db_user = models.User(email=user.email, hashed_password=fake_hashed_password)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user


def get_items(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.Item).offset(skip).limit(limit).all()


def create_user_item(db: Session, item: schemas.ItemCreate, user_id: int):
    db_item = models.Item(**item.dict(), owner_id=user_id)
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item
  1. 创建FastAPI应用。在main.py中编写如下代码：
from typing import List

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session

from . import crud, models, schemas
from .database import SessionLocal, engine

models.Base.metadata.create_all(bind=engine)

app = FastAPI()


# Dependency
def get_db():
    try:
        db = SessionLocal()
        yield db
    finally:
        db.close()


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=List[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)


@app.get("/items/", response_model=List[schemas.Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    items = crud.get_items(db, skip=skip, limit=limit)
    return items

在sql.app同级目录下，执行如下命令：

uvicorn sql_app.main:app --reload
上次编辑于:
贡献者: QI
2024 © ImQi1