FastAPI + SQLAlchemy 项目实战:RBAC用户权限管理系统学习笔记 前言 RBAC(Role-Based Access Control,基于角色的访问控制)是目前企业级系统最通用的权限管理模型,核心思路是用户关联角色、角色关联权限 ,通过角色实现用户和权限的解耦,极大提升权限管理的灵活性和可维护性。
本笔记将完整记录一套轻量化企业级 RBAC 用户权限管理系统 的开发全过程,涵盖需求分析、项目设计、数据库设计、代码实现、异常处理、接口测试等全流程,所有代码可直接复制运行,适合零基础入门 FastAPI 项目实战,同时掌握后台权限系统的核心开发逻辑。
技术栈 :FastAPI + SQLAlchemy + Pydantic + JWT + Python3.10+ + SQLite(可无缝切换 MySQL/PostgreSQL)
一、项目需求分析 1.1 功能需求 本项目实现一套基础通用的后台权限管理系统,核心功能覆盖用户、角色、权限全链路管理,具体需求如下:
用户模块 :用户注册、账号密码登录、用户新增/查询/修改/删除(CRUD)、用户分页列表查询
鉴权模块 :基于 JWT 实现令牌签发、过期校验、接口权限拦截、全局登录用户获取
角色模块 :角色创建、角色查询、角色权限绑定、角色用户关联
权限菜单模块 :菜单权限配置、角色菜单权限分配、接口访问权限控制
基础功能 :全局统一接口响应格式、全局异常捕获处理、数据库会话统一管理、分页查询封装
1.2 非功能需求
高性能 :基于 FastAPI 异步特性,支持高并发接口请求
可拓展性 :分层架构设计(路由、业务、模型、数据层),方便后续新增功能
规范性 :统一响应结构、统一异常处理、代码分层清晰、注释完善
可测试性 :原生支持 Swagger 自动接口文档,支持 Postman 全接口测试
二、项目整体设计 2.1 项目架构设计 采用经典的分层架构 ,自上而下分为:路由层(Router)、数据校验层(Schema)、业务逻辑层(Service)、数据访问层(ORM模型)、数据库层,彻底解耦各模块职责,符合软件工程开发规范。
各层级职责:
Router 路由层 :接收前端请求、路由分发、基础权限拦截、参数接收
Schema 模型层 :基于 Pydantic 实现请求参数校验、响应数据格式化
Service 业务层 :处理核心业务逻辑、数据组装、权限判断
Model ORM层 :SQLAlchemy 数据库模型映射、数据表关系定义
公共工具层 :JWT鉴权、异常处理、响应封装、数据库会话管理
2.2 完整项目结构 以下为标准化企业级项目目录结构,结构清晰、便于维护和拓展:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 fastapi-rbac-demo/ ├── app/ │ ├── __init__.py │ ├── main.py │ ├── core/ │ │ ├── __init__.py │ │ ├── config.py │ │ ├── database.py │ │ ├── exception.py │ │ └── security.py │ ├── models/ │ │ ├── __init__.py │ │ ├── user.py │ │ ├── role.py │ │ └── menu.py │ ├── schemas/ │ │ ├── __init__.py │ │ ├── user.py │ │ ├── role.py │ │ └── menu.py │ ├── services/ │ │ ├── __init__.py │ │ ├── user.py │ │ ├── role.py │ │ └── menu.py │ ├── api/ │ │ ├── __init__.py │ │ ├── deps.py │ │ ├── user.py │ │ ├── role.py │ │ └── menu.py │ └── common/ │ ├── __init__.py │ ├── response.py │ └── pagination.py ├── requirements.txt └── README.md
2.3 项目依赖配置 新建 requirements\.txt 文件,配置项目所需所有依赖,安装即可使用:
1 2 3 4 5 6 7 fastapi==0.104.1 uvicorn==0.24.0 sqlalchemy==2.0.23 pydantic==2.4.2 pyjwt==2.8.0 passlib==1.7.4 python-multipart==0.0.6
安装命令:pip install \-r requirements\.txt
三、数据库设计 3.1 数据库设计原则 基于 RBAC 权限模型,设计三张核心业务表 + 两张中间关联表,实现用户-角色-菜单权限 的多对多关联关系,满足灵活的权限分配需求。
核心表结构:用户表、角色表、菜单表、用户角色中间表、角色菜单中间表
3.2 数据表结构详情 3.2.1 用户表(sys_user) 存储系统用户基础信息,包含账号、密码、昵称、状态等核心字段
字段名
字段类型
是否主键
默认值
字段说明
id
Integer
是
自增
用户唯一ID
username
String(50)
否
无
登录账号(唯一)
password
String(100)
否
无
加密后密码
nickname
String(50)
否
无
用户昵称
status
Integer
否
1
用户状态 1-正常 0-禁用
create_time
DateTime
否
当前时间
创建时间
update_time
DateTime
否
当前时间
更新时间
3.2.2 角色表(sys_role) 存储系统角色信息,如超级管理员、普通用户、运营人员等
字段名
字段类型
是否主键
默认值
字段说明
id
Integer
是
自增
角色唯一ID
role_name
String(50)
否
无
角色名称
role_code
String(50)
否
无
角色唯一标识(如admin、user)
description
String(200)
否
空
角色描述
create_time
DateTime
否
当前时间
创建时间
存储系统接口权限、菜单路由信息,用于控制接口访问权限
字段名
字段类型
是否主键
默认值
字段说明
id
Integer
是
自增
菜单ID
menu_name
String(50)
否
无
菜单/权限名称
permission
String(100)
否
无
权限标识(如user:add、user:list)
parent_id
Integer
否
0
父级菜单ID
3.2.4 中间关联表
四、核心配置与公共模块实现 4.1 全局配置文件(core/config.py) 统一管理项目密钥、数据库地址、JWT过期时间等全局配置,方便后期修改
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 from pydantic_settings import BaseSettingsfrom typing import Optional class Settings (BaseSettings ): PROJECT_NAME: str = "FastAPI-RBAC权限管理系统" PROJECT_VERSION: str = "1.0.0" SECRET_KEY: str = "fastapi-rbac-secret-key-2026" ALGORITHM: str = "HS256" ACCESS_TOKEN_EXPIRE_MINUTES: int = 120 DATABASE_URL: str = "sqlite:///./rbac.db" class Config : case_sensitive = True settings = Settings()
4.2 数据库Session管理(core/database.py) 基于SQLAlchemy实现数据库引擎创建、会话管理,封装全局数据库依赖,实现接口自动获取数据库连接,请求结束自动关闭会话
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 from sqlalchemy import create_enginefrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy.orm import sessionmakerfrom app.core.config import settingsengine = create_engine( settings.DATABASE_URL, connect_args={"check_same_thread" : False } if "sqlite" in settings.DATABASE_URL else {} ) SessionLocal = sessionmaker(autocommit=False , autoflush=False , bind=engine) Base = declarative_base() def get_db (): db = SessionLocal() try : yield db finally : db.close()
4.3 JWT鉴权与密码加密(core/security.py) 实现密码哈希加密、JWT令牌生成、令牌校验功能,是登录鉴权的核心模块
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 from datetime import datetime, timedeltafrom typing import Any , Union , Optional from jose import jwt, JWTErrorfrom passlib.context import CryptContextfrom app.core.config import settingspwd_context = CryptContext(schemes=["bcrypt" ], deprecated="auto" ) def create_access_token ( subject: Union [str , Any ], expires_delta: Optional [timedelta] = None ) -> str : """生成JWT访问令牌""" if expires_delta: expire = datetime.utcnow() + expires_delta else : expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES) to_encode = {"exp" : expire, "sub" : str (subject)} encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM) return encoded_jwt def verify_password (plain_password: str , hashed_password: str ) -> bool : """校验密码""" return pwd_context.verify(plain_password, hashed_password) def get_password_hash (password: str ) -> str : """密码加密""" return pwd_context.hash (password)
4.4 统一响应结构(common/response.py) 封装全局统一接口返回格式,所有接口返回数据格式统一,便于前端解析
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 from typing import Any , Optional from fastapi.responses import JSONResponseclass ResponseModel : """统一响应模型""" def __init__ (self, code: int , msg: str , data: Any = None ): self .code = code self .msg = msg self .data = data @staticmethod def success (data: Any = None , msg: str = "请求成功" ) -> JSONResponse: """成功响应""" return JSONResponse( status_code=200 , content={"code" : 200 , "msg" : msg, "data" : data} ) @staticmethod def fail (code: int = 500 , msg: str = "请求失败" , data: Any = None ) -> JSONResponse: """失败响应""" return JSONResponse( status_code=200 , content={"code" : code, "msg" : msg, "data" : data} ) resp = ResponseModel()
4.5 全局异常处理(core/exception.py) 捕获全局异常、自定义业务异常,统一异常返回格式,避免后端报错直接暴露给前端
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 from fastapi import FastAPI, Requestfrom fastapi.exceptions import RequestValidationErrorfrom starlette.exceptions import HTTPException as StarletteHTTPExceptionfrom app.common.response import respclass BusinessException (Exception ): def __init__ (self, code: int , msg: str ): self .code = code self .msg = msg def register_exception (app: FastAPI ): """注册全局异常处理器""" @app.exception_handler(StarletteHTTPException ) async def http_exception_handler (request: Request, exc: StarletteHTTPException ): return resp.fail(code=exc.status_code, msg=exc.detail) @app.exception_handler(RequestValidationError ) async def validation_exception_handler (request: Request, exc: RequestValidationError ): err_msg = exc.errors()[0 ]["msg" ] return resp.fail(code=400 , msg=f"参数校验失败:{err_msg} " ) @app.exception_handler(BusinessException ) async def business_exception_handler (request: Request, exc: BusinessException ): return resp.fail(code=exc.code, msg=exc.msg) @app.exception_handler(Exception ) async def global_exception_handler (request: Request, exc: Exception ): return resp.fail(code=500 , msg=f"服务器异常:{str (exc)} " )
封装通用分页查询方法,适配所有列表查询接口,简化分页逻辑
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 from typing import List , Any from sqlalchemy.orm import Queryclass Pagination : """通用分页工具""" def __init__ (self, page: int = 1 , page_size: int = 10 ): self .page = page self .page_size = page_size def paginate (self, query: Query ) -> dict : """分页查询""" total = query.count() offset = (self .page - 1 ) * self .page_size items = query.offset(offset).limit(self .page_size).all () return { "list" : items, "total" : total, "page" : self .page, "page_size" : self .page_size, "total_page" : (total + self .page_size - 1 ) // self .page_size }
五、ORM模型关系映射实现 基于SQLAlchemy实现数据表模型映射,定义用户-角色、角色-菜单 多对多关联关系,自动关联查询数据
5.1 中间表模型(models/__init__.py) 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 from sqlalchemy import Column, Integer, ForeignKeyfrom app.core.database import Basesys_user_role = Base.metadata.tables.get("sys_user_role" ) or Base( __tablename__ = "sys_user_role" , Column("user_id" , Integer, ForeignKey("sys_user.id" ), primary_key=True ), Column("role_id" , Integer, ForeignKey("sys_role.id" ), primary_key=True ) ) sys_role_menu = Base.metadata.tables.get("sys_role_menu" ) or Base( __tablename__ = "sys_role_menu" , Column("role_id" , Integer, ForeignKey("sys_role.id" ), primary_key=True ), Column("menu_id" , Integer, ForeignKey("sys_menu.id" ), primary_key=True ) )
5.2 用户模型(models/user.py) 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 from datetime import datetimefrom sqlalchemy import Column, Integer, String, DateTimefrom sqlalchemy.orm import relationshipfrom app.core.database import Basefrom app.models import sys_user_roleclass User (Base ): __tablename__ = "sys_user" id = Column(Integer, primary_key=True , index=True , autoincrement=True ) username = Column(String(50 ), unique=True , nullable=False , comment="登录账号" ) password = Column(String(100 ), nullable=False , comment="密码" ) nickname = Column(String(50 ), comment="用户昵称" ) status = Column(Integer, default=1 , comment="状态1正常0禁用" ) create_time = Column(DateTime, default=datetime.now, comment="创建时间" ) update_time = Column(DateTime, default=datetime.now, onupdate=datetime.now, comment="更新时间" ) roles = relationship("Role" , secondary=sys_user_role, back_populates="users" )
5.3 角色模型(models/role.py) 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 from datetime import datetimefrom sqlalchemy import Column, Integer, String, DateTimefrom sqlalchemy.orm import relationshipfrom app.core.database import Basefrom app.models import sys_user_role, sys_role_menuclass Role (Base ): __tablename__ = "sys_role" id = Column(Integer, primary_key=True , index=True , autoincrement=True ) role_name = Column(String(50 ), nullable=False , comment="角色名称" ) role_code = Column(String(50 ), unique=True , nullable=False , comment="角色标识" ) description = Column(String(200 ), default="" , comment="角色描述" ) create_time = Column(DateTime, default=datetime.now, comment="创建时间" ) users = relationship("User" , secondary=sys_user_role, back_populates="roles" ) menus = relationship("Menu" , secondary=sys_role_menu, back_populates="roles" )
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 from sqlalchemy import Column, Integer, Stringfrom sqlalchemy.orm import relationshipfrom app.core.database import Basefrom app.models import sys_role_menuclass Menu (Base ): __tablename__ = "sys_menu" id = Column(Integer, primary_key=True , index=True , autoincrement=True ) menu_name = Column(String(50 ), nullable=False , comment="菜单名称" ) permission = Column(String(100 ), nullable=False , comment="权限标识" ) parent_id = Column(Integer, default=0 , comment="父级ID" ) roles = relationship("Role" , secondary=sys_role_menu, back_populates="menus" )
六、Pydantic Schema数据模型 用于请求参数校验、响应数据格式化,过滤敏感字段(如密码),保证接口数据安全和规范性
6.1 用户Schema(schemas/user.py) 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 from typing import List , Optional from pydantic import BaseModel, Fieldclass UserRegister (BaseModel ): username: str = Field(..., description="登录账号" ) password: str = Field(..., description="登录密码" ) nickname: Optional [str ] = Field(None , description="用户昵称" ) class UserLogin (BaseModel ): username: str password: str class UserInfo (BaseModel ): id : int username: str nickname: Optional [str ] status: int class Config : from_attributes = True class UserPageQuery (BaseModel ): page: int = 1 page_size: int = 10 username: Optional [str ] = None
6.2 角色、菜单Schema 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 from pydantic import BaseModel, Optional class RoleCreate (BaseModel ): role_name: str role_code: str description: Optional [str ] = None class RoleInfo (BaseModel ): id : int role_name: str role_code: str description: Optional [str ] class Config : from_attributes = True from pydantic import BaseModelclass MenuCreate (BaseModel ): menu_name: str permission: str parent_id: int = 0 class MenuInfo (BaseModel ): id : int menu_name: str permission: str parent_id: int class Config : from_attributes = True
七、依赖项封装(api/deps.py) 封装全局依赖,实现JWT登录校验、获取当前登录用户、接口权限拦截
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 from typing import Generator, Optional from fastapi import Depends, HTTPException, statusfrom fastapi.security import OAuth2PasswordBearerfrom sqlalchemy.orm import Sessionfrom jose import jwt, JWTErrorfrom app.core.database import get_dbfrom app.core.config import settingsfrom app.models.user import Userfrom app.core.exception import BusinessExceptionoauth2_scheme = OAuth2PasswordBearer(tokenUrl="/user/login" ) def get_current_user (token: str = Depends(oauth2_scheme ), db: Session = Depends(get_db ) ) -> User: """获取当前登录用户""" try : payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM]) user_id: str = payload.get("sub" ) if user_id is None : raise BusinessException(code=401 , msg="令牌无效" ) except JWTError: raise BusinessException(code=401 , msg="令牌过期或无效" ) user = db.query(User).filter (User.id == user_id).first() if not user: raise BusinessException(code=401 , msg="用户不存在" ) if user.status == 0 : raise BusinessException(code=403 , msg="账号已被禁用" ) return user def check_permission (permission: str ): """权限校验装饰器""" def wrapper (user: User = Depends(get_current_user ) ): permissions = [] for role in user.roles: for menu in role.menus: permissions.append(menu.permission) if permission not in permissions: raise BusinessException(code=403 , msg="无接口访问权限" ) return user return wrapper
八、业务逻辑层Service实现 8.1 用户业务(services/user.py) 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 from sqlalchemy.orm import Sessionfrom typing import Optional , Dict from app.models.user import Userfrom app.schemas.user import UserRegister, UserPageQueryfrom app.core.security import get_password_hash, verify_password, create_access_tokenfrom app.core.exception import BusinessExceptionfrom app.common.pagination import Paginationclass UserService : @staticmethod def register (db: Session, data: UserRegister ) -> Dict : """用户注册""" exist_user = db.query(User).filter (User.username == data.username).first() if exist_user: raise BusinessException(code=400 , msg="账号已存在" ) db_user = User( username=data.username, password=get_password_hash(data.password), nickname=data.nickname ) db.add(db_user) db.commit() db.refresh(db_user) return {"id" : db_user.id , "username" : db_user.username} @staticmethod def login (db: Session, username: str , password: str ) -> Dict : """用户登录""" user = db.query(User).filter (User.username == username).first() if not user: raise BusinessException(code=400 , msg="账号或密码错误" ) if not verify_password(password, user.password): raise BusinessException(code=400 , msg="账号或密码错误" ) if user.status == 0 : raise BusinessException(code=403 , msg="账号已禁用" ) access_token = create_access_token(subject=user.id ) return {"access_token" : access_token, "token_type" : "bearer" } @staticmethod def get_user_page (db: Session, query: UserPageQuery ) -> Dict : """用户分页查询""" q = db.query(User) if query.username: q = q.filter (User.username.like(f"%{query.username} %" )) page = Pagination(page=query.page, page_size=query.page_size) return page.paginate(q) @staticmethod def delete_user (db: Session, user_id: int ): """删除用户""" user = db.query(User).filter (User.id == user_id).first() if not user: raise BusinessException(code=400 , msg="用户不存在" ) db.delete(user) db.commit() user_service = UserService()
8.2 角色、菜单业务Service 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 from sqlalchemy.orm import Sessionfrom app.models.role import Rolefrom app.schemas.role import RoleCreatefrom app.core.exception import BusinessExceptionclass RoleService : @staticmethod def create_role (db: Session, data: RoleCreate ) -> Role: exist_role = db.query(Role).filter (Role.role_code == data.role_code).first() if exist_role: raise BusinessException(code=400 , msg="角色标识已存在" ) role = Role(**data.model_dump()) db.add(role) db.commit() db.refresh(role) return role role_service = RoleService() from sqlalchemy.orm import Sessionfrom app.models.menu import Menufrom app.schemas.menu import MenuCreateclass MenuService : @staticmethod def create_menu (db: Session, data: MenuCreate ) -> Menu: menu = Menu(**data.model_dump()) db.add(menu) db.commit() db.refresh(menu) return menu menu_service = MenuService()
九、路由接口Router实现 9.1 用户路由(api/user.py) 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 from fastapi import APIRouter, Dependsfrom sqlalchemy.orm import Sessionfrom app.core.database import get_dbfrom app.schemas.user import UserRegister, UserLogin, UserPageQueryfrom app.services.user import user_servicefrom app.common.response import respfrom app.api.deps import get_current_userrouter = APIRouter(prefix="/user" , tags=["用户模块" ]) @router.post("/register" , summary="用户注册" ) def register (data: UserRegister, db: Session = Depends(get_db ) ): res = user_service.register(db, data) return resp.success(data=res, msg="注册成功" ) @router.post("/login" , summary="用户登录" ) def login (data: UserLogin, db: Session = Depends(get_db ) ): res = user_service.login(db, data.username, data.password) return resp.success(data=res, msg="登录成功" ) @router.get("/page" , summary="用户分页列表" ) def user_page (query: UserPageQuery = Depends( ), db: Session = Depends(get_db ), current_user = Depends(get_current_user ) ): res = user_service.get_user_page(db, query) return resp.success(data=res) @router.delete("/{user_id}" , summary="删除用户" ) def delete_user (user_id: int , db: Session = Depends(get_db ), current_user = Depends(get_current_user ) ): user_service.delete_user(db, user_id) return resp.success(msg="删除成功" )
9.2 角色、菜单路由 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 from fastapi import APIRouter, Dependsfrom sqlalchemy.orm import Sessionfrom app.core.database import get_dbfrom app.schemas.role import RoleCreatefrom app.services.role import role_servicefrom app.common.response import respfrom app.api.deps import get_current_userrouter = APIRouter(prefix="/role" , tags=["角色模块" ]) @router.post("/create" , summary="创建角色" ) def create_role (data: RoleCreate, db: Session = Depends(get_db ), current_user = Depends(get_current_user ) ): res = role_service.create_role(db, data) return resp.success(data=res, msg="角色创建成功" ) from fastapi import APIRouter, Dependsfrom sqlalchemy.orm import Sessionfrom app.core.database import get_dbfrom app.schemas.menu import MenuCreatefrom app.services.menu import menu_servicefrom app.common.response import respfrom app.api.deps import get_current_userrouter = APIRouter(prefix="/menu" , tags=["菜单权限模块" ]) @router.post("/create" , summary="创建菜单权限" ) def create_menu (data: MenuCreate, db: Session = Depends(get_db ), current_user = Depends(get_current_user ) ): res = menu_service.create_menu(db, data) return resp.success(data=res, msg="权限创建成功" )
十、项目入口文件(main.py) 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 from fastapi import FastAPIimport uvicornfrom app.core.database import engine, Basefrom app.core.exception import register_exceptionfrom app.api import user, role, menuBase.metadata.create_all(bind=engine) app = FastAPI(title="FastAPI-RBAC权限管理系统" , version="1.0.0" ) register_exception(app) app.include_router(user.router) app.include_router(role.router) app.include_router(menu.router) @app.get("/" , summary="项目根路径" ) def root (): return {"msg" : "FastAPI RBAC权限系统运行成功,访问/docs查看接口文档" } if __name__ == "__main__" : uvicorn.run(app="app.main:app" , host="0.0.0.0" , port=8000 , reload=True )
十一、接口测试 11.1 Swagger在线文档测试 FastAPI 原生自动生成交互式接口文档,项目启动后可通过以下地址访问:
测试流程:
启动项目,访问 /docs 进入接口文档页面
调用 /user/register 注册测试账号
调用 /user/login 获取 JWT Token
点击文档右上角 Authorize,输入 Bearer 令牌 完成授权
测试需要登录的分页查询、角色创建等接口
11.2 Postman 测试 1. 新建 Postman 集合,配置全局请求地址 http://127\.0\.0\.1:8000
2. 调用登录接口获取 token,在请求头添加 Authorization: Bearer \{token\}
3. 依次测试注册、登录、分页查询、角色权限创建等接口,验证数据返回、异常拦截、权限控制功能是否正常
十二、完整项目开发流程总结 本项目严格遵循企业级后端开发流程,标准化开发步骤如下,可复用至所有 FastAPI 项目:
12.1 第一步:环境搭建与依赖配置 创建项目目录、配置依赖文件、安装所需第三方库,搭建基础项目架构。
12.2 第二步:数据库设计与建表 根据业务需求设计数据表结构,确定字段、关联关系,基于SQLAlchemy编写ORM模型,定义多对多中间表,实现数据库模型映射。
12.3 第三步:公共模块封装 封装数据库会话、JWT鉴权、密码加密、统一响应、全局异常、分页工具等公共基础模块,为业务开发提供支撑。
12.4 第四步:Schema数据模型定义 基于Pydantic定义请求参数、响应数据模型,实现参数自动校验、敏感字段过滤。
12.5 第五步:Service业务逻辑开发 编写核心业务逻辑,处理数据校验、数据库增删改查、业务规则判断,解耦路由和业务逻辑。
12.6 第六步:Router路由接口开发 定义接口路由、请求方式、接口注释,配置依赖项(登录校验、权限校验),调用业务层方法返回统一格式数据。
12.7 第七步:全局配置与项目整合 注册路由、全局异常处理器,配置项目启动参数,完成项目整体整合。
12.8 第八步:接口测试与调试 通过 Swagger、Postman 全量测试接口,验证正常流程、异常流程、权限拦截、分页查询等功能。
十三、项目总结 本笔记完整实现了FastAPI + SQLAlchemy 轻量化RBAC权限管理系统 ,覆盖后端开发全流程核心知识点。通过本项目实战,可熟练掌握 FastAPI 分层开发思想、SQLAlchemy ORM 关联映射、JWT 鉴权机制、RBAC权限模型、全局异常与响应封装、分页查询等企业级核心技能。
项目代码结构规范、拓展性强,可在此基础上继续拓展按钮权限、数据权限、日志记录、密码重置、角色权限批量分配 等高级功能。