摘要本文基于 FastAPI 异步 SQLAlchemy 实战项目系统梳理路由层、数据库配置层、ORM 模型层、crud 操作层、schemas 校验层、utils 工具层共六层的标准导入规范逐层给出导入语句、功能作用与完整可复用代码示例并标注 CORS 凭证冲突、default函数引用、Pydantic v2from_attributes等实战易错点可作为日常开发速查手册。目录全局总览项目分层导包速查表一、路由层导包main.py / routers 目录1.1 创建项目主应用1.2 模块化拆分路由1.3 依赖注入数据库会话1.4 接口查询参数校验1.5 HTTP 异常与状态码1.6 跨域中间件配置二、数据库配置层导包config/db_conf.py三、ORM 模型层导包models 目录3.1 基础模型基类与类型注解3.2 基础字段类型3.3 外键约束3.4 自定义索引3.5 枚举字段3.6 可选类型注解四、数据库操作层导包crud 目录4.1 普通查询 select4.2 聚合函数 func4.3 更新语句 update五、数据校验层导包schemas 目录六、全局工具层导包utils / 全局通用6.1 密码加密工具6.2 唯一 ID 生成Token6.3 时间处理全局总览项目分层导包速查表开发优先查阅此总览快速定位分层对应的所有导入无需通篇翻阅# 项目分层导包总览速查极简版 # 每层 所属目录 核心导入包 常用导入对象 1. 路由层routers / main.py 核心包fastapi 导入对象FastAPI、APIRouter、Depends、Query、HTTPException、status、CORSMiddleware 2. 数据库配置层config 核心包sqlalchemy.ext.asyncio 导入对象create_async_engine、async_sessionmaker、AsyncSession 3. 模型层models 核心包1sqlalchemy.orm 导入对象DeclarativeBase、Mapped、mapped_column 核心包2sqlalchemy 导入对象Integer、String、Text、DateTime、ForeignKey、Index、Enum 4. 数据库操作层crud 核心包1sqlalchemy 导入对象select、func、update 核心包2sqlalchemy.ext.asyncio 导入对象AsyncSession 5. 数据校验层schemas 核心包pydantic 导入对象BaseModel、ConfigDict 6. 工具层utils/全局通用 核心包1passlib.context 导入对象CryptContext 核心包2系统标准库 导入对象uuid、datetime、timedelta、typing.Optional一、路由层导包main.py / routers 目录适用场景项目主程序、所有接口路由文件负责接口创建、路由拆分、参数校验、异常处理、跨域配置1.1 创建项目主应用导入语句from fastapi import FastAPI功能作用初始化 FastAPI 主程序实例# main.pyfromfastapiimportFastAPI# 创建全局应用实例appFastAPI()1.2 模块化拆分路由导入语句from fastapi import APIRouter功能作用拆分业务模块路由统一接口前缀与接口分类标签# routers/news.pyfromfastapiimportAPIRouter# 定义新闻模块路由routerAPIRouter(prefix/api/news,tags[news])1.3 依赖注入数据库会话导入语句from fastapi import Depends功能作用全局依赖注入常用于注入数据库会话、权限校验# routers/news.pyfromfastapiimportDependsfromconfig.db_confimportget_dbfromsqlalchemy.ext.asyncioimportAsyncSession# 接口依赖数据库会话asyncdefget_categories(db:AsyncSessionDepends(get_db)):pass1.4 接口查询参数校验导入语句from fastapi import Query功能作用URL 参数校验、别名适配、数值范围限制、默认值配置# routers/news.pyfromfastapiimportQuery# 分页、分类参数校验asyncdefget_news_list(category_id:intQuery(...,aliascategoryId),page_size:intQuery(10,aliaspageSize,le100),):pass1.5 HTTP 异常与状态码导入语句from fastapi import HTTPException, status功能作用主动抛出标准 HTTP 错误统一接口异常返回格式# routers/users.pyfromfastapiimportHTTPException,status# 规范写法推荐用 status 枚举避免硬编码魔法数字raiseHTTPException(status_codestatus.HTTP_400_BAD_REQUEST,detail用户已存在)# 简写写法raiseHTTPException(status_code404,detail新闻不存在)1.6 跨域中间件配置导入语句from fastapi.middleware.cors import CORSMiddleware功能作用解决前后端联调跨域请求拦截问题# main.pyfromfastapiimportFastAPIfromfastapi.middleware.corsimportCORSMiddleware appFastAPI()# 全局跨域配置app.add_middleware(CORSMiddleware,# ⚠️ 注意allow_credentialsTrue 时浏览器禁止 Access-Control-Allow-Origin 为 *# 必须显式列出前端域名否则带 Cookie / 鉴权头的跨域请求会被浏览器直接拦截。allow_origins[http://localhost:5173,https://your-frontend.com],# 显式域名勿用 *allow_credentialsTrue,allow_methods[*],allow_headers[*],)排坑提示如果你只是本地联调、且前端不走登录态不传 Cookie / Authorization可以简化成allow_origins[*]同时把allow_credentials设为False只要二者同时满足通配 带凭证就会失效。二、数据库配置层导包config/db_conf.py适用场景仅数据库配置文件用于创建异步数据库引擎、会话工厂、全局 DB 依赖导入语句from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession功能作用适配 aiomysql 异步数据库实现非阻塞数据库连接# config/db_conf.pyfromsqlalchemy.ext.asyncioimportcreate_async_engine,async_sessionmaker,AsyncSession# 创建异步数据库引擎async_enginecreate_async_engine(mysqlaiomysql://root:123456localhost:3306/news_app?charsetutf8,echoTrue,pool_size10,max_overflow20)# 创建异步会话工厂AsyncSessionLocalasync_sessionmaker(bindasync_engine,class_AsyncSession,expire_on_commitFalse)# 全局获取数据库会话依赖函数asyncdefget_db():asyncwithAsyncSessionLocal()assession:try:yieldsessionawaitsession.commit()exceptException:# 捕获具体异常而非裸 except裸 except 会连 KeyboardInterrupt 一起吞掉awaitsession.rollback()raisefinally:awaitsession.close()小提示生产环境建议再加pool_pre_pingTrue在取出连接前做一次心跳探测自动踢掉被数据库端已断开的死连接。三、ORM 模型层导包models 目录适用场景所有数据表模型定义文件用于创建数据表结构、字段约束、索引、外键3.1 基础模型基类与类型注解导入语句from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column# models/base.pyfromdatetimeimportdatetimefromsqlalchemy.ormimportDeclarativeBase,Mapped,mapped_columnfromsqlalchemyimportDateTime# 所有模型的父类统一公共字段classBase(DeclarativeBase):# ⚠️ 关键default 必须传函数对象datetime.now不能加括号调用datetime.now()。# 加括号会在类定义时立即执行一次导致所有记录都用同一个导入时刻的时间戳。created_at:Mapped[datetime]mapped_column(DateTime,defaultdatetime.now,comment创建时间)updated_at:Mapped[datetime]mapped_column(DateTime,defaultdatetime.now,onupdatedatetime.now,comment更新时间)3.2 基础字段类型导入语句from sqlalchemy import DateTime, Integer, String, Text# models/news.pyfromdatetimeimportdatetimefromsqlalchemy.ormimportMapped,mapped_columnfromsqlalchemyimportDateTime,Integer,String,Textfrommodels.baseimportBaseclassNews(Base):__tablename__newsid:Mapped[int]mapped_column(Integer,primary_keyTrue,autoincrementTrue,comment新闻ID)title:Mapped[str]mapped_column(String(255),nullableFalse,comment新闻标题)content:Mapped[str]mapped_column(Text,nullableFalse,comment新闻内容)publish_time:Mapped[datetime]mapped_column(DateTime,defaultdatetime.now,comment发布时间)3.3 外键约束导入语句from sqlalchemy import ForeignKey# models/news.pyfromsqlalchemyimportForeignKey# 关联新闻分类表category_id:Mapped[int]mapped_column(Integer,ForeignKey(news_category.id),nullableFalse)3.4 自定义索引导入语句from sqlalchemy import Index# models/news.pyfromsqlalchemyimportIndexclassNews(Base):__tablename__news# 省略字段定义...# 数据表索引命名统一以 idx_ 前缀开头见名知意__table_args__(Index(idx_category_id,category_id),Index(idx_publish_time,publish_time))命名规范索引名统一用idx_字段名前缀外键列索引也用idx_不要混用fk_前缀避免一眼看不出它本质是索引。3.5 枚举字段导入语句from sqlalchemy import Enum as SAEnum# models/users.pyfromenumimportEnumfromtypingimportOptionalfromsqlalchemyimportEnumasSAEnumfromsqlalchemy.ormimportMapped,mapped_columnfrommodels.baseimportBaseclassGenderEnum(Enum):malemalefemalefemaleunknownunknownclassUser(Base):gender:Mapped[Optional[str]]mapped_column(SAEnum(GenderEnum),defaultGenderEnum.unknown,comment性别)3.6 可选类型注解导入语句from typing import Optional# models/news.pyfromtypingimportOptional# 可空字段定义description:Mapped[Optional[str]]mapped_column(String(500),comment新闻简介)四、数据库操作层导包crud 目录适用场景所有数据库增删改查业务逻辑文件4.1 普通查询 select导入语句from sqlalchemy import select# crud/news.pyfromsqlalchemyimportselectfromsqlalchemy.ext.asyncioimportAsyncSessionfrommodels.newsimportNewsasyncdefget_news_by_category(db:AsyncSession,category_id:int,skip:int0,limit:int10):stmtselect(News).where(News.category_idcategory_id).offset(skip).limit(limit)resultawaitdb.execute(stmt)returnresult.scalars().all()4.2 聚合函数 func导入语句from sqlalchemy import func# crud/news.pyfromsqlalchemyimportselect,funcfromsqlalchemy.ext.asyncioimportAsyncSessionfrommodels.newsimportNews# 统计分类下新闻总数asyncdefcount_news_by_category(db:AsyncSession,category_id:int):stmtselect(func.count(News.id)).where(News.category_idcategory_id)resultawaitdb.execute(stmt)returnresult.scalar_one()4.3 更新语句 update导入语句from sqlalchemy import update# crud/news.pyfromsqlalchemyimportupdatefromsqlalchemy.ext.asyncioimportAsyncSessionfrommodels.newsimportNews# 新闻浏览量自增asyncdefadd_news_view(db:AsyncSession,news_id:int):stmtupdate(News).where(News.idnews_id).values(viewsNews.views1)awaitdb.execute(stmt)awaitdb.commit()五、数据校验层导包schemas 目录适用场景定义接口请求参数、响应返回体模型自动完成数据校验导入语句from pydantic import BaseModel, ConfigDict# schemas/users.pyfrompydanticimportBaseModel,ConfigDict# 请求参数模型classUserRequest(BaseModel):username:strpassword:str# 响应返回模型classUserResponse(BaseModel):id:intusername:str# Pydantic v2 写法用 model_config ConfigDict 替代 v1 的 class Config: orm_modemodel_configConfigDict(from_attributesTrue)版本提示class Config: orm_mode True是 Pydanticv1语法。Pydanticv2已改为model_config ConfigDict(from_attributesTrue)orm_mode虽作为别名暂时可用但已废弃建议直接写新写法。本文示例统一用 v2。六、全局工具层导包utils / 全局通用6.1 密码加密工具导入语句from passlib.context import CryptContext# utils/security.pyfrompasslib.contextimportCryptContext# 初始化加密器pwd_contextCryptContext(schemes[bcrypt],deprecatedauto)# 密码哈希加密defget_password_hash(password:str):returnpwd_context.hash(password)6.2 唯一 ID 生成Token导入语句import uuid# crud/users.pyimportuuid# 生成唯一登录 Tokentokenstr(uuid.uuid4())6.3 时间处理导入语句from datetime import datetime, timedelta# crud/users.pyfromdatetimeimportdatetime,timedelta# 设置 Token 7 天过期expires_atdatetime.now()timedelta(days7)
