当使用FastAPI构建后端时，通常会从一个单一的 app.py 文件开始。虽然这种方法适用于小型项目，但随着应用程序的增长，维护和扩展会变得越来越困难。
在这篇博客文章中，我们将探讨如何将一个使用单体 app.py 文件构建的 FastAPI 应用程序 重构为一个更结构化的架构，该架构由 Routers、Controllers、Services 和 Repositories 组成。
在我们开始重构过程之前，让我们先看一下我们将要处理的API。我们将构建一个简单的Todo应用程序，包含以下端点，
这些API将允许用户对待办事项执行CRUD（创建、读取、更新、删除）操作。每个待办事项将具有以下属性，
**id**: 一个唯一的待办事项标识符
**title**: 待办事项的标题或描述
**completed**: 一个布尔值，表示待办事项是否已完成
现在我们已经了解了我们要处理的API，让我们来看看一些先决条件以及如何实现它们。
前置条件在我们开始最初的实现和重构之前，让我们先设置我们的 FastAPI 项目。
初始方法：app.py 中包含所有内容安装Python
设置并激活虚拟环境
python3 -m venv venv source env/bin/activate # 在Windows上使用 `venv\Scripts\activate`创建
requirements.txt并添加FastAPI和Uvicorn到依赖列表中fastapi uvicorn安装依赖
pip3 install -r requirements.txt
让我们从一个简单的Todo API 开始，该API 完全实现在根目录下的 app.py 文件中：
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
app = FastAPI()
# Pydantic 模型用于创建一个新的待办事项
class TodoCreate(BaseModel):
title: str
# Pydantic 模型用于表示一个待办事项，继承自 TodoCreate 并添加了 id 和 completed 字段
class Todo(TodoCreate):
id: int
completed: bool = False
# 内存存储用于待办事项（模拟数据库）
todos = []
# 端点用于创建一个新的待办事项
@app.post("/todos", response_model=Todo)
def create_todo(todo: TodoCreate):
# 创建一个新的待办事项，id 自增
new_todo = Todo(id=len(todos) + 1, **todo.model_dump())
todos.append(new_todo) # 将新待办事项添加到列表中
return new_todo # 返回创建的待办事项作为响应
# 端点用于获取所有待办事项
@app.get("/todos", response_model=list[Todo])
def get_todos():
return todos # 返回待办事项列表作为响应
# 端点用于通过其 id 获取特定的待办事项
@app.get("/todos/{todo_id}", response_model=Todo)
def get_todo(todo_id: int):
for todo in todos:
if todo.id == todo_id:
return todo # 如果找到待办事项，则返回
# 如果未找到待办事项，则抛出 HTTPException，状态码为 404，消息为 "待办事项未找到"
raise HTTPException(status_code=404, detail="待办事项未找到")
# 端点用于通过其 id 更新待办事项
@app.put("/todos/{todo_id}", response_model=Todo)
def update_todo(todo_id: int, updated_todo: TodoCreate):
for todo in todos:
if todo.id == todo_id:
todo.title = updated_todo.title # 更新待办事项的标题
return todo # 返回更新后的待办事项
# 如果未找到待办事项，则抛出 HTTPException，状态码为 404，消息为 "待办事项未找到"
raise HTTPException(status_code=404, detail="待办事项未找到")
# 端点用于通过其 id 删除待办事项
@app.delete("/todos/{todo_id}")
def delete_todo(todo_id: int):
for index, todo in enumerate(todos):
if todo.id == todo_id:
del todos[index] # 从列表中删除待办事项
return {"message": "待办事项删除成功"} # 返回成功消息
# 如果未找到待办事项，则抛出 HTTPException，状态码为 404，消息为 "待办事项未找到"
raise HTTPException(status_code=404, detail="待办事项未找到")
# 主块用于使用 Uvicorn 服务器运行应用程序
if __name__ == "__main__":
import uvicorn
uvicorn.run("app:app", port=3000, host="0.0.0.0", reload=True)Pydantic 是一个用于 Python 的数据验证和解析库，它使用 Python 类型注解来定义和验证数据结构。通过提供自动类型检查和解析，Pydantic 确保数据的完整性，使得处理结构化数据更加容易。
为了启动API，我们使用命令 ,**_python3 app.py_**
嗯……
虽然这种方法适用于小型应用程序，但它存在几个缺点：
- 所有路由、业务逻辑和数据存储都混合在同一个文件中。
- 随着应用程序的增长，维护和扩展变得困难。
- 测试各个组件变得具有挑战性。
- 代码的重用性有限。
让我们开始重构之旅…
介绍路由结构我们应用程序的第一步是引入路由器。
API 中的路由器将相关的端点（路由）组织并分组到单独的模块或文件中，增强了代码的模块化和结构。
创建一个名为 routers 的新目录，并添加一个名为 todo_router.py 的文件
从 fastapi 导入 APIRouter
router = APIRouter()
@router.post("/todos")
def 创建待办事项():
pass
@router.get("/todos")
def 获取待办事项():
pass
@router.get("/todos/{todo_id}")
def 获取单个待办事项(todo_id: int):
pass
@router.put("/todos/{todo_id}")
def 更新待办事项(todo_id: int):
pass
@router.delete("/todos/{todo_id}")
def 删除待办事项(todo_id: int):
pass现在更新 app.py 以使用路由器，
from fastapi import FastAPI
from routers import todo_router
app = FastAPI()
app.include_router(todo_router.router)
if __name__ == "__main__":
import uvicorn
uvicorn.run("app:app", port=3000, host="0.0.0.0", reload=True)通过引入路由器，我们将与待办事项相关的路由从主 app.py 文件中分离出来，使其更加清晰和专注。
接下来，我们将介绍控制器来处理请求处理逻辑。
控制器处理传入的请求，协调应用程序的逻辑，并管理API路由（端点）和业务层之间数据的流动。
创建一个名为 controllers 的新目录，并添加一个名为 todo_controller.py 的文件
from fastapi import HTTPException
from pydantic import BaseModel
class TodoCreate(BaseModel):
标题: str
class Todo(TodoCreate):
id: int
完成: bool = False
class TodoController:
def __init__(self):
self.todos = []
def 创建待办事项(self, 待办事项: TodoCreate):
新待办事项 = Todo(id=len(self.todos) + 1, **待办事项.model_dump())
self.todos.append(新待办事项)
return 新待办事项
def 获取待办事项(self):
return self.todos
def 获取单个待办事项(self, 待办事项_id: int):
for 待办事项 in self.todos:
if 待办事项.id == 待办事项_id:
return 待办事项
raise HTTPException(status_code=404, detail="待办事项未找到")
def 更新待办事项(self, 待办事项_id: int, 更新后的待办事项: TodoCreate):
for 待办事项 in self.todos:
if 待办事项.id == 待办事项_id:
待办事项.标题 = 更新后的待办事项.标题
return 待办事项
raise HTTPException(status_code=404, detail="待办事项未找到")
def 删除待办事项(self, 待办事项_id: int):
for index, 待办事项 in enumerate(self.todos):
if 待办事项.id == 待办事项_id:
del self.todos[index]
return {"message": "待办事项删除成功"}
raise HTTPException(status_code=404, detail="待办事项未找到")更新 todo_router.py 以使用控制器，
from fastapi import APIRouter
from controllers.todo_controller import TodoController, TodoCreate, Todo
router = APIRouter()
todo_controller = TodoController()
@router.post("/todos", response_model=Todo)
def create_todo(todo: TodoCreate):
return todo_controller.create_todo(todo)
@router.get("/todos", response_model=list[Todo])
def get_todos():
return todo_controller.get_todos()
@router.get("/todos/{todo_id}", response_model=Todo)
def get_todo(todo_id: int):
return todo_controller.get_todo(todo_id)
@router.put("/todos/{todo_id}", response_model=Todo)
def update_todo(todo_id: int, updated_todo: TodoCreate):
return todo_controller.update_todo(todo_id, updated_todo)
@router.delete("/todos/{todo_id}")
def delete_todo(todo_id: int):
return todo_controller.delete_todo(todo_id)现在，让我们引入一个服务层来处理业务逻辑。
服务包含了API的核心业务逻辑，实现了应用程序所需的具体功能和操作。它们将复杂的业务规则从API端点中抽象出来。
创建一个名为 services 的新目录，并添加一个名为 todo_service.py 的文件
从 pydantic 导入 BaseModel
class TodoCreate(BaseModel):
title: str
class Todo(TodoCreate):
id: int
completed: bool = False
class TodoService:
def __init__(self):
self.todos = []
def create_todo(self, todo: TodoCreate) -> Todo:
new_todo = Todo(id=len(self.todos) + 1, **todo.model_dump())
self.todos.append(new_todo)
return new_todo
def get_todos(self) -> list[Todo]:
return self.todos
def get_todo(self, todo_id: int) -> Todo | None:
for todo in self.todos:
if todo.id == todo_id:
return todo
return None
def update_todo(self, todo_id: int, updated_todo: TodoCreate) -> Todo | None:
for todo in self.todos:
if todo.id == todo_id:
todo.title = updated_todo.title
return todo
return None
def delete_todo(self, todo_id: int) -> bool:
for index, todo in enumerate(self.todos):
if todo.id == todo_id:
del self.todos[index]
return True
return False更新 todo_controller.py 以使用服务，
from fastapi import HTTPException
from services.todo_service import TodoService, TodoCreate, Todo
class TodoController:
def __init__(self):
self.todo_service = TodoService()
def create_todo(self, todo: TodoCreate):
return self.todo_service.create_todo(todo)
def get_todos(self):
return self.todo_service.get_todos()
def get_todo(self, todo_id: int):
todo = self.todo_service.get_todo(todo_id)
if todo is None:
raise HTTPException(status_code=404, detail="Todo not found")
return todo
def update_todo(self, todo_id: int, updated_todo: TodoCreate):
todo = self.todo_service.update_todo(todo_id, updated_todo)
if todo is None:
raise HTTPException(status_code=404, detail="Todo not found")
return todo
def delete_todo(self, todo_id: int):
if self.todo_service.delete_todo(todo_id):
return {"message": "Todo deleted successfully"}
raise HTTPException(status_code=404, detail="Todo not found")最后，让我们介绍一个仓库层来处理数据持久化。
仓库为数据持久化操作提供了抽象层，封装了与数据库或外部数据源的交互。它们提供了存储、检索、更新和删除数据的标准方法。
创建一个名为 repositories 的新目录，并添加一个名为 todo_repository.py 的文件
from pydantic import BaseModel
class TodoCreate(BaseModel):
title: str
class Todo(TodoCreate):
id: int
completed: bool = False
class TodoRepository:
def __init__(self):
self.todos = []
def create_todo(self, todo: TodoCreate) -> Todo:
new_todo = Todo(id=len(self.todos) + 1, **todo.model_dump())
self.todos.append(new_todo)
return new_todo
def get_todos(self) -> list[Todo]:
return self.todos
def get_todo(self, todo_id: int) -> Todo | None:
for todo in self.todos:
if todo.id == todo_id:
return todo
return None
def update_todo(self, todo_id: int, updated_todo: TodoCreate) -> Todo | None:
for todo in self.todos:
if todo.id == todo_id:
todo.title = updated_todo.title
return todo
return None
def delete_todo(self, todo_id: int) -> bool:
for index, todo in enumerate(self.todos):
if todo.id == todo_id:
del self.todos[index]
return True
return False更新 todo_service.py 以使用仓库，
from repositories.todo_repository import TodoRepository, TodoCreate, Todo
class TodoService:
def __init__(self):
self.todo_repository = TodoRepository()
def create_todo(self, todo: TodoCreate) -> Todo:
return self.todo_repository.create_todo(todo)
def get_todos(self) -> list[Todo]:
return self.todo_repository.get_todos()
def get_todo(self, todo_id: int) -> Todo | None:
return self.todo_repository.get_todo(todo_id)
def update_todo(self, todo_id: int, updated_todo: TodoCreate) -> Todo | None:
return self.todo_repository.update_todo(todo_id, updated_todo)
def delete_todo(self, todo_id: int) -> bool:
return self.todo_repository.delete_todo(todo_id)我们的重构之旅到这里就结束了…
在这次从单一的 app.py 到使用 FastAPI 构建的结构化、模块化架构的旅程中，我们将 Todo API 转变为一个更可扩展和易于维护的应用程序。通过采用路由器、控制器、服务和仓库，我们实现了更清晰的关注点分离，并增强了管理项目增长时复杂性的能力。
- 提高可维护性: 每个组件——路由、控制器、服务和仓库——现在都处理特定的责任，减少了在进行更改时产生意外副作用的风险。
- 增强可测试性: 通过分层设计，单元测试变得更加简单。我们可以独立测试每个组件，确保应用程序的健壮性和可靠性。
- 可扩展性和灵活性: 模块化设计使可扩展性更容易。新功能可以添加或现有功能可以修改，而无需对整个代码库进行广泛的重构。这种灵活性还扩展到无缝切换数据库或更新业务逻辑。
我们的仓库现在看起来像以下这样……
通过将我们的FastAPI应用程序重构为模块化架构，我们为持续的增长和灵活性奠定了坚实的基础。这种方法不仅增强了我们当前的开发工作，还为我们未来的挑战和机遇做好了准备。
拥抱模块化不仅仅是关于组织代码，更是关于培养一种高效、协作和持续改进的文化。
再见!!
點擊查看更多內容
為 TA 點贊
評論
評論
共同學習，寫下你的評論
評論加載中...
作者其他優質文章
正在加載中
感謝您的支持，我會繼續努力的～
掃碼打賞，你說多少就多少
贊賞金額會直接到老師賬戶
支付方式
打開微信掃一掃，即可進行掃碼打賞哦