使用 Python FastAPI 实现分层架构

在现代web应用开发中，使用分层架构可以使应用保持高内聚、低耦合，易于管理和扩展。FastAPI是一种极其高效的Python Web框架，适合构建RESTful API。下面，我们将详细介绍如何用FastAPI实现分层架构，并提供一个完整的示例。

流程概述

以下是实现FastAPI分层架构的主要步骤：

甘特图表示

以下是本项目的甘特图，用于可视化任务的进度：

gantt
    title FastAPI分层架构项目进度
    dateFormat  YYYY-MM-DD
    section 项目设计
    项目结构设计         :done, 2023-10-01, 1d
    section 数据模型
    创建基础模型         :done, 2023-10-02, 2d
    section 数据层
    创建数据访问层       :done, 2023-10-04, 3d
    section 业务逻辑层
    创建业务逻辑层       :done, 2023-10-07, 3d
    section API层
    创建API层            :active, 2023-10-10, 3d
    section 测试
    启动和测试应用       : 2023-10-13, 2d

步骤详解

1. 项目结构设计

首先，我们需要定义一个合理的项目结构。这有助于管理代码和模块化。

my_fastapi_app/
├── main.py
├── models/
│   └── __init__.py
│   └── item.py
├── database/
│   └── __init__.py
│   └── db.py
├── services/
│   └── __init__.py
│   └── item_service.py
└── routers/
    └── __init__.py
    └── item_router.py

2. 创建基础模型

我们需要创建数据模型，表示我们的数据结构。

在models/item.py中定义模型：

from pydantic import BaseModel

class Item(BaseModel):
    id: int
    name: str
    description: str = None
    price: float
    is_offer: bool = None

# Item类定义了项目的基本结构，包括ID、名称、描述、价格及是否为特价等字段。

3. 创建数据访问层

接下来，我们需要实现与数据库交互的数据访问层。

在database/db.py中定义数据库连接：

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"  # 使用SQLite数据库

engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

# 上述代码设置了数据库的连接和会话，使用了SQLite数据库，适合快速开发和测试。

在models/item.py中，引入数据库模型：

from sqlalchemy import Boolean, Column, Float, Integer, String
from database.db import Base

class ItemORM(Base):
    __tablename__ = "items"
    
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    description = Column(String, index=True)
    price = Column(Float, index=True)
    is_offer = Column(Boolean, default=False)

# ItemORM类定义了SQLite数据库中的items表，字段与基础模型相对应。

4. 创建业务逻辑层

在这里我们将创建一个用于处理业务逻辑的服务层。

在services/item_service.py中：

from sqlalchemy.orm import Session
from models.item import ItemORM, Item

def get_item(db: Session, item_id: int):
    return db.query(ItemORM).filter(ItemORM.id == item_id).first()

def create_item(db: Session, item: Item):
    db_item = ItemORM(**item.dict())
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

# get_item 和 create_item 函数分别用于获取和创建项目，利用SQLAlchemy执行数据库操作。

5. 创建API层

最后，我们需要连接相应的HTTP请求和业务层。

在routers/item_router.py中：

from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from database.db import SessionLocal
from models.item import Item
from services.item_service import get_item, create_item

router = APIRouter()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@router.post("/items/", response_model=Item)
def create_item_route(item: Item, db: Session = Depends(get_db)):
    return create_item(db=db, item=item)

@router.get("/items/{item_id}", response_model=Item)
def read_item(item_id: int, db: Session = Depends(get_db)):
    db_item = get_item(db=db, item_id=item_id)
    if db_item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return db_item

# 上述代码定义了"/items/" POST接口和"/items/{item_id}" GET接口，使用业务逻辑层的函数来处理具体请求。

在main.py中启动FastAPI应用：

from fastapi import FastAPI
from routers.item_router import router as item_router
from database.db import Base, engine

Base.metadata.create_all(bind=engine)  # 创建数据库表

app = FastAPI()

app.include_router(item_router)

# 启动FastAPI应用，将路由器item_router包含在内。

6. 启动和测试应用

运行应用：

uvicorn main:app --reload

测试API，可以使用工具如Postman或cURL。此外，还可以通过浏览器访问

结论

通过以上步骤，我们实现了一个基于FastAPI的简单分层架构。项目是模块化的，按照功能分层，易于扩展和维护。后续可以在此基础上不断增加功能，优化项目结构。

分层架构模式将帮助开发者在处理复杂业务时保持代码的可读性和可维护性。希望这个示例能够帮助到刚入行的小白开发者。在你的学习和项目开发中，实践和摸索是提升技能的关键！