介绍
自从引入后端
开发以来,Python 越来越受欢迎,与PHP和.Net等现有的重量级语言竞争。它通过引入简单性和强大功能,使开发人员体验更加高效和简化。尽管众所周知,Python 比同类产品慢,但它在这个生态系统中蓬勃发展。
已经开发了几个用于开发 Web API 的框架,例如Django和Flask,但底层的速度问题始终存在。因此,另一个 Python 框架 FastAPI 被开发出来来解决这个问题。
我们将介绍的步骤:
- 什么是 FastAPI
- 使用 FastAPI 的好处
- FastAPI 与其他 Python 框架的比较
- 通过为库存应用程序构建 REST API 来了解 FastAPI
- FastAPI 中的高级概念
什么是 FastAPI
FastAPI 是一种现代 Python 微框架,可简化使用 Python 编程创建 Web API 的过程。它允许开发人员快速轻松地构建 API,确保最佳性能和轻松管理,而不会影响代码质量和效率。
它提供了许多优势,包括卓越的速度、优于其他几个 Python 后端框架以及与 Express.js 等流行框架竞争。
FastAPI 提供了 Flask 的简单性,因为它与 Flask 非常相似,但仍然包含开箱即用的配置,例如验证、文档和响应编码。
使用 FastAPI 的好处
正如之前所强调的,FastAPI 因其卓越的优势和广泛的优势而脱颖而出。让我们深入研究它提供的一些显着优势:
- 性能:FastAPI 通过充分利用关键库和工具(例如Pydantic和ASGI生态系统)的潜力来最大限度地提高性能。此外,由于其基于 Starlette 框架的坚实基础,它无缝集成了异步/等待功能的强大功能。
- 可扩展性:FastAPI 的模块化和简单性允许与负载均衡器无缝集成,促进可扩展性并确保高效的资源利用。
- 自动文档:通过要求各种 FastAPI 组件的显式定义,Pydantic 的集成允许 FastAPI 能够自动生成其 API 文档。FastAPI 提供 Swagger API 文档。
- 易用性: FastAPI是一个Python框架,因此继承了使用Python的好处。不仅如此,FastAPI 使创建服务器和构建端点变得简单快捷。
- 请求验证:FastAPI 提供请求验证以及用户可读的更详细的错误消息。这也归因于它使用 Pydantic 来进行请求数据类型规范。
FastAPI 与其他 Python 框架的比较
FastAPI 是后端 API 生态系统中相对较新的成员,与 Flask 和 Django 等老牌 Python 巨头竞争。虽然 Flask 和 Django 被认为是该领域的领先框架,但了解它们与 FastAPI 的比较也很重要。
Django 与 FastAPI
Django是一个功能丰富的Python后端框架,包含各种内置库,可以满足各种项目的需求。它具有 ORM、身份验证机制和路由功能等强大功能,适合开发复杂的 Web 应用程序。
另一方面,FastAPI 是一个灵活的微框架,专门设计为轻量级的。虽然它缺乏大型的图书馆生态系统,但它的速度非常快,从而弥补了这一点。与受应用系统限制的 Django 不同,FastAPI 使用现代 Python 技术来释放其固有优势并提高其性能。
Flask 与 FastAPI
Flask 是 Python 开发人员用来快速构建 Web 应用程序的轻量级框架。Flask 以其设计脱颖而出,为开发人员构建应用程序时提供更多控制和灵活性,确保应用程序根据他们的特定需求或要求进行定制。
另一方面,FastAPI 专注于以卓越的速度开发高性能和可扩展的应用程序。它具有本文中讨论的其他优点,并且最适合复杂的应用程序。
Pyramid与 FastAPI
Pyramid 是 Python 最受欢迎的后端框架之一的另一种有趣风格。它秉承“仅使用您需要的”理念,这意味着它提供了一个简约的核心,可以通过各种附加组件和库进行补充。这种模块化方法使开发人员能够挑选和选择他们特定用例所需的组件,从而形成一个轻量级且高度可定制的框架。
另一方面,FastAPI 优先考虑开发人员的生产力和易用性。它具有简单直观的 API 设计,以及清晰的文档和广泛的示例。它还附带了一些预构建的工具,例如自动文档生成。
入门
使用 FastAPI,您可以通过几个步骤轻松设置项目。首先,与任何其他 Python 项目一样,您需要设置虚拟环境。之后,您需要安装软件包 FastAPI 和 Uvicorn。
为此,请运行命令:
python -m pip install fastapi 'uvicorn[standard]'
这样,您就可以开始创建应用程序所需的端点。
fastapi和uvicorn包对于设置 FastAPI 项目至关重要。uvicorn包创建运行 FastAPI 设置的服务器,而fastapi提供创建端点所需的方法和配置。
创建您的第一个应用
要创建您的第一个路由,请创建一个文件main.py,其中将包含您的所有代码。在文本编辑器中打开该文件并添加以下内容:
from fastapi import FastAPI
fastapi = FastAPI()
@fastapi.get("/")
async def home():
return {"data": "Hello World"}
在上面的代码片段中,您会注意到您正在从fastapi模块导入FastAPI类,然后实例化该类。
然后,该类的实例FastAPI可以用作处理函数的装饰器来设置端点。此实例提供 REST API 谓词(例如 PUT、DELETE、PATCH、GET 和 POST)以及设置资源路径的方法。
完成后,您可以在终端上运行以下命令来启动服务器。
uvicorn main:fastapi --reload
main代表模块导入,fastapi是FastAPI的一个实例。上面的命令启动服务器,然后可以通过浏览器访问http://127.0.0.1:8000。
要添加另一个路由,只需创建一个处理函数,例如
def handler():
return { "data": "from handler"}
之后,使用您创建的 FastAPI 实例,添加装饰器
@fastapi.post("/home-page")
到函数。这会将函数转换为 API 端点。
您可以使用@fastapi.post、@fastapi.put或@fastapi.patch或@fastapi.delete创建不同的方法。
使用 FastAPI 模型管理请求和响应主体
向端点发送和接收数据是 API 开发的一个基本方面。在发送方面,向端点发送数据的方法有多种,其中之一包括:
路径参数:此方法涉及将短数据直接附加到 URL 路径。要在FastAPI的端点中实现此功能,您可以参考以下示例:
# other data goes here
@fastapi.get("/{name}")
async def get_name(name: str):
return { "name": name }
在上面的示例中,名称是从 URL 中提取并作为参数传递给get_name函数的路径参数。这样,您就可以方便地访问通过URL路径发送的数据。
查询参数:与路径参数类似,但不同之处在于查询参数附加在 URL 的问号(“?”)之后。要在FastAPI的端点中实现查询参数,您可以参考以下示例:
@fastapi.get("/")
async def get_api_data(data_type: str, skip: int = 0, limit: int = 10):
return { "data_type": data_type, "skip": skip, "limit": limit }
在上面的示例中,skip和limit参数是查询参数。它们是通过端点(“/”)后的 URL 使用?key=value格式提供的。Skip和limit参数的默认值分别为 0 和 10,但可以通过在 URL 中提供不同的值来覆盖。
Body参数:这种类型与前面提到的不同。这是一种对数据进行编码并将其附加到对端点发出的请求的方法。body参数的实现方式如下:
from pydantic import BaseModel
from fastapi import FastAPI
fastapi = FastAPI()
class Item(BaseModel):
name: str
price: float
@fastapi.post("/items")
async def create_item(item: Item):
return {"item": item}
在上面的例子中,我们定义了一个继承自Pydantic 提供的BaseModel 的Item类。该系统定义了整个请求主体参数,并为 FastAPI 提供了更多上下文。
标头和 Cookie:向服务器发送额外上下文信息的主要方法之一是通过标头和 Cookie。要使用 FastAPI 执行此操作,只需按照以下示例操作即可。\
对于标题:
from fastapi import FastAPI, Header
app = FastAPI()
@app.get("/items")
async def read_items(user_agent: str = Header(None)):
return {"User-Agent": user_agent}
在上面的示例中,user_agent参数被赋予默认值Header(None),它告诉 FastAPI 从请求中提取标头的值。如果请求中存在标头,则user_agent将被设置为该标头;否则, user_agent将被设置为None。
饼干:
from fastapi import FastAPI, Cookie
app = FastAPI()
@app.get("/items/")
async def read_items(session_token: str = Cookie(None)):
return {"session_token": session_token}
在这个例子中,与 headers 类似,session_token参数被赋予默认值Cookie(None),指示 FastAPI 应该将请求上设置的 cookie 注入到session_token参数中。如果未设置 cookie,则session_token的值为None。
预览 API 文档
服务器在浏览器上运行后,访问http://127.0.0.1:8000/docs
通过为库存应用程序构建 REST API 来了解 FastAPI
为了演示 FastAPI 的强大功能,您将为假设的库存应用程序构建 REST API。该API将连接到数据库,支持图片上传,并具有受保护的路由。该 API 将具有以下端点:
- GET /items - 获取服务器上存储的所有项目
- GET /items/{item_id} - 从服务器获取特定项目
- POST /items - 将新项目添加到我们的服务器
- PATCH /items/{item_id} - 更新服务器上的项目
- DELETE /items/{item_id} - 从库存中删除项目
然后,您将创建更多端点来处理文件服务,并且还将使用中间件向某些端点添加数据库和某种身份验证。
先决条件
要继续操作,您需要具备以下条件:
- Python 知识
- 了解 HTTP、JSON、REST API 和 Python 的虚拟环境
- 一个终端
- 已安装Python 3.10
设置项目
为此,请按照说明设置虚拟环境。完成后,您必须在虚拟环境文件夹中创建src目录。这是您的代码所在的位置。
使用代码编辑器打开该目录。
安装依赖项
您只需要 Uvicorn 和 FastAPI 等基本依赖项即可完成此任务。但是,由于您将通过合并数据库连接和文件上传系统来增强服务器的功能,因此有必要安装其他依赖项(例如数据库和 SQLAlchemy)。
要继续,请运行命令:
pip install 'fastapi[all]' 'uvicorn[standard]' databases sqlalchemy
这将安装该项目的所有必需的依赖项。
创建您的端点
在src目录中,创建一个新文件main.py;这将是您的应用程序的入口点。
因此,在main.py上添加以下内容:
from fastapi import FastAPI
from fastapi import HTTPException
from .utils import find_item
from pydantic import BaseModel, Field
from typing import Optional
fastapi = FastAPI()
inventory = [
{ "id": 1, "name": "Treasure", "quantity": 3 }
]
class Item(BaseModel):
name: str
quantity: int
class ItemUpdate(BaseModel):
name: Optional[str] = Field(None, description="Optional name of the item")
quantity: Optional[int] = Field(None, description="Optional quantity of the item")
@fastapi.get("/items")
async def get_items():
return {"items": inventory}
@fastapi.get("/items/{item_id}")
async def get_item(item_id: int):
item, idx = find_item(inventory, lambda x: x["id"] == item_id)
return { "item": item }
@fastapi.delete("/items/{item_id}")
async def delete_item(item_id: int, authenticated: bool = Depends(authenticate)):
item, idx = find_item(inventory, lambda x: x["id"] == item_id)
if idx == -1: return HTTPException(404, "item not found")
inventory.pop(idx)
return { "item": item }
@fastapi.post("/items")
async def add_item(data: Item):
item = {
"id": len(inventory) + 1,
"name": data.name,
"quantity": data.quantity
}
inventory.append(item)
return item
@fastapi.patch("/items/{item_id}")
async def update_item(item_id: int, item_update: ItemUpdate):
item, idx = find_item(inventory, lambda x: x["id"] == item_id)
if idx == -1:
raise HTTPException(status_code=404, detail="Item not found")
if item_update.name is not None:
item["name"] = item_update.name
if item_update.quantity is not None:
item["quantity"] = item_update.quantity
inventory[idx] = item
return item
在上面的代码片段中,有五个端点处理程序。您有一个本地库存数据存储,用于保存所有添加的项目。ItemUpdate类指定补丁端点的主体参数,允许可选参数。导入来自typing模块的Optional类和来自pydantic的Field类来创建这些可选字段。
将实用程序函数添加到src 目录中的文件utils.py中。
接下来,您可以运行服务器以通过文档测试端点。
FastAPI 中的高级概念
API 通常不是基本的,例如创建的库存 API。有时,您需要在执行请求或处理文件之前保留数据或验证凭据。
对于您的库存应用程序,您将添加验证中间件、文件上传和数据库。
实施身份验证中间件
中间件就像 API 世界的阀门。它们可以用来做很多事情,比如限制对某些用户的访问、为请求添加额外的上下文等等。为了演示其功能,您将创建一个中间件,允许访问具有特定凭据的客户端,并将其添加到库存应用程序的某些端点。
在src目录中创建一个名为middleware.py的新文件并添加以下代码:
from fastapi import HTTPException, Depends
from fastapi.security import HTTPBasic, HTTPBasicCredentials
security = HTTPBasic()
def authenticate(credentials: HTTPBasicCredentials = Depends(security)):
correct_username = 'admin'
correct_password = 'password'
if credentials.username != correct_username or credentials.password != correct_password:
raise HTTPException(status_code=401, detail="Unauthorized")
return True
在上面的例子中,中间件authenticate实现了基本身份验证。要将其添加到所需的路由处理程序,请转到main.py文件,找到要添加身份验证的端点并添加参数:
authenticated: bool = Depends(authenticate)
之后,处理程序将如下所示:
@fastapi.post("/items")
async def add_item(data: Item, authenticated: bool = Depends(authenticate)):
item = {
"name": data.name,
"id": len(inventory) + 1,
"quantity": data.quantity
}
inventory.append(item)
return item
确保从您创建的middleware.py导入身份验证并从fastapi导入 Depends 。
数据库集成
要将数据库添加到我们现有的库存应用程序中,我们必须首先确保安装了sqlalchemy和数据库。然后在src目录下创建一个文件“ database.py ” ,然后我们需要导入以下内容:
from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
由于 SQLite 的简单性,您将使用它。
接下来,您需要通过将代码添加到文件来创建数据库引擎
DATABASE_URL = "sqlite:///./database.db"
engine = create_engine(DATABASE_URL)
这是数据库的基本连接配置。
之后,创建SessionLocal类,该类在调用时创建一个数据库会话,以及Base类,该类将作为所有数据库模型继承的BaseModel 。
在同一个database.py文件中,添加以下内容:
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
接下来,通过添加以下内容来创建数据库模型:
class DBItem(Base):
__tablename__ = "items"
id = Column(Integer, primary_key=True, index=True)
quantity = Column(Integer)
name = Column(String)
之后,您需要将DBItem和SessionLocal从 database.py文件导入到main.py文件。
此外,请确保相应地更新路由处理程序。
@fastapi.post("/items")
def create_item(item: Item, authenticated: bool = Depends(authenticate)):
db = SessionLocal()
new_item = DBItem(name=item.name, quantity=item.quantity)
db.add(new_item)
db.commit()
db.refresh(new_item)
return {"item" :new_item}
@fastapi.get("/items")
async def get_items():
db = SessionLocal()
items = db.query(DBItem).all()
return { "items": items }
@fastapi.get("/items/{item_id}")
def get_item(item_id: int):
db = SessionLocal()
item = db.query(DBItem).filter(DBItem.id == item_id).first()
if not item:
raise HTTPException(status_code=404, detail="Item not found")
return {"item": item}
@fastapi.patch("/items/{item_id}")
def update_item(item_id: int, item: ItemUpdate, authenticated: bool = Depends(authenticate)):
db = SessionLocal()
db_item = db.query(DBItem).filter(DBItem.id == item_id).first()
if not db_item:
raise HTTPException(status_code=404, detail="Item not found")
db_item.name = item.name
db_item.quantity = item.quantity
db.commit()
db.refresh(db_item)
return { "item": db_item}
@fastapi.delete("/items/{item_id}")
def delete_item(item_id: int, authenticated: bool = Depends(authenticate)):
db = SessionLocal()
db_item = db.query(DBItem).filter(DBItem.id == item_id).first()
if not db_item:
raise HTTPException(status_code=404, detail="Item not found")
db.delete(db_item)
db.commit()
return {"message": "Item deleted"}
在上面的代码中,您会注意到您正在为每个端点实例化 SessionLocal ,然后将其用于执行数据库查询。执行查询后,您可以将其提交到数据库进行持久化。
文件上传
使用FastAPI,可以非常轻松地完成文件上传。我们将使用现有的 API 来演示如何实现文件上传和服务。首先,您必须通过添加以下内容来更新database.py文件中的 DBItem:
class DBItem(Base):
# other database fields goes here
image_src = Column(String)
之后,在main.py上,从fastapi导入File和UploadFile,导入os,然后从starlette.responses导入FileResponse。之后,添加以下内容:
@fastapi.patch("/item-image/{item_id}")
async def upload_file(item_id: int, file: UploadFile = File()):
db = SessionLocal()
db_item = db.query(DBItem).filter(DBItem.id == item_id).first()
if not db_item:
raise HTTPException(status_code=404, detail="Item not found")
file_path = os.path.join("uploads", file.filename)
with open(file_path, "wb") as f:
f.write(await file.read())
db_item.image_src = file.filename
db.commit()
db.refresh(db_item)
return {"item": db_item}
@fastapi.get("/static/{file}")
async def serve_file(file: str):
return FileResponse(os.path.join("uploads", file))
上面代码片段中的第一个路由处理程序更新指定项目的image_src字段并将文件上传到服务器。然后第二个,serve_file,处理从服务器检索文件。
Starlette 提供了几种其他类型的响应,包括FileResponse,它可以在路由处理程序上使用。
可以在此处找到此库存 API 的源代码。
结论
恭喜您已经走到这一步了!到目前为止,您应该已经获得了使用 FastAPI 构建后端应用程序的宝贵见解。FastAPI 是一款卓越、用户友好且高效的 API 开发工具。它提供了与微框架相关的灵活性并提供卓越的性能,使其成为满足您的 API 开发需求的绝佳选择。