FastAPI笔记
认识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程序](/assets/image-20231004201853251-rCGqxdgo.png)
上述代码中,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响应,运行结果如下图。
![运行结果](/assets/image-20231004202225412-2deX8l3L.png)
在浏览器中输入“http://127.0.0.1:8000/items/5?q=somequery
”,运行结果如下图。
![运行结果](/assets/image-20231004202321125-SiOFFpbB.png)
如果想使用异步async/await功能,应使用async def,示例如下。
@app.get('/')
async def index():
return {'Hello': 'World'}
API文档
FastAPI提供了两种文档:交互式API文档和备用API文档。
交互式API文档
在浏览器中输入127.0.0.1:8000/docs,将自动显示交互式API文档,如下图。
![运行结果](/assets/image-20231005094816411-Nbkj1eNF.png)
点击“/items/{item_id}”选项,会展开URL信息,单击Try It Out按钮,可以填写item_id和q参数。填写完成后点击Execute按钮,运行结果如下图。
![运行结果](/assets/image-20231005095049493-KCWndUDL.png)
备用API文档
在浏览器中输入“127.0.0.1:8000/redoc”,将显示备用文档,如下图。
![备用文档](/assets/image-20231005095152865-V3a7duII.png)
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错误页,如下图。
![数据类型转换](/assets/image-20231005114945795-eFl_aH1B.png)
错误的原因就是foo不是int类型的数据,由于不存在对应的路由。而尝试输入float类型的数据也会报错。
在浏览器中访问127.0.0.1:8000/docs,将会在交互式API文档中显示参数类型,如下图。
![参数类型展示](/assets/image-20231005115208130-Fg-HuOSM.png)
指定路径顺序
在创建路径时,可能会遇到一个路由匹配两个路由模式串的情况,这时候就需要指定路径顺序,来决定该路由应该由哪个路由处理程序处理。
在下方的例子中,我们设置两个路由:/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}
![运行结果](/assets/image-20231005183030191-P464mJOo.png)
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}'
![运行结果](/assets/image-20231005183559762-afY4dw83.png)
由于它们都是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}
![运行结果](/assets/image-20231005185000619-1dL9ulGR.png)
查询参数为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}
![运行结果](/assets/image-20231006104803999-4Z2x2ArW.png)
使用交互式文档添加参数的页面如下图。
![运行结果](/assets/image-20231006104937943-vYZ4bcJI.png)
必需的查询参数
当一个查询参数没有默认值时,它就是一个必需参数。当查询参数不包含该参数时,FastAPI就会返回一个友好的错误页面。
from fastapi import FastAPI
app = FastAPI()
@app.get('/')
async def index(short: bool):
return {'short': short}
上方的例子中,Query参数short没有设置默认值,在大窗口中的URL没有此参数,就会返回错误信息。小窗口中包含了short参数,它就会返回正确的结果。
![运行结果](/assets/image-20231006105121198-heP89kb2.png)
Request Body请求体
什么是请求体
将数据从客户端发送到服务器时,可以将其作为请求体发送。请求体是客户端发送到API的数据,响应体是API发送到客户端的数据。API几乎总是必须发送一个响应体,但是客户端并不需要一直发送请求体。
定义请求体需要使用pydantic模型,不能通过GET请求发送请求体。发送请求体数据,必须使用POST、PUT、DELETE、PATCH等请求。
创建数据模型
- 定义模型。需要从pydantic中导入BaseModel,代码如下:
from pydantic import BaseModel
- 创建模型。声明数据模型为一个类,且继承自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
}
- 将模型定义为参数。将上方定义的模型添加到路径操作中,和定义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文档](/assets/image-20231006110546175-huC8lsZp.png)
同时定义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按钮,运行结果如下图。
![添加参数](/assets/image-20231006112712141-0cPVkSsN.png)
![](/assets/image-20231006112734711-tPv_FH5l.png)
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数据库。
为提高代码的可读性,规定项目结构如下:
![项目结构](/assets/image-20231022215101568-_b_ANfCP.png)
执行如下步骤实现该功能。
- 安装SQLAlchemy,命令如下:
pip install sqlalchemy
- 创建MySQL数据库,命名为fastapi,命令如下:
CREATE DATABASE fastapi DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
- 使用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()
- 创建数据模型。在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")
- 创建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是为了实现表关联。
- 创建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
- 创建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