Python后端标准分包思想（完整版）

本文整理Python后端（以FastAPI为主，适配Flask）最标准、最通用、企业级的分包架构，包含所有核心分层（schema、controller、service、crud等），明确各层职责、调用关系和目录规范，可直接作为项目搭建模板和学习参考。

一、完整分包体系（核心分层）

Python后端标准分包共8个核心模块，各司其职、层层解耦，完整结构如下（按调用优先级排序）：

 app/
 ├── main.py                 # 项目入口（启动服务、注册路由）
 ├── database.py              # 数据库连接配置（引擎、会话、连接池）
 ├── settings.py              # 全局项目配置（环境变量、端口、密钥等）
 ├── schemas/                 # 数据模型层（请求/响应参数、数据校验）
 ├── controllers/             # 接口路由层（接收请求、返回结果）
 ├── services/                # 业务逻辑层（核心业务规则、流程控制）
 ├── crud/                    # 数据库操作层（基础增删改查）
 ├── models/                  # 数据库表模型层（ORM映射、表结构定义）
 ├── utils/                   # 工具函数层（通用工具、与业务无关）
 └── common/                  # 公共模块层（全局常量、异常、公共配置）

二、各层详细职责（人话版，必懂）

1. schemas（数据模型层）

核心职责：只定义数据结构，负责数据的“进”（请求参数）和“出”（响应结果），以及数据校验。

核心内容：

• 基于Pydantic定义模型（请求模型、响应模型）

• 字段限制（如必填、长度、类型、枚举）

• 数据序列化/反序列化（与数据库模型、接口参数映射）

• 请求参数校验（避免非法数据进入业务层）

禁忌：不写业务逻辑、不直接操作数据库、不处理接口请求。

# schemas/pet.py
from pydantic import BaseModel, Field, validator
from datetime import date
from typing import Optional
from enum import Enum

# 枚举类（也可以放在common，这里演示）
class PetStatusEnum(str, Enum):
    available = "available"
    adopted = "adopted"
    pending = "pending"

# 请求模型：创建宠物
class PetCreateRequest(BaseModel):
    name: str = Field(..., min_length=1, max_length=50, description="宠物名字")
    species: str = Field(..., description="品种，如：狗、猫")
    age_months: int = Field(..., ge=0, le=360, description="年龄（月）")
    price: float = Field(..., ge=0, description="领养价格")
    description: Optional[str] = Field(None, max_length=500)
    status: PetStatusEnum = PetStatusEnum.available

    @validator('name')
    def validate_name(cls, v):
        if not v.strip():
            raise ValueError('宠物名字不能为空')
        return v.strip()

# 请求模型：查询宠物列表
class PetQueryParams(BaseModel):
    species: Optional[str] = None
    min_age: Optional[int] = Field(None, ge=0)
    max_age: Optional[int] = Field(None, le=360)
    status: Optional[PetStatusEnum] = None
    page: int = Field(1, ge=1)
    page_size: int = Field(10, ge=1, le=100)

# 响应模型：宠物信息
class PetResponse(BaseModel):
    id: int
    name: str
    species: str
    age_months: int
    price: float
    description: Optional[str]
    status: str
    created_at: str
    updated_at: str

    class Config:
        # 允许从ORM对象转换
        from_attributes = True

# 通用分页响应模型
class PageResponse(BaseModel):
    total: int
    page: int
    page_size: int
    items: list

2. controllers（接口路由层，别名：router/api）

核心职责：作为前端与后端的桥梁，只负责“接请求、收参数、返结果”。

核心内容：

• 定义接口路由

• 接收前端传递的参数（依赖schemas模型做校验）

• 调用service层的业务方法，不直接操作数据库

• 统一返回格式（成功/失败响应体）、捕获接口异常

禁忌：不写复杂业务逻辑、不直接写SQL/ORM语句、不处理数据校验。

# controllers/pet_controller.py
from fastapi import APIRouter, Depends, HTTPException, Query
from typing import List
from sqlalchemy.orm import Session

from database import get_db
from schemas.pet import (
    PetCreateRequest, PetResponse, PetQueryParams, PageResponse
)
from services.pet_service import PetService
from common.response import success_response, error_response
from common.exceptions import BusinessException

# 创建路由器
router = APIRouter(prefix="/api/pets", tags=["宠物管理"])

@router.post("/", response_model=dict, summary="创建宠物")
async def create_pet(
    pet_data: PetCreateRequest,  # 自动校验
    db: Session = Depends(get_db)
):
    """创建新宠物（仅管理员）"""
    try:
        pet_service = PetService(db)
        pet = pet_service.create_pet(pet_data.dict())
        return success_response(data=pet, message="创建成功")
    except BusinessException as e:
        return error_response(code=e.code, message=str(e))
    except Exception as e:
        return error_response(message=f"创建失败：{str(e)}")

@router.get("/", response_model=dict, summary="获取宠物列表")
async def get_pet_list(
    species: Optional[str] = Query(None, description="品种"),
    min_age: Optional[int] = Query(None, ge=0),
    max_age: Optional[int] = Query(None, le=360),
    status: Optional[str] = None,
    page: int = Query(1, ge=1),
    page_size: int = Query(10, ge=1, le=100),
    db: Session = Depends(get_db)
):
    """分页查询宠物列表"""
    try:
        # 组装查询参数
        params = PetQueryParams(
            species=species,
            min_age=min_age,
            max_age=max_age,
            status=status,
            page=page,
            page_size=page_size
        )
        pet_service = PetService(db)
        result = pet_service.get_pet_list(params)
        return success_response(data=result)
    except Exception as e:
        return error_response(message=str(e))

@router.get("/{pet_id}", response_model=dict, summary="获取宠物详情")
async def get_pet_detail(
    pet_id: int,
    db: Session = Depends(get_db)
):
    """根据ID获取宠物详细信息"""
    pet_service = PetService(db)
    pet = pet_service.get_pet_by_id(pet_id)
    if not pet:
        return error_response(code=404, message="宠物不存在")
    return success_response(data=pet)

@router.put("/{pet_id}", summary="更新宠物信息")
async def update_pet(
    pet_id: int,
    pet_data: PetCreateRequest,
    db: Session = Depends(get_db)
):
    """更新宠物信息（仅管理员）"""
    pet_service = PetService(db)
    pet = pet_service.update_pet(pet_id, pet_data.dict())
    if not pet:
        return error_response(code=404, message="宠物不存在")
    return success_response(data=pet, message="更新成功")

@router.delete("/{pet_id}", summary="删除宠物")
async def delete_pet(
    pet_id: int,
    db: Session = Depends(get_db)
):
    """删除宠物（仅管理员）"""
    pet_service = PetService(db)
    success = pet_service.delete_pet(pet_id)
    if not success:
        return error_response(code=404, message="宠物不存在")
    return success_response(message="删除成功")

3. services（业务逻辑层）

核心职责：项目的核心，负责所有业务规则、流程控制和逻辑判断。

核心内容：

• 实现核心业务逻辑（如宠物领养判断、权限校验、流程串联）

• 组合多个crud操作（如“领养宠物”=查询宠物+查询用户+新增领养记录）

• 事务控制（如操作失败回滚数据）

• 调用utils工具函数、common公共常量

禁忌：不直接操作数据库原生语句、不接收前端请求、不定义数据结构。

# services/pet_service.py
from typing import Dict, List, Optional
from sqlalchemy.orm import Session
from datetime import datetime

from crud.pet_crud import PetCRUD
from services.adopt_service import AdoptService
from common.exceptions import BusinessException
from utils.logger import log_business

class PetService:
    """宠物业务逻辑层"""
    
    def __init__(self, db: Session):
        self.db = db
        self.pet_crud = PetCRUD(db)
    
    @log_business("创建宠物")
    def create_pet(self, pet_data: Dict) -> Dict:
        """创建宠物（业务规则：价格不能为负数，年龄不能太大）"""
        # 业务校验
        if pet_data.get('price', 0) < 0:
            raise BusinessException("宠物价格不能为负数")
        
        if pet_data.get('age_months', 0) > 360:
            raise BusinessException("宠物年龄不能超过30岁")
        
        # 调用数据库操作
        pet = self.pet_crud.create(pet_data)
        
        # 记录日志（业务操作）
        print(f"[业务日志] 创建宠物成功：{pet['name']}")
        
        return pet
    
    def get_pet_list(self, query_params) -> Dict:
        """获取宠物列表（支持过滤和分页）"""
        # 构建过滤条件
        filters = {}
        if query_params.species:
            filters['species'] = query_params.species
        if query_params.status:
            filters['status'] = query_params.status
        
        # 年龄范围过滤
        age_range = None
        if query_params.min_age or query_params.max_age:
            age_range = (query_params.min_age, query_params.max_age)
        
        # 查询数据
        pets, total = self.pet_crud.get_list(
            filters=filters,
            age_range=age_range,
            page=query_params.page,
            page_size=query_params.page_size
        )
        
        return {
            "total": total,
            "page": query_params.page,
            "page_size": query_params.page_size,
            "items": pets
        }
    
    def get_pet_by_id(self, pet_id: int) -> Optional[Dict]:
        """根据ID获取宠物详情"""
        return self.pet_crud.get_by_id(pet_id)
    
    @log_business("更新宠物")
    def update_pet(self, pet_id: int, update_data: Dict) -> Optional[Dict]:
        """更新宠物信息（业务规则：已领养的不能修改）"""
        # 先查询宠物
        pet = self.pet_crud.get_by_id(pet_id)
        if not pet:
            return None
        
        # 业务规则：已领养的宠物不能修改
        if pet['status'] == 'adopted':
            raise BusinessException("已领养的宠物无法修改信息")
        
        # 执行更新
        updated_pet = self.pet_crud.update(pet_id, update_data)
        
        return updated_pet
    
    @log_business("删除宠物")
    def delete_pet(self, pet_id: int) -> bool:
        """删除宠物（业务规则：已领养的不能删除）"""
        pet = self.pet_crud.get_by_id(pet_id)
        if not pet:
            return False
        
        if pet['status'] == 'adopted':
            raise BusinessException("已领养的宠物无法删除")
        
        return self.pet_crud.delete(pet_id)
    
    def adopt_pet(self, pet_id: int, user_id: int) -> Dict:
        """
        领养宠物（组合多个CRUD操作，事务控制）
        业务规则：
        1. 宠物必须存在且状态为available
        2. 用户必须存在且信用良好
        3. 创建领养记录
        4. 更新宠物状态为adopted
        """
        # 1. 检查宠物状态
        pet = self.pet_crud.get_by_id(pet_id)
        if not pet:
            raise BusinessException("宠物不存在")
        
        if pet['status'] != 'available':
            raise BusinessException("该宠物已被领养或不可领养")
        
        # 2. 调用用户服务检查用户（示例调用其他service）
        from services.user_service import UserService
        user_service = UserService(self.db)
        user = user_service.get_user_by_id(user_id)
        if not user:
            raise BusinessException("用户不存在")
        
        if user.get('credit_score', 0) < 60:
            raise BusinessException("用户信用分不足，无法领养")
        
        # 3. 创建领养记录
        adopt_service = AdoptService(self.db)
        adopt_record = adopt_service.create_adopt_record({
            'pet_id': pet_id,
            'user_id': user_id,
            'status': 'pending',
            'created_at': datetime.now()
        })
        
        # 4. 更新宠物状态（事务：如果失败要回滚）
        try:
            # 开启事务（由数据库会话管理）
            self.pet_crud.update(pet_id, {'status': 'adopted'})
            self.db.commit()
        except Exception as e:
            self.db.rollback()
            raise BusinessException(f"领养失败：{str(e)}")
        
        return {
            'adopt_id': adopt_record['id'],
            'pet': pet,
            'user': user
        }

4. crud（数据库操作层）

核心职责：只负责最基础的数据库增删改查，提供“原子化”数据库操作能力。

核心内容：

• 封装单表操作（查一条、查列表、新增、修改、删除）

• 基于SQLAlchemy ORM编写基础操作（不写复杂联表查询，复杂查询可在service层组合）

• 只关注数据库操作，不关心业务逻辑

禁忌：不写业务判断、不处理接口请求、不定义数据校验规则。

# crud/pet_crud.py
from typing import Dict, List, Optional, Tuple
from sqlalchemy.orm import Session
from sqlalchemy import and_, or_, desc, asc
from models.pet import PetModel

class PetCRUD:
    """宠物数据库操作层 - 只做原子化CRUD"""
    
    def __init__(self, db: Session):
        self.db = db
    
    def create(self, data: Dict) -> Dict:
        """创建单条记录"""
        pet = PetModel(**data)
        self.db.add(pet)
        self.db.commit()
        self.db.refresh(pet)
        return self._to_dict(pet)
    
    def get_by_id(self, pet_id: int) -> Optional[Dict]:
        """根据ID查询"""
        pet = self.db.query(PetModel).filter(PetModel.id == pet_id).first()
        return self._to_dict(pet) if pet else None
    
    def get_list(
        self,
        filters: Dict = None,
        age_range: Tuple[int, int] = None,
        page: int = 1,
        page_size: int = 10,
        order_by: str = "-id"
    ) -> Tuple[List[Dict], int]:
        """
        查询列表（支持过滤、分页、排序）
        返回：(数据列表, 总数量)
        """
        query = self.db.query(PetModel)
        
        # 应用过滤条件
        if filters:
            for key, value in filters.items():
                if hasattr(PetModel, key) and value is not None:
                    query = query.filter(getattr(PetModel, key) == value)
        
        # 年龄范围过滤
        if age_range:
            min_age, max_age = age_range
            if min_age is not None:
                query = query.filter(PetModel.age_months >= min_age)
            if max_age is not None:
                query = query.filter(PetModel.age_months <= max_age)
        
        # 排序（处理 -id 表示降序）
        if order_by.startswith("-"):
            order_field = getattr(PetModel, order_by[1:])
            query = query.order_by(desc(order_field))
        else:
            order_field = getattr(PetModel, order_by)
            query = query.order_by(asc(order_field))
        
        # 获取总数
        total = query.count()
        
        # 分页
        query = query.offset((page - 1) * page_size).limit(page_size)
        pets = query.all()
        
        return [self._to_dict(pet) for pet in pets], total
    
    def update(self, pet_id: int, update_data: Dict) -> Optional[Dict]:
        """更新记录"""
        pet = self.db.query(PetModel).filter(PetModel.id == pet_id).first()
        if not pet:
            return None
        
        for key, value in update_data.items():
            if hasattr(pet, key) and value is not None:
                setattr(pet, key, value)
        
        self.db.commit()
        self.db.refresh(pet)
        return self._to_dict(pet)
    
    def delete(self, pet_id: int) -> bool:
        """删除记录"""
        pet = self.db.query(PetModel).filter(PetModel.id == pet_id).first()
        if not pet:
            return False
        
        self.db.delete(pet)
        self.db.commit()
        return True
    
    def update_status(self, pet_id: int, status: str) -> bool:
        """单独更新状态（原子操作）"""
        pet = self.db.query(PetModel).filter(PetModel.id == pet_id).first()
        if not pet:
            return False
        pet.status = status
        self.db.commit()
        return True
    
    def _to_dict(self, pet: PetModel) -> Dict:
        """ORM对象转字典"""
        return {
            'id': pet.id,
            'name': pet.name,
            'species': pet.species,
            'age_months': pet.age_months,
            'price': float(pet.price) if pet.price else 0,
            'description': pet.description,
            'status': pet.status,
            'created_at': pet.created_at.strftime('%Y-%m-%d %H:%M:%S') if pet.created_at else None,
            'updated_at': pet.updated_at.strftime('%Y-%m-%d %H:%M:%S') if pet.updated_at else None
        }

5. models（数据库表模型层）

核心职责：定义数据库表结构，实现ORM映射（将Python类映射到数据库表）。

核心内容：

• 定义表字段（字段名、类型、主键、外键、默认值）

• 关联表关系（一对一、一对多、多对多）

• 与schemas模型做映射（数据库数据 ↔ 接口数据）

禁忌：不写业务逻辑、不处理接口请求、不包含工具方法。

# models/pet.py
from sqlalchemy import Column, Integer, String, Float, DateTime, Text, Enum as SQLEnum
from sqlalchemy.sql import func
from database import Base
import enum


# 宠物状态枚举
class PetStatus(str, enum.Enum):
    AVAILABLE = "available"
    ADOPTED = "adopted"
    PENDING = "pending"

class PetModel(Base):
    """宠物表模型"""
    __tablename__ = "pets"
    
    id = Column(Integer, primary_key=True, index=True, autoincrement=True)
    name = Column(String(50), nullable=False, comment="宠物名字")
    species = Column(String(50), nullable=False, comment="品种")
    age_months = Column(Integer, nullable=False, default=0, comment="年龄（月）")
    price = Column(Float, nullable=False, default=0, comment="领养价格")
    description = Column(Text, nullable=True, comment="描述")
    status = Column(
        SQLEnum(PetStatus),
        nullable=False,
        default=PetStatus.AVAILABLE,
        comment="状态：available/adopted/pending"
    )
    
    # 审计字段
    created_at = Column(DateTime, server_default=func.now(), comment="创建时间")
    updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now(), comment="更新时间")
    
    # 外键关联示例（一对多）
    # user_id = Column(Integer, ForeignKey("users.id"))
    
    def __repr__(self):
        return f"<Pet(id={self.id}, name={self.name}, status={self.status})>"

# models/user.py（用户表示例）
from sqlalchemy import Column, Integer, String, DateTime, Boolean
from sqlalchemy.sql import func
from database import Base

class UserModel(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String(50), unique=True, nullable=False, index=True)
    email = Column(String(100), unique=True, nullable=False)
    hashed_password = Column(String(200), nullable=False)
    credit_score = Column(Integer, default=100)  # 信用分
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, server_default=func.now())

6. utils（工具函数层，全名：utilities）

核心职责：存放纯工具、纯函数，与项目业务无关，可复用性极强（换个项目也能直接用）。

核心内容：

• 密码加密/解密（如bcrypt、md5）

• Token生成/解析（如JWT）

• 时间格式化（如datetime转字符串、时区转换）

• 文件处理（上传、下载、压缩、格式转换）

• 字符串处理（截取、替换、校验）

• 第三方服务调用（发送短信、邮件、图片上传）

• 自定义工具函数（如随机数生成、数据格式转换）

关键特点：无状态、纯函数、不依赖项目业务逻辑。

6.1 JWT工具

# utils/jwt_utils.py - JWT工具
import jwt
from datetime import datetime, timedelta
from typing import Dict
from settings import settings

def create_access_token(data: Dict, expires_delta: timedelta = None) -> str:
    """生成JWT token"""
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(
        to_encode,
        settings.JWT_SECRET_KEY,
        algorithm=settings.JWT_ALGORITHM
    )
    return encoded_jwt

def decode_token(token: str) -> Dict:
    """解析JWT token"""
    try:
        payload = jwt.decode(
            token,
            settings.JWT_SECRET_KEY,
            algorithms=[settings.JWT_ALGORITHM]
        )
        return payload
    except jwt.ExpiredSignatureError:
        raise Exception("Token已过期")
    except jwt.InvalidTokenError:
        raise Exception("无效的Token")

6.2 密码加密工具

# utils/password_utils.py - 密码加密工具
import bcrypt
import hashlib

def hash_password(password: str) -> str:
    """加密密码（bcrypt）"""
    salt = bcrypt.gensalt()
    hashed = bcrypt.hashpw(password.encode('utf-8'), salt)
    return hashed.decode('utf-8')

def verify_password(plain_password: str, hashed_password: str) -> bool:
    """验证密码"""
    return bcrypt.checkpw(
        plain_password.encode('utf-8'),
        hashed_password.encode('utf-8')
    )

def md5_hash(text: str) -> str:
    """MD5哈希（用于非安全场景）"""
    return hashlib.md5(text.encode('utf-8')).hexdigest()

6.3 时间处理工具

# utils/date_utils.py - 时间处理工具
from datetime import datetime, timedelta
from typing import Optional

def format_datetime(dt: datetime, format_str: str = "%Y-%m-%d %H:%M:%S") -> str:
    """格式化时间"""
    if not dt:
        return ""
    return dt.strftime(format_str)

def parse_datetime(date_str: str, format_str: str = "%Y-%m-%d %H:%M:%S") -> Optional[datetime]:
    """解析时间字符串"""
    try:
        return datetime.strptime(date_str, format_str)
    except ValueError:
        return None

def get_age_from_birthday(birthday: datetime) -> int:
    """根据生日计算年龄"""
    today = datetime.now()
    return today.year - birthday.year - ((today.month, today.day) < (birthday.month, birthday.day))

6.4 日志装饰器

# utils/logger.py - 日志装饰器
import functools
from datetime import datetime

def log_business(operation: str):
    """业务日志装饰器"""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            print(f"[{datetime.now()}] 开始执行业务：{operation}")
            try:
                result = func(*args, **kwargs)
                print(f"[{datetime.now()}] 业务执行成功：{operation}")
                return result
            except Exception as e:
                print(f"[{datetime.now()}] 业务执行失败：{operation}, 错误：{str(e)}")
                raise
        return wrapper
    return decorator