Python FastAPI+SQLAlchemy2.0 企业级目录结构

作者：马育民 • • 阅读：10066

介绍

企业级 FastAPI+SQLAlchemy2.0 开发核心要求：模块化、可维护、可扩展、符合生产规范、支持团队协作。我给你一套工业级标准架构（最通用、大厂最常用），直接可落地。

目录结构

your_project/
├── .env                # 环境变量配置
├── .env.example
├── .gitignore
├── data/               # 存放SQLite db、上传文件、临时数据
│   └── db.sqlite3
├── sql/                # 存放初始化SQL、种子SQL、升级SQL
│   ├── init.sql
│   └── seed.sql
├── alembic/              # 数据库迁移
├── app/
│   ├── api/              # 控制器层 Controller
│   │   ├── v1/           # 接口版本 v1
│   │   ├── deps.py       # 全局依赖（DB、登录、权限）
│   │   └── api.py        # 总路由入口
│   ├── core/             # 核心配置
│   │   ├── config.py     # 项目配置（.env）
│   │   ├── security.py   # JWT、密码加密
│   │   ├── exceptions.py # 全局异常处理
│   │   └── response.py   # 统一返回格式
│   ├── db/               # 数据库连接
│   │   ├── base.py       # ORM 基类
│   │   └── session.py    # DB 会话
│   ├── models/           # 数据库实体 Entity
│   ├── schemas/          # 请求/响应 DTO
│   ├── services/         # 业务逻辑 Service
│   ├── crud/             # 数据访问 DAO
│   └── main.py           # 项目入口
├── requirements.txt      # 依赖
└── run.py                # 启动脚本

为什么创建 app 子目录

为了：把【业务代码】和【项目配置/文件】彻底分开！**

  • 工程根目录 = 只放配置、数据库、上传文件、工具
  • app/ = 只放代码

创建 app 目录的目的

1. 代码 vs 数据 分离（最重要！）
  • app/：纯代码（接口、模型、逻辑、工具）
  • 根目录：数据/配置（数据库、上传文件、.env、pyproject.toml）

这样做的好处：

  • 备份代码时，不用带上 GB 级的数据库和上传文件
  • 不会把业务文件不小心提交到 Git
  • 代码干净，不污染
2. 让项目变成「Python 包」

创建 app/ + __init__.py
你的项目就变成了一个标准 Python 包：

from app.config import settings
from.app.api.upload import router

导入清晰、不乱、不报错
这是大型项目必须的。

3. 方便内部模块化
app/
  ├── api/        # 接口
  ├── models/     # 数据库模型
  ├── core/       # 核心逻辑
  ├── utils/      # 工具

不用翻来翻去，结构一目了然。

4. 为未来扩展做准备

为扩展准备： 让当前项目本身能无限变大！

对比

❌ 不建 app/（混乱）
your-project/
  main.py
  config.py
  database.py
  upload/
  db.sqlite3
  .env
  pyproject.toml

文件堆在一起，大了必乱。

✅ 建 app/（标准、干净、可扩展）
your-project/
  # 根目录：配置 + 数据 + 环境
  db.sqlite3
  upload/
  .env
  pyproject.toml

  # app/：只放代码
  app/
    main.py
    config.py
    api/
    models/
    core/

建议

  1. 练手、小项目：不要建 app 文件夹，直接根目录平铺。
  2. 写正式项目、找工作：一定要用 app 分层结构。
  3. 数据库文件永远放data/，不要丢根目录。
  4. .env永远放根目录，不要塞进app里面。
  5. 路由不写业务、业务不写SQL，严格分层。

分层架构

和 Java SSM / SpringBoot 完全对应：

Python 层 Java 对应 职责
api/ Controller 接收请求、参数校验、返回响应
schemas/ DTO/VO 请求体、响应体格式
services/ Service 业务逻辑、事务、规则
crud/ DAO/Mapper 数据库增删改查
models/ Entity 数据库表结构
db/ 数据源配置 数据库连接
core/ 配置/工具 全局配置、异常、安全

请求流程（标准）

请求 → API 路由 → 依赖验证 → Service 业务 → CRUD 数据库 → 返回响应

绝对不允许：

  • API 里写业务逻辑
  • Service 里写原生 SQL
  • Controller 直接操作数据库

命名规范

目录命名

  1. 全部小写
  2. 单词之间不用分隔符（api / core / services）
  3. 复数优先：models / schemas / services / crud
  4. 版本号固定：v1 / v2 / v3（小写）

文件命名规范

通用规则

  • 全部小写
  • 多单词用下划线 _ 分隔
  • 禁止空格、横杠、中文、大写字母

api/ 控制器层

api/
├── v1/
│   ├── __init__.py
│   ├── user_api.py      # 用户接口
│   ├── auth_api.py      # 认证接口
│   ├── order_api.py     # 订单接口
│   └── admin_api.py     # 管理后台接口

命名：业务名_api.py

接口 URL 命名规范（RESTful 企业标准）

  • 小写
  • 复数资源
  • 下划线分隔
  • 版本号在前
/api/v1/users
/api/v1/users/{user_id}
/api/v1/orders
/api/v1/auth/login
/api/v1/auth/logout

统一返回格式（企业标准）

{
  "code": 200,    // 业务状态码
  "msg": "成功",   // 消息
  "data": {}      // 数据
}

services/ 业务层

services/
├── __init__.py
├── user_service.py
├── auth_service.py
└── order_service.py

命名：业务名_service.py

crud/ 数据访问层

crud/
├── __init__.py
├── user_crud.py
├── order_crud.py
└── product_crud.py

命名：业务名_crud.py

models/ 模型层

models/
├── __init__.py
├── user.py
├── role.py
├── order.py
└── product.py

命名：单数、小写、下划线：user.py / order.py

core/ 核心层

core/
├── config.py       # 固定
├── security.py     # 固定
├── exceptions.py   # 固定
└── response.py     # 固定

db/ 数据库层

db/
├── base.py      # 固定：ORM 基类
└── session.py   # 固定：数据库会话

schemas/ DTO 层

schemas/
├── __init__.py
├── user_schema.py
├── auth_schema.py
└── order_schema.py

命名：业务名_schema.py

类命名规范

1. ORM 模型（models）

  • 大驼峰（UpperCamelCase）
  • 单数
  • 名词
# models/user.py
class User(Base):
    __tablename__ = "users"  # 表名：复数、小写、下划线

class Order(Base):
    __tablename__ = "orders"

2. Pydantic Schema（schemas）

# 用户创建请求 DTO
class UserCreate(BaseModel):
    username: str
    password: str

# 用户响应 DTO
class UserResp(BaseModel):
    id: int
    username: str

规则：

  • 请求：UserCreate / OrderCreate
  • 响应：UserResp / OrderResp
  • 更新：UserUpdate

3. Service 业务类

class UserService:
    def get_user_by_id(self, user_id: int):
        ...

命名：业务名 + Service

4. CRUD 数据访问类

class UserCRUD:
    def get(self, db: Session, id: int):
        ...

命名：业务名 + CRUD

函数/方法命名

通用规则

  • 小写 + 下划线
  • 动词开头
  • 见名知意

1. Controller 接口函数

@router.get("/users/{user_id}")
def get_user(user_id: int): ...

@router.post("/users")
def create_user(user_in: UserCreate): ...

2. Service 业务函数

def get_user_by_id(db: Session, user_id: int): ...
def create_user(db: Session, user_in: UserCreate): ...
def check_user_permission(db: Session, user_id: int): ...

3. CRUD 函数（标准模板）

def get(db: Session, id: int): ...          # 单条查询
def get_list(db: Session): ...             # 列表
def create(db: Session, obj_in): ...        # 创建
def update(db: Session, db_obj, obj_in): ...# 更新
def delete(db: Session, id: int): ...       # 删除

变量/常量命名

1. 变量

  • 小写 + 下划线
    user_info
user_list
current_user
access_token

2. 常量（core/config.py）

  • 全部大写 + 下划线
    SECRET_KEY
ACCESS_TOKEN_EXPIRE_MINUTES
DATABASE_URL
REDIS_URL

案例

下面以添加用户、查找用户为例实现代码

1. app/core/config.py 配置

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    API_PREFIX: str = "/api/v1"
    DB_URL: str = "mysql+pymysql://root:123456@localhost:3306/fastapi"
    JWT_SECRET: str = "my-secret"
    JWT_ALGORITHM: str = "HS256"

settings = Settings()

2. app/db/session.py 数据库连接

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.core.config import settings

engine = create_engine(settings.DB_URL, pool_pre_ping=True)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

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

3. app/db/base.py ORM 基类

from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()

4. app/models/user.py 模型（Entity）

from sqlalchemy import Column, Integer, String
from app.db.base import Base

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String(50), unique=True)
    email = Column(String(100))
    hashed_password = Column(String(255))

5. app/schemas/user.py DTO

from pydantic import BaseModel, EmailStr

class UserCreate(BaseModel):
    username: str
    email: EmailStr
    password: str

class UserResp(BaseModel):
    id: int
    username: str
    email: str
    class Config:
        orm_mode = True

6. app/crud/user.py DAO

from sqlalchemy.orm import Session
from app.models.user import User
from app.schemas.user import UserCreate

def create_user(db: Session, user: UserCreate, hashed_pwd: str):
    db_user = User(username=user.username, email=user.email, hashed_password=hashed_pwd)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

def get_user(db: Session, user_id: int):
    return db.query(User).filter(User.id == user_id).first()

7. app/services/user_service.py Service

from sqlalchemy.orm import Session
from app.crud import user as user_crud
from app.schemas.user import UserCreate
from app.core.security import get_password_hash
from fastapi import HTTPException

class UserService:
    @staticmethod
    def create_user(db: Session, user_in: UserCreate):
        # 业务逻辑：检查用户名是否存在
        hashed_pwd = get_password_hash(user_in.password)
        return user_crud.create_user(db, user_in, hashed_pwd)

    @staticmethod
    def get_user(db: Session, user_id: int):
        user = user_crud.get_user(db, user_id)
        if not user:
            raise HTTPException(status_code=404, detail="用户不存在")
        return user

8. app/api/v1/users.py Controller

from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from app.api.deps import get_db
from app.schemas.user import UserCreate, UserResp
from app.services.user_service import UserService

router = APIRouter()

@router.post("/", response_model=UserResp)
def create_user(user: UserCreate, db: Session = Depends(get_db)):
    return UserService.create_user(db, user)

@router.get("/{user_id}", response_model=UserResp)
def get_user(user_id: int, db: Session = Depends(get_db)):
    return UserService.get_user(db, user_id)

9. app/main.py 入口

from fastapi import FastAPI
from app.api.api import api_router
from app.core.config import settings

app = FastAPI(title="企业级 FastAPI 架构")
app.include_router(api_router, prefix=settings.API_PREFIX)

优势

✅ 完全解耦：API、业务、数据库三层分离
✅ 适合团队协作
✅ 可维护、可扩展、可测试
✅ 生产环境稳定
✅ SQLAlchemy 2.0 标准用法
✅ FastAPI 官方推荐架构
✅ 和 Java 架构思维一致

为什么api/有 v1

版本只对外，不对内，v1/v2 是对外接口版本，给前端/第三方调用用的

后端接口是对外暴露的：

  • 前端 App、小程序、第三方系统 都在调用
  • 迭代不能直接改老接口，会崩老版本客户端
  • 所以要做接口版本兼容

举例：

  • 老用户用旧 App 访问：/api/v1/user
  • 新用户用新 App 访问：/api/v2/user

接口地址隔离，互不影响，这就是 v1 的意义。

为什么 models / crud / service / schemas 不用加 v1？

这些是内部私有逻辑，不对外暴露

前端永远不会直接调用：

  • models 数据库实体
  • crud 数据库操作
  • service 业务逻辑
  • schemas 内部DTO

都是后端自己内部调用，没必要版本化。

内部逻辑统一维护一套即可

v1、v2 接口 共用同一套底层 models、service、crud
只在路由层做适配转发，不用重复写两套数据库、两套业务。

架构逻辑：

前端请求 → api/v1 或 api/v2 路由
                ↓
        共用同一套 service / crud / models

避免代码冗余爆炸

如果每个版本都复制一套 models、service：
v1_models、v2_models、v1_service、v2_service……
后期维护直接崩盘，重复代码一大堆。

什么时候内部也要拆分？

只有业务完全不兼容、底层表结构都变了，才会单独拆一套内部逻辑，
但这属于新项目，不是同一个项目的 v1/v2。

总结

  1. api/v1：对外接口版本，兼容前端老客户端
  2. models/crud/service：内部底层，全局共用一套，不用版本
  3. 版本只做在入口路由层，底层统一复用，不重复造轮子

原文出处：/show_1GW3GjtM1lKA.html

吉ICP备15003043号