fastapi教程（一）：初识 fastapi

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

关键特性:

快速：可与 NodeJS 和 Go 并肩的极高性能（归功于 Starlette 和 Pydantic）。最快的 Python web 框架之一。
高效编码：提高功能开发速度约 200％至 300％。
更少 bug：减少约 40％的人为（开发者）导致错误。
智能：极佳的编辑器支持。处处皆可自动补全，减少调试时间。
简单：设计的易于使用和学习，阅读文档的时间更短。
简短：使代码重复最小化。通过不同的参数声明实现丰富功能。bug 更少。
健壮：生产可用级别的代码。还有自动生成的交互式文档。
标准化：基于（并完全兼容）API 的相关开放标准：OpenAPI (以前被称为 Swagger) 和 JSON Schema。

简单来体验一下：
环境：

Python 3.11
Windows 10
fastapi 0.68.0

1，安装 fastapi

pip install fastapi

2，安装一个 ASGI 服务器

pip install "uvicorn[standard]"

生产环境可以使用 Uvicorn 或者 Hypercorn。

ASGI(Asynchronous Server Gateway Interface)是 Python 异步服务器网关接口的缩写。
它定义了异步编程风格的 Python Web 服务器与 Web 服务器和应用程序之间的标准接口。
ASGI 旨在解决传统 Python Web 服务器(如WSGI)在高并发和长连接场景下的不足，使用异步编程模型来提高服务器的并发处理能力。
ASGI服务器具有以下几个关键特点:

异步和事件驱动：ASGI 服务器采用异步非阻塞的编程模型，可以高效处理大量并发连接，避免进程/线程阻塞。它使用事件循环机制，在有事件到来时处理，否则释放资源做其他事情。
支持长连接和WebSocket：ASGI 服务器本身支持长连接和 WebSocket 协议，可用于构建实时通信应用，如在线聊天室、协作平台等。
标准统一接口：ASGI 定义了统一的服务器端接口规范，使Web服务器、中间件和应用程序之间松散耦合，易于拓展和维护。
高性能：相比传统 WSGI，ASGI 编程模型使用异步协程可以大幅降低上下文切换和内存开销，从而提高并发性能。

目前支持 ASGI 的服务器有 Uvicorn、Daphne、Hypercorn 等，应用框架包括 Django Channels、FastAPI 等。ASGI 正在逐渐取代 WSGI 成为 Python Web 应用开发的新标准，特别适合于需要高并发、实时双向通信的应用场景。

3，创建一个 main.py 文件

from typing import Unionfrom fastapi import FastAPI# 创建 FastAPI 实例
app = FastAPI()# 定义路由，处理 get 请求
@app.get("/")
# 定义路由处理函数
async def read_root():return {"Hello": "World"}# 定义路由，处理 get 请求，接收一个路径参数 item_id
@app.get("/items/{item_id}")
async def read_item(item_id: int,q: str = 'default',
):return {"item_id": item_id, "q": q}

4，运行项目

$ uvicorn main:app --reload
INFO:     Will watch for changes in these directories: ['I:\\Python\\Web\\FastAPI\\fastApiProject1']
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [14776] using WatchFiles
INFO:     Started server process [24944]
INFO:     Waiting for application startup.
INFO:     Application startup complete.

main：main.py 文件（一个 Python “模块”）。
app：在 main.py 文件中通过 app = FastAPI() 创建的对象。
–reload：让服务器在更新代码后重新启动。仅在开发时使用该选项。

5，浏览器访问路由

访问 http://127.0.0.1:8000/ 可以在页面看到路由处理函数返回的数据：
在这里插入图片描述
访问 http://127.0.0.1:8000/items/5?q=somequery 可以在页面看到路由处理函数返回的数据：

6，体验类型声明的优势

浏览器访问 http://127.0.0.1:8000/docs 可以到达 API 界面：
在这里插入图片描述

但是国内可能打不开。
fastapi访问/docs和/redoc接口文档显示空白或无法加载
FastAPI 自动文档Swagger UI 打不开。显示空白

打开这个页面，我们可以看到非常详细的 API 描述：
在这里插入图片描述

item_id：integer类型，是一个路径参数，而且是必需的。
q：string 类型，是一个查询参数，而且不是必需的。

这是怎么做到的？回到代码：

@app.get("/items/{item_id}")
def read_item(item_id: int,q: str = 'default',
):return {"item_id": item_id, "q": q}

路由装饰器中定义了路径 /items/{item_id}，路径参数 item_id 在路由处理函数中明确定义为 int 数据类型，且未指定默认值，路径参数是用户必须传递的。

路由处理函数中，q 明确定义为 str 数据类型，且赋予了一个默认值 default，这种不是路径参数的路由处理函数参数就是查询参数。

把默认值设为 None 即可声明可选的查询参数

最重要的特征就是，我们为参数添加了类型声明，这样做有非常多好处：

由 FastAPI 通过类型声明自动将解析请求中的字符串转为声明的数据类型：

2，更方便地利用编辑器的自动提示：

7，借助 Pydantic 完成数据验证

多数情况我们会用 POST（最常用）、PUT 操作来发送数据，具体就是将数据放在这些请求的请求体中。
既然路径参数和查询参数都可以通过定义数据类型来完成默认的类型转化和校验，那么作为传述数据主要载体的请求体，自然也支持数据类型的定义带来的好处，响应体也是如此！

请求体和响应体的数据定义需要使用 pandatic，这是一个流行的数据验证工具。使用 Pydantic 模型声明请求体，能充分利用它的功能和优点：

以 JSON 形式读取请求体。（在必要时）把请求体转换为对应的类型
校验数据：数据无效时返回错误信息，并指出错误数据的确切位置和内容
把接收的数据赋值给参数 item：把函数中请求体参数的类型声明为 Item，还能获得代码补全等编辑器支持
为模型生成 JSON Schema，在项目中所需的位置使用
用于 API 文档 UI

from fastapi import FastAPI
from pydantic import BaseModelclass Item(BaseModel):name: strdescription: str | None = Noneprice: floattax: float | None = Noneapp = FastAPI()@app.post("/items/")
async def create_item(item: Item):return item

修改代码并保存后，服务器将会自动重载（因为在上面的步骤中你向 uvicorn 命令添加了 --reload 选项来启动服务）

我们通过使继承 pydantic 的 BaseModel 类来定义了一个类，通常会将它称为模型，在这个模型中我们定义了一些字段及其数据类型，然后我们用这个模型定义一个查询参数的类型。

使用Pydantic 模型的参数，是请求体。
与声明查询参数一样，包含默认值的模型属性是可选的，否则就是必选的。默认值为 None 的模型属性也是可选的。

上述模型声明如下 JSON 对象（即 Python 字典）：

{"name": "Foo","description": "An optional description","price": 45.2,"tax": 3.5
}

运行项目，使用接口调试工具进行调试：
在这里插入图片描述
如果请求体数据必填项缺失：

发送：
{"name": "Foo","description": "An optional description","tax": 3.5
}
返回：
{"detail": [{"type": "missing","loc": ["body","price"],"msg": "Field required","input": {"name": "Foo","description": "An optional description","tax": 3.5}}]
}

如果数据类型错误：

发送：
{"name": "Foo","description": "An optional description","price": "error type","tax": 3.5
}
返回：
{"detail": [{"type": "float_parsing","loc": ["body","price"],"msg": "Input should be a valid number, unable to parse string as a number","input": "error type"}]
}

这就是 pandatic 实现请求体数据校验的最大好处！

接下来看看响应体：可以在任意的路径操作中使用 response_model 参数来声明用于响应的模型。

from typing import Anyfrom fastapi import FastAPI
from pydantic import BaseModel, EmailStrapp = FastAPI()class UserIn(BaseModel):username: strpassword: stremail: EmailStrfull_name: str | None = Noneclass UserOut(BaseModel):username: stremail: EmailStrfull_name: str | None = None@app.post("/user/", response_model=UserOut)
async def create_user(user: UserIn) -> Any:return userif __name__ == "__main__":import uvicornuvicorn.run(app)

即便我们的路径操作函数将会返回包含密码的相同输入用户，但我们已经将 response_model 声明为了不包含密码的 UserOut 模型，FastAPI 将会负责过滤掉未在输出模型中声明的所有数据。

发送：
{"username": "Jonas Vingegaard","password": "123456","email": "test@test.com","full_name": "Jonas Vingegaard Rasmussen"
}
返回：
{"username": "Jonas Vingegaard","email": "test@test.com","full_name": "Jonas Vingegaard Rasmussen"
}发送：
{"username": "Jonas Vingegaard","password": "123456","email": "test@test.com"
}
返回：
{"username": "Jonas Vingegaard","email": "test@test.com","full_name": null
}发送：
{"username": "Jonas Vingegaard","password": "123456"
}
返回：
{"detail": [{"type": "missing","loc": ["body","email"],"msg": "Field required","input": {"username": "Jonas Vingegaard","password": "123456"}}]
}

8，使用数据库

任何SQLAlchemy支持的数据库，如：

PostgreSQL
MySQL
SQLite
Oracle
Microsoft SQL Server，等等其它数据库

首先你需要安装SQLAlchemy:

pip install sqlalchemy

创建目录：

└── sql_app├── __init__.py├── crud.py├── database.py├── main.py├── models.py└── schemas.py

定义数据库：

# sql_app/database.pyfrom sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker# 为 SQLAlchemy 定义数据库 URL地址
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
# SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"# 创建 SQLAlchemy 引擎
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)# 创建一个SessionLocal类
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)# 创建一个Base类
Base = declarative_base()

创建数据库模型:

# sql_app/models.pyfrom sqlalchemy import Boolean, Column, ForeignKey, Integer, Stringfrom sql_app.database import Baseclass User(Base):__tablename__ = "users"__table_args__ = {'extend_existing': True}id = Column(Integer, primary_key=True)email = Column(String, unique=True)hashed_password = Column(String)is_active = Column(Boolean, default=True)

创建 Pydantic 模型:

# sql_app/schemas.pyfrom pydantic import BaseModelclass UserBase(BaseModel):email: strclass UserCreate(UserBase):password: strclass User(UserBase):id: intis_active: boolclass Config:orm_mode = True

主程序——controller：

# main.pyfrom fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session
from sql_app import crud, models, schemas
from sql_app.database import SessionLocal, enginemodels.Base.metadata.create_all(bind=engine)app = FastAPI()# Dependency
def get_db():db = SessionLocal()try:yield dbfinally: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_userif __name__ == "__main__":import uvicornuvicorn.run(app)

添加 crud 操作实现：

# crud.pyfrom sqlalchemy.orm import Sessionfrom sql_app import models, schemasdef 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

去启动程序，访问 API 页面：
在这里插入图片描述
POST 创建一些数据，然后就能 GET 它们了。

9，文件的上传与下载

简单的文件上传：

from urllib.parse import quotefrom fastapi import FastAPI, UploadFile
from fastapi.responses import StreamingResponseapp = FastAPI()@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile):# 分块将大文件保存到磁盘file_object = file.filefile_name = file.filename# 判断文件夹是否存在，不存在则创建import osif not os.path.exists("files"):os.mkdir("files")# 将文件写入到文件夹with open(f"files/{file_name}", "wb") as f:while True:# 读取文件chunk = file_object.read(10000)if not chunk:breakf.write(chunk)return {"filename": file.filename}if __name__ == "__main__":import uvicornuvicorn.run(app)

大文件下载：


def file_iterator(file_path, chunk_size=8192):with open(file_path, mode="rb") as file_like:while True:chunk = file_like.read(chunk_size)if not chunk:breakyield chunk# 流式下载文件
@app.get("/downloadfile/")
async def download_file(file_name: str):file_path = f"files/{file_name}"# 判断文件是否存在import osif not os.path.exists(file_path):return {"msg": "file not found"}else:headers = {'Content-Disposition': f'attachment; filename="{quote(file_name)}"'}return StreamingResponse(file_iterator(file_path),media_type="application/octet-stream",headers=headers)

10，部署 fastapi 项目

先收集项目依赖：

pip freeze > requirements.txt

然后将项目发送到服务器。
在服务器安装 ASGI 服务器：

pip install "uvicorn[standard]"

然后运行服务器程序：

uvicorn main:app --host 0.0.0.0 --port 80