7. common（公共模块层，全名：common/core）

核心职责：存放项目全局共用的配置、常量、异常、类，贯穿整个项目。

核心内容：

• 全局常量（如状态码、宠物领养状态、权限标识）

• 枚举类（如GenderEnum、AdoptStatusEnum、ErrorCodeEnum）

• 公共异常类（如CustomException、ParamsError、BusinessError）

• 公共返回格式（统一响应体，如{"code":200,"msg":"success","data":{}}）

• 全局中间件（如日志中间件、跨域中间件、权限中间件）

• 全局依赖（如登录校验依赖、数据库会话依赖）

关键特点：项目内全局复用，与业务相关但不包含具体业务逻辑。

7.1 统一响应格式

# common/response.py - 统一响应格式
from typing import Any, Optional

def success_response(
    data: Any = None,
    message: str = "success",
    code: int = 200
) -> dict:
    """成功响应"""
    return {
        "code": code,
        "message": message,
        "data": data
    }

def error_response(
    message: str = "error",
    code: int = 400,
    data: Any = None
) -> dict:
    """失败响应"""
    return {
        "code": code,
        "message": message,
        "data": data
    }

# 通用分页响应包装
def page_response(items: list, total: int, page: int, page_size: int) -> dict:
    return {
        "items": items,
        "total": total,
        "page": page,
        "page_size": page_size,
        "total_pages": (total + page_size - 1) // page_size
    }

7.2 自定义异常

# common/exceptions.py - 自定义异常
class BusinessException(Exception):
    """业务异常"""
    def __init__(self, message: str, code: int = 400):
        self.code = code
        self.message = message
        super().__init__(self.message)

class NotFoundException(BusinessException):
    """资源不存在异常"""
    def __init__(self, message: str = "资源不存在"):
        super().__init__(message, code=404)

class PermissionDeniedException(BusinessException):
    """权限不足异常"""
    def __init__(self, message: str = "权限不足"):
        super().__init__(message, code=403)

class ValidationException(BusinessException):
    """数据校验异常"""
    def __init__(self, message: str = "数据校验失败"):
        super().__init__(message, code=422)

7.3 全局常量

# common/constants.py - 全局常量
# 状态码常量
class StatusCode:
    SUCCESS = 200
    BAD_REQUEST = 400
    UNAUTHORIZED = 401
    FORBIDDEN = 403
    NOT_FOUND = 404
    INTERNAL_ERROR = 500

# 业务常量
ADOPT_STATUS = {
    "pending": "待审核",
    "approved": "已批准",
    "rejected": "已拒绝",
    "completed": "已完成"
}

PET_SPECIES = ["狗", "猫", "兔子", "仓鼠", "鸟"]

# 权限标识
PERMISSIONS = {
    "admin": ["pet:create", "pet:update", "pet:delete", "user:list"],
    "user": ["pet:list", "pet:view", "adopt:create"]
}

7.4 枚举类

# common/enums.py - 枚举类
from enum import Enum

class GenderEnum(str, Enum):
    MALE = "male"
    FEMALE = "female"
    OTHER = "other"

class AdoptStatusEnum(str, Enum):
    PENDING = "pending"
    APPROVED = "approved"
    REJECTED = "rejected"
    COMPLETED = "completed"

class RoleEnum(str, Enum):
    ADMIN = "admin"
    USER = "user"
    GUEST = "guest"

7.5 全局yi

# common/dependencies.py - 全局依赖
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from sqlalchemy.orm import Session

from database import get_db
from utils.jwt_utils import decode_token
from common.exceptions import PermissionDeniedException

security = HTTPBearer()

async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security),
    db: Session = Depends(get_db)
):
    """获取当前登录用户（依赖注入）"""
    token = credentials.credentials
    try:
        payload = decode_token(token)
        user_id = payload.get("user_id")
        if not user_id:
            raise HTTPException(status_code=401, detail="无效的token")
        
        # 从数据库查询用户（避免重复查询可以缓存）
        from crud.user_crud import UserCRUD
        user_crud = UserCRUD(db)
        user = user_crud.get_by_id(user_id)
        
        if not user:
            raise HTTPException(status_code=401, detail="用户不存在")
        
        return user
    except Exception as e:
        raise HTTPException(status_code=401, detail=str(e))

def require_admin(current_user: dict = Depends(get_current_user)):
    """要求管理员权限"""
    if current_user.get('role') != 'admin':
        raise PermissionDeniedException("需要管理员权限")
    return current_user

8. database.py（数据库连接层）

核心职责：负责数据库的连接配置，提供数据库会话。

核心内容：

• 数据库引擎配置（连接地址、用户名、密码、端口）

• 数据库会话创建（SessionLocal）

• 数据库迁移相关配置（如alembic关联）

• # database.py
  from sqlalchemy import create_engine
  from sqlalchemy.ext.declarative import declarative_base
  from sqlalchemy.orm import sessionmaker, Session
  from typing import Generator
  
  from settings import settings
  
  # 创建数据库引擎
  # SQLite示例（开发环境）
  # engine = create_engine(
  #     "sqlite:///./pet_adopt.db",
  #     connect_args={"check_same_thread": False},
  #     echo=True
  # )
  
  # MySQL示例（生产环境）
  engine = create_engine(
      settings.DATABASE_URL,
      pool_pre_ping=True,  # 连接前检查是否有效
      pool_recycle=3600,   # 1小时回收连接
      echo=settings.DEBUG   # 开发环境打印SQL
  )
  
  # 创建会话工厂
  SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
  
  # 创建基类（供models继承）
  Base = declarative_base()
  
  def get_db() -> Generator[Session, None, None]:
      """
      获取数据库会话（依赖注入用）
      使用方式：db: Session = Depends(get_db)
      """
      db = SessionLocal()
      try:
          yield db
      finally:
          db.close()
  
  # 初始化数据库（创建所有表）
  def init_db():
      Base.metadata.create_all(bind=engine)

9. settings.py（全局配置层）

核心职责：存放项目全局配置，统一管理环境变量、服务配置。

核心内容：

• 服务配置（端口、主机、调试模式）

• 环境变量（开发/测试/生产环境区分）

• 第三方服务配置（如短信API密钥、OSS配置）

• 密钥配置（如JWT密钥、加密密钥）

# settings.py
from pydantic_settings import BaseSettings
from typing import Optional
import os
from dotenv import load_dotenv

load_dotenv()  # 加载.env文件

class Settings(BaseSettings):
    """应用配置"""
    
    # 基础配置
    APP_NAME: str = "宠物领养系统"
    DEBUG: bool = os.getenv("DEBUG", "True").lower() == "true"
    HOST: str = os.getenv("HOST", "0.0.0.0")
    PORT: int = int(os.getenv("PORT", "8000"))
    
    # 数据库配置
    # SQLite
    # DATABASE_URL: str = "sqlite:///./pet_adopt.db"
    
    # MySQL
    DB_HOST: str = os.getenv("DB_HOST", "localhost")
    DB_PORT: int = int(os.getenv("DB_PORT", "3306"))
    DB_USER: str = os.getenv("DB_USER", "root")
    DB_PASSWORD: str = os.getenv("DB_PASSWORD", "123456")
    DB_NAME: str = os.getenv("DB_NAME", "pet_adopt")
    
    @property
    def DATABASE_URL(self) -> str:
        """动态构建数据库连接字符串"""
        return f"mysql+pymysql://{self.DB_USER}:{self.DB_PASSWORD}@{self.DB_HOST}:{self.DB_PORT}/{self.DB_NAME}?charset=utf8mb4"
    
    # JWT配置
    JWT_SECRET_KEY: str = os.getenv("JWT_SECRET_KEY", "your-secret-key-change-in-production")
    JWT_ALGORITHM: str = "HS256"
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
    REFRESH_TOKEN_EXPIRE_DAYS: int = 7
    
    # 第三方服务配置
    SMS_API_KEY: Optional[str] = os.getenv("SMS_API_KEY")
    SMS_API_SECRET: Optional[str] = os.getenv("SMS_API_SECRET")
    
    OSS_ACCESS_KEY: Optional[str] = os.getenv("OSS_ACCESS_KEY")
    OSS_SECRET_KEY: Optional[str] = os.getenv("OSS_SECRET_KEY")
    OSS_BUCKET_NAME: Optional[str] = os.getenv("OSS_BUCKET_NAME")
    OSS_ENDPOINT: Optional[str] = os.getenv("OSS_ENDPOINT")
    
    # Redis配置
    REDIS_HOST: str = os.getenv("REDIS_HOST", "localhost")
    REDIS_PORT: int = int(os.getenv("REDIS_PORT", "6379"))
    REDIS_DB: int = int(os.getenv("REDIS_DB", "0"))
    REDIS_PASSWORD: Optional[str] = os.getenv("REDIS_PASSWORD")
    
    class Config:
        env_file = ".env"
        case_sensitive = True

# 创建全局配置实例
settings = Settings()

.env文件示例

# .env
DEBUG=True
HOST=0.0.0.0
PORT=8000

DB_HOST=localhost
DB_PORT=3306
DB_USER=root
DB_PASSWORD=123456
DB_NAME=pet_adopt

JWT_SECRET_KEY=your-super-secret-key-please-change
SMS_API_KEY=your-sms-api-key

10. main.py（项目入口）

核心职责：启动FastAPI服务，注册路由、中间件、依赖等。

核心内容：

• 创建FastAPI实例

• 注册controllers层的路由

• 加载中间件、全局依赖

• 启动服务（if __name__ == "__main__";: uvicorn.run()）

# main.py
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.trustedhost import TrustedHostMiddleware
import uvicorn
from datetime import datetime

from settings import settings
from database import init_db, get_db
from common.response import error_response
from common.exceptions import BusinessException

# 导入所有路由
from controllers.pet_controller import router as pet_router
# from controllers.user_controller import router as user_router
# from controllers.adopt_controller import router as adopt_router

# 创建FastAPI应用实例
app = FastAPI(
    title=settings.APP_NAME,
    description="宠物领养系统API文档",
    version="1.0.0",
    docs_url="/docs" if settings.DEBUG else None,  # 生产环境可关闭
    redoc_url="/redoc" if settings.DEBUG else None
)

# ========== 配置中间件 ==========

# 1. CORS跨域中间件（允许前端调用）
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"] if settings.DEBUG else ["https://your-domain.com"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 2. 信任主机中间件（安全）
if not settings.DEBUG:
    app.add_middleware(
        TrustedHostMiddleware,
        allowed_hosts=["your-domain.com", "api.your-domain.com"]
    )

# 3. 自定义请求日志中间件
@app.middleware("http")
async def log_requests(request: Request, call_next):
    """记录所有请求日志"""
    start_time = datetime.now()
    
    # 记录请求信息
    print(f"[{start_time}] {request.method} {request.url.path}")
    
    # 处理请求
    response = await call_next(request)
    
    # 计算耗时
    process_time = (datetime.now() - start_time).total_seconds()
    response.headers["X-Process-Time"] = str(process_time)
    
    print(f"[{datetime.now()}] 完成 {request.method} {request.url.path} - 耗时: {process_time:.3f}s")
    
    return response

# ========== 全局异常处理 ==========

@app.exception_handler(BusinessException)
async def business_exception_handler(request: Request, exc: BusinessException):
    """业务异常统一处理"""
    return error_response(
        message=exc.message,
        code=exc.code
    )

@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    """全局异常兜底"""
    print(f"未处理的异常: {exc}")
    return error_response(
        message="服务器内部错误",
        code=500
    )

# ========== 注册路由 ==========

app.include_router(pet_router)
# app.include_router(user_router, prefix="/api")
# app.include_router(adopt_router, prefix="/api")

# ========== 健康检查接口 ==========

@app.get("/health", tags=["系统"])
async def health_check():
    """健康检查接口"""
    return {
        "status": "healthy",
        "service": settings.APP_NAME,
        "timestamp": datetime.now().isoformat()
    }

@app.get("/", tags=["系统"])
async def root():
    """根路径"""
    return {
        "message": f"欢迎使用{settings.APP_NAME}API",
        "docs": "/docs" if settings.DEBUG else None
    }

# ========== 启动事件 ==========

@app.on_event("startup")
async def startup_event():
    """应用启动时执行"""
    print(f"正在启动{settings.APP_NAME}...")
    # 初始化数据库表
    init_db()
    print(f"数据库连接成功")
    print(f"API文档地址: http://{settings.HOST}:{settings.PORT}/docs")
    print(f"服务已启动在: http://{settings.HOST}:{settings.PORT}")

@app.on_event("shutdown")
async def shutdown_event():
    """应用关闭时执行"""
    print(f"{settings.APP_NAME}正在关闭...")

# ========== 启动服务 ==========

if __name__ == "__main__":
    uvicorn.run(
        "main:app",
        host=settings.HOST,
        port=settings.PORT,
        reload=settings.DEBUG,  # 开发模式自动重载
        log_level="info"
    )

三、核心调用关系（清晰易懂）

调用流程：前端 → 接口层 → 业务层 → 数据库操作层 → 数据库，schema、utils、common贯穿所有层。

 前端请求
   ↓
 controllers（接收请求、校验参数）
   ↓ 调用
 service（业务逻辑判断、流程控制）
   ↓ 调用
 crud（基础数据库操作）
   ↓ 操作
 models（表模型）
   ↓ 连接
 database（数据库连接）
 ​
 【辅助支撑】
 schema：给controllers、service、crud提供数据结构和校验
 utils：给所有层提供工具函数（如service调用密码加密、token生成）
 common：给所有层提供常量、异常、公共返回（如service抛出common定义的异常）

四、分包核心思想（为什么要这么分）

1. 解耦：各层职责单一，修改某一层不影响其他层（如改数据库操作只动crud，改业务只动service）。

2. 复用：utils工具函数、common公共配置可全局复用；crud、service可被多个controllers调用。

3. 易维护：目录结构清晰，新人可快速定位“接口在哪、业务在哪、数据库操作在哪”。

4. 易测试：每层独立，可单独做单元测试（如测试service不依赖controller，测试crud不依赖service）。

5. 可扩展：后期新增功能（如新增“宠物寄养”接口），只需新增对应schema、controller、service、crud，不影响原有代码。

五、新手必踩坑（禁忌清单）

• ❌ 不分层：在controllers里直接写SQL、写业务逻辑（最常见错误）。

• ❌ 跳过service层：controllers直接调用crud（简单小项目可临时用，复杂项目必规范）。

• ❌ crud里写业务判断：如在crud的delete方法里判断“是否允许删除”（应放在service层）。

• ❌ utils里写业务逻辑：如在utils里写“宠物领养校验”（utils应与业务无关）。

• ❌ schema里写数据库操作：如在Pydantic模型里写查询方法（schema只负责数据结构）。

六、总结

Python后端标准分包思想的核心：每层只做自己分内的事，职责单一、层层解耦、全局复用。

记住一句话：schema管数据，controller管接口，service管业务，crud管数据库，utils管工具，common管公共，剩下的配置和入口负责串联整个项目。

这套架构适配所有Python后端项目（FastAPI、Flask），无论是小项目还是企业级项目，都能直接套用，是最规范、最通用的分包方式。

pet-adopt-system/
├── schemas/           # 数据模型层
│   ├── pet.py
│   └── user.py
├── controllers/       # 接口路由层
│   ├── pet_controller.py
│   └── user_controller.py
├── services/          # 业务逻辑层
│   ├── pet_service.py
│   └── user_service.py
├── crud/              # 数据库操作层
│   ├── pet_crud.py
│   └── user_crud.py
├── models/            # 数据库表模型
│   ├── pet.py
│   └── user.py
├── utils/             # 工具函数层
│   ├── jwt_utils.py
│   ├── password_utils.py
│   ├── date_utils.py
│   └── logger.py
├── common/            # 公共模块层
│   ├── response.py
│   ├── exceptions.py
│   ├── constants.py
│   ├── enums.py
│   └── dependencies.py
├── database.py        # 数据库连接
├── settings.py        # 全局配置
├── main.py            # 项目入口
├── requirements.txt   # 依赖包
└── .env              # 环境变量