跳至主要內容

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
上次编辑于:
贡献者: 棋.