FastApiAdmin 项目后端
📖 项目介绍
项目概述
FastApiAdmin 后端是一个基于 Python FastAPI 构建的现代化后台管理系统后端服务。
- 🚀 高性能：基于 FastAPI 异步框架，提供卓越的性能
- 🧠 智能类型：Pydantic 模型提供完整的类型检查
- 🔐 安全认证：JWT + OAuth2 认证机制
- 🛡️ 权限控制：RBAC 模型实现细粒度权限控制
- 📊 数据库支持：SQLAlchemy ORM 支持多种数据库
- 🔄 异步任务：APScheduler 实现定时任务
- 📚 自动文档：自动生成 Swagger 和 Redoc API 文档
核心特性
🔐 权限管理
- ✅ JWT 认证机制
- ✅ OAuth2 支持
- ✅ RBAC 权限模型
- ✅ 菜单权限控制
- ✅ 按钮级别权限控制
- ✅ 数据权限控制
🗄️ 数据管理
- ✅ SQLAlchemy ORM
- ✅ Alembic 数据库迁移
- ✅ 多数据库支持
- ✅ 连接池管理
- ✅ 事务管理
🔄 异步任务
- ✅ APScheduler 定时任务
- ✅ 异步任务队列
- ✅ 任务状态监控
📊 缓存系统
- ✅ Redis 缓存支持
- ✅ 缓存预热
- ✅ 缓存失效策略
📈 日志监控
- ✅ 结构化日志
- ✅ 日志级别控制
- ✅ 日志文件轮转
- ✅ 在线用户监控
🏗️ 技术栈
核心框架
- FastAPI - 现代、快速（高性能）的 Web 框架
- Pydantic - 数据验证和设置管理
- SQLAlchemy - SQL 工具包和 ORM
- Alembic - 数据库迁移工具
认证授权
- JWT - JSON Web Tokens
- OAuth2 - 开放授权标准
- PassLib - 密码哈希库
数据库
- MySQL - 关系型数据库
- Redis - 内存数据库
- MongoDB - 文档数据库（可选）
异步任务
- APScheduler - 高级 Python 调度库
日志监控
- Loguru - 简化 Python 日志记录
- Python-json-logger - JSON 格式日志
⚙️ 环境配置
环境变量说明
项目支持多种环境配置：
开发环境 (.env.dev.py)
python
# 应用配置
APP_ENV = "development"
DEBUG = True
# 服务器配置
HOST = "127.0.0.1"
PORT = 8000
# 数据库配置
DATABASE_URL = "mysql+pymysql://root:password@localhost:3306/fastapi_admin"
# Redis配置
REDIS_URL = "redis://localhost:6379/0"
# JWT配置
JWT_SECRET_KEY = "your-secret-key"
JWT_ALGORITHM = "HS256"
JWT_ACCESS_TOKEN_EXPIRE_MINUTES = 30
# 其他配置生产环境 (.env.prod.py)
python
# 应用配置
APP_ENV = "production"
DEBUG = False
# 服务器配置
HOST = "0.0.0.0"
PORT = 8000
# 数据库配置
DATABASE_URL = "mysql+pymysql://user:password@db:3306/fastapi_admin"
# Redis配置
REDIS_URL = "redis://redis:6379/0"
# JWT配置
JWT_SECRET_KEY = "your-production-secret-key"
JWT_ALGORITHM = "HS256"
JWT_ACCESS_TOKEN_EXPIRE_MINUTES = 30
# 其他配置📁 项目结构详解
完整目录结构
plain
FastapiAdmin/backend/
├── app/ # 应用核心代码
│ ├── api/ # API 接口层
│ │ ├── v1/ # API v1 版本
│ │ │ ├── module_common/ # 公共模块
│ │ │ │ ├── file/ # 文件接口
│ │ │ │ └── health/ # 健康检查
│ │ │ ├── module_monitor/ # 监控模块
│ │ │ │ ├── cache/ # 缓存监控
│ │ │ │ ├── online/ # 在线用户
│ │ │ │ ├── resource/ # 资源监控
│ │ │ │ └── server/ # 服务器监控
│ │ │ ├── module_system/ # 系统模块
│ │ │ │ ├── auth/ # 认证接口
│ │ │ │ ├── dept/ # 部门管理
│ │ │ │ ├── dict/ # 字典管理
│ │ │ │ ├── log/ # 日志管理
│ │ │ │ ├── menu/ # 菜单管理
│ │ │ │ ├── notice/ # 公告管理
│ │ │ │ ├── params/ # 参数管理
│ │ │ │ ├── position/ # 职位管理
│ │ │ │ ├── role/ # 角色管理
│ │ │ │ └── user/ # 用户管理
│ │ └── __init__.py # API 初始化
│ ├── common/ # 公共代码
│ │ ├── constant.py # 常量定义
│ │ ├── dataclasses.py # 数据类定义
│ │ ├── enums.py # 枚举定义
│ │ ├── request.py # 请求相关
│ │ └── response.py # 响应相关
│ ├── config/ # 配置管理
│ │ ├── path_conf.py # 路径配置
│ │ └── setting.py # 应用配置
│ ├── core/ # 核心功能
│ │ ├── base_crud.py # 基础 CRUD
│ │ ├── base_model.py # 基础模型
│ │ ├── base_params.py # 基础参数
│ │ ├── base_schema.py # 基础模式
│ │ ├── database.py # 数据库配置
│ │ ├── dependencies.py # 依赖注入
│ │ ├── discover.py # 自动发现
│ │ ├── exceptions.py # 异常处理
│ │ ├── http_limit.py # HTTP 限流
│ │ ├── logger.py # 日志配置
│ │ ├── middlewares.py # 中间件
│ │ ├── permission.py # 权限管理
│ │ ├── redis_crud.py # Redis CRUD
│ │ ├── router_class.py # 路由类
│ │ ├── security.py # 安全相关
│ │ ├── serialize.py # 序列化
│ │ └── validator.py # 数据验证
│ ├── plugin/ # 插件系统
│ │ ├── module_application/ # 应用模块
│ │ │ ├── ai/ # AI 模块
│ │ │ ├── job/ # 任务模块
│ │ │ ├── myapp/ # 我的应用
│ │ │ └── workflow/ # 工作流
│ │ ├── module_example/ # 示例模块
│ │ ├── module_generator/ # 代码生成
│ │ └── init_app.py # 插件初始化
│ ├── scripts/ # 脚本工具
│ │ ├── data/ # 初始化数据
│ │ └── initialize.py # 初始化脚本
│ ├── utils/ # 工具函数
│ │ ├── banner.py # 启动横幅
│ │ ├── captcha_util.py # 验证码工具
│ │ ├── common_util.py # 通用工具
│ │ ├── console.py # 控制台工具
│ │ ├── cron_util.py # 定时任务工具
│ │ ├── excel_util.py # Excel 工具
│ │ ├── hash_bcrpy_util.py # 哈希工具
│ │ ├── import_util.py # 导入工具
│ │ ├── ip_local_util.py # IP 工具
│ │ ├── re_util.py # 正则工具
│ │ ├── string_util.py # 字符串工具
│ │ ├── time_util.py # 时间工具
│ │ └── upload_util.py # 上传工具
│ ├── alembic/ # 数据库迁移
│ │ ├── README
│ │ ├── env.py
│ │ └── script.py.mako
│ └── __init__.py # 应用初始化
├── env/ # 环境配置
│ ├── .env.dev.example # 开发环境示例
│ └── .env.prod.example # 生产环境示例
├── static/ # 静态资源
├── tests/ # 测试代码
│ ├── conftest.py
│ └── test_main.py
├── alembic.ini # Alembic 配置
├── banner.txt # 启动横幅
├── main.py # 应用入口
├── pyproject.toml # Python 项目配置
└── requirements.txt # 依赖包列表核心文件说明
🔧 应用入口
| 文件 | 说明 |
|---|---|
main.py | 应用入口文件，初始化 FastAPI 应用 |
app/config/setting.py | 应用配置管理 |
app/core/database.py | 数据库连接配置 |
🗂️ 核心目录详解
1. API 接口层 (app/api/v1/)
采用模块化架构设计：
- module_common/: 公共模块，包含文件上传、健康检查等通用接口
- module_monitor/: 监控模块，包含缓存监控、在线用户、服务器监控等
- module_system/: 系统模块，包含用户、角色、菜单、权限等核心系统功能
每个模块内部采用分层架构：
- controller.py: 控制器，处理 HTTP 请求和响应
- service.py: 业务逻辑层，实现核心业务逻辑
- crud.py: 数据访问层，处理数据库操作
- model.py: 数据模型，定义数据库表结构
- schema.py: 数据模式，定义 API 数据结构
2. 核心功能 (app/core/)
- base_model.py: 基础数据模型
- base_crud.py: 基础 CRUD 操作
- database.py: 数据库连接和会话管理
- security.py: 安全相关（JWT、密码哈希等）
- permission.py: 权限管理
- middlewares.py: 中间件
- logger.py: 日志配置
- validator.py: 数据验证
3. 插件系统 (app/plugin/)
- module_application/: 应用模块，包含 AI、任务、工作流等功能
- module_example/: 示例模块，提供开发参考
- module_generator/: 代码生成模块，自动生成前后端代码
4. 工具函数 (app/utils/)
- common_util.py: 通用工具函数
- time_util.py: 时间处理工具
- excel_util.py: Excel 处理工具
- upload_util.py: 文件上传工具
5. 公共代码 (app/common/)
- constant.py: 常量定义
- enums.py: 枚举定义
- response.py: 统一响应格式
🔗 API 接口管理
模块化 API 设计
python
# app/api/v1/module_system/user/controller.py
from fastapi import APIRouter, Depends
from app.core.dependencies import get_current_active_user
from app.core.security import User
from .schema import UserCreate, UserUpdate, UserInfo
from .service import UserService
router = APIRouter(prefix="/user", tags=["user"])
@router.post("/", response_model=UserInfo)
async def create_user(
user: UserCreate,
current_user: User = Depends(get_current_active_user)
):
"""创建用户"""
return await UserService.create_user(user)
@router.get("/{user_id}", response_model=UserInfo)
async def get_user(
user_id: int,
current_user: User = Depends(get_current_active_user)
):
"""获取用户信息"""
return await UserService.get_user_by_id(user_id)
@router.put("/{user_id}", response_model=UserInfo)
async def update_user(
user_id: int,
user: UserUpdate,
current_user: User = Depends(get_current_active_user)
):
"""更新用户"""
return await UserService.update_user(user_id, user)
@router.delete("/{user_id}")
async def delete_user(
user_id: int,
current_user: User = Depends(get_current_active_user)
):
"""删除用户"""
return await UserService.delete_user(user_id)业务逻辑层
python
# app/api/v1/module_system/user/service.py
from .crud import UserCRUD
from .schema import UserCreate, UserUpdate
from app.core.security import get_password_hash
class UserService:
@staticmethod
async def create_user(user: UserCreate):
"""创建用户"""
# 密码哈希处理
user.password = get_password_hash(user.password)
# 创建用户
return await UserCRUD.create(user)
@staticmethod
async def get_user_by_id(user_id: int):
"""根据 ID 获取用户"""
return await UserCRUD.get_by_id(user_id)
@staticmethod
async def update_user(user_id: int, user: UserUpdate):
"""更新用户"""
# 如果更新密码，需要哈希处理
if user.password:
user.password = get_password_hash(user.password)
return await UserCRUD.update(user_id, user)
@staticmethod
async def delete_user(user_id: int):
"""删除用户"""
return await UserCRUD.delete(user_id)数据访问层
python
# app/api/v1/module_system/user/crud.py
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.database import get_db
from .model import User
from .schema import UserCreate, UserUpdate
from app.core.base_crud import BaseCRUD
class UserCRUD(BaseCRUD):
model = User
@classmethod
async def create(cls, obj_in: UserCreate):
"""创建用户"""
async with get_db() as db:
db_obj = User(**obj_in.dict())
db.add(db_obj)
await db.commit()
await db.refresh(db_obj)
return db_obj
@classmethod
async def get_by_id(cls, id: int):
"""根据 ID 获取用户"""
async with get_db() as db:
return await db.get(cls.model, id)
@classmethod
async def update(cls, id: int, obj_in: UserUpdate):
"""更新用户"""
async with get_db() as db:
db_obj = await db.get(cls.model, id)
if db_obj:
update_data = obj_in.dict(exclude_unset=True)
for field, value in update_data.items():
setattr(db_obj, field, value)
await db.commit()
await db.refresh(db_obj)
return db_obj
@classmethod
async def delete(cls, id: int):
"""删除用户"""
async with get_db() as db:
db_obj = await db.get(cls.model, id)
if db_obj:
await db.delete(db_obj)
await db.commit()
return db_obj🔐 认证与权限
JWT 认证实现
python
# app/core/security.py
from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext
from app.core.setting import settings
# 密码哈希上下文
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
# JWT 配置
ALGORITHM = "HS256"
# 验证密码
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)
# 创建访问令牌
def create_access_token(subject: str, expires_delta: timedelta = None) -> str:
"""创建访问令牌"""
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=ALGORITHM)
return encoded_jwt
# 验证令牌
def verify_token(token: str) -> dict:
"""验证令牌"""
try:
payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[ALGORITHM])
return payload
except JWTError:
return None依赖注入
python
# app/core/dependencies.py
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.database import get_db
from app.core.security import verify_token
from app.api.v1.module_system.user.crud import UserCRUD
# OAuth2 密码承载令牌
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/system/auth/login")
# 获取当前用户
async def get_current_user(token: str = Depends(oauth2_scheme)):
"""获取当前用户"""
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
payload = verify_token(token)
if payload is None:
raise credentials_exception
user_id = payload.get("sub")
if user_id is None:
raise credentials_exception
user = await UserCRUD.get_by_id(int(user_id))
if user is None:
raise credentials_exception
return user
# 获取当前活跃用户
async def get_current_active_user(current_user = Depends(get_current_user)):
"""获取当前活跃用户"""
if not current_user.status:
raise HTTPException(status_code=400, detail="Inactive user")
return current_user
# 获取当前管理员用户
async def get_current_admin_user(current_user = Depends(get_current_active_user)):
"""获取当前管理员用户"""
if not current_user.is_admin:
raise HTTPException(status_code=403, detail="Not enough permissions")
return current_user📊 数据库设计
基础模型
python
# app/core/base_model.py
from sqlalchemy import Column, Integer, DateTime, Boolean
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime
Base = declarative_base()
class BaseModel(Base):
"""基础模型"""
__abstract__ = True
id = Column(Integer, primary_key=True, index=True, comment="主键")
create_time = Column(DateTime, default=datetime.now, comment="创建时间")
update_time = Column(DateTime, default=datetime.now, onupdate=datetime.now, comment="更新时间")
status = Column(Boolean, default=True, comment="状态")核心模型示例
python
# app/api/v1/module_system/user/model.py
from sqlalchemy import Column, String, Integer, Boolean, DateTime, ForeignKey
from sqlalchemy.orm import relationship
from app.core.base_model import BaseModel
class User(BaseModel):
"""用户模型"""
__tablename__ = "sys_user"
username = Column(String(50), unique=True, index=True, nullable=False, comment="用户名")
password = Column(String(100), nullable=False, comment="密码")
nickname = Column(String(50), comment="昵称")
email = Column(String(100), unique=True, index=True, comment="邮箱")
phone = Column(String(20), comment="手机号")
avatar = Column(String(255), comment="头像")
dept_id = Column(Integer, ForeignKey("sys_dept.id"), comment="部门ID")
post_id = Column(Integer, ForeignKey("sys_position.id"), comment="职位ID")
is_admin = Column(Boolean, default=False, comment="是否管理员")
# 关系
dept = relationship("Dept", back_populates="users")
post = relationship("Position", back_populates="users")
roles = relationship("Role", secondary="sys_user_role", back_populates="users")🔄 异步任务
任务管理
python
# app/plugin/module_application/job/service.py
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.jobstores.memory import MemoryJobStore
from apscheduler.executors.asyncio import AsyncIOExecutor
from app.core.setting import settings
# 任务存储
jobstores = {
'default': MemoryJobStore()
}
# 执行器
executors = {
'default': AsyncIOExecutor()
}
# 调度器配置
job_defaults = {
'coalesce': False,
'max_instances': 1
}
# 创建调度器
scheduler = AsyncIOScheduler(
jobstores=jobstores,
executors=executors,
job_defaults=job_defaults
)
# 启动调度器
def start_scheduler():
"""启动调度器"""
if not scheduler.running:
scheduler.start()
# 停止调度器
def stop_scheduler():
"""停止调度器"""
if scheduler.running:
scheduler.shutdown()
# 添加任务
def add_job(func, trigger, id=None, name=None, args=None, kwargs=None, **trigger_args):
"""添加任务"""
return scheduler.add_job(
func,
trigger,
id=id,
name=name,
args=args,
kwargs=kwargs,
**trigger_args
)
# 移除任务
def remove_job(job_id):
"""移除任务"""
scheduler.remove_job(job_id)
# 暂停任务
def pause_job(job_id):
"""暂停任务"""
scheduler.pause_job(job_id)
# 恢复任务
def resume_job(job_id):
"""恢复任务"""
scheduler.resume_job(job_id)
# 获取任务
def get_job(job_id):
"""获取任务"""
return scheduler.get_job(job_id)
# 获取所有任务
def get_jobs():
"""获取所有任务"""
return scheduler.get_jobs()🧪 测试
测试配置
python
# tests/conftest.py
import pytest
from fastapi.testclient import TestClient
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
from app.main import app
from app.core.database import Base, get_db
from app.core.setting import settings
# 创建测试数据库引擎
engine = create_async_engine(
settings.TEST_DATABASE_URL,
echo=True
)
# 创建测试会话工厂
TestingSessionLocal = sessionmaker(
engine,
class_=AsyncSession,
expire_on_commit=False
)
# 覆盖依赖项
async def override_get_db():
async with TestingSessionLocal() as session:
yield session
app.dependency_overrides[get_db] = override_get_db
# 测试客户端
@pytest.fixture
def client():
"""测试客户端"""
return TestClient(app)
# 测试数据库会话
@pytest.fixture
async def db():
"""测试数据库会话"""
async with TestingSessionLocal() as session:
yield session
# 初始化测试数据库
@pytest.fixture(scope="module", autouse=True)
async def init_db():
"""初始化测试数据库"""
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
yield
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)API 测试示例
python
# tests/test_user.py
import pytest
from fastapi.testclient import TestClient
from app.main import app
client = TestClient(app)
# 测试登录
def test_login():
"""测试登录"""
response = client.post(
"/api/v1/system/auth/login",
data={
"username": "admin",
"password": "123456"
}
)
assert response.status_code == 200
assert "access_token" in response.json()
return response.json()["access_token"]
# 测试获取用户列表
def test_get_users():
"""测试获取用户列表"""
token = test_login()
response = client.get(
"/api/v1/system/user",
headers={"Authorization": f"Bearer {token}"}
)
assert response.status_code == 200
assert isinstance(response.json(), list)
# 测试创建用户
def test_create_user():
"""测试创建用户"""
token = test_login()
response = client.post(
"/api/v1/system/user",
headers={"Authorization": f"Bearer {token}"},
json={
"username": "testuser",
"password": "123456",
"nickname": "测试用户",
"email": "test@example.com",
"phone": "13800138000",
"dept_id": 1,
"post_id": 1
}
)
assert response.status_code == 200
assert response.json()["username"] == "testuser"🚀 部署
Docker 部署
dockerfile
# Dockerfile
FROM python:3.10-slim
WORKDIR /app
# 安装系统依赖
RUN apt-get update && apt-get install -y --no-install-recommends \
gcc \
python3-dev \
default-libmysqlclient-dev \
&& rm -rf /var/lib/apt/lists/*
# 复制依赖文件
COPY requirements.txt .
# 安装 Python 依赖
RUN pip install --no-cache-dir -r requirements.txt
# 复制应用代码
COPY . .
# 暴露端口
EXPOSE 8001
# 启动应用
CMD ["python", "main.py", "run"]Docker Compose 部署
yaml
# docker-compose.yml
version: '3.8'
services:
backend:
build: ./backend
ports:
- "8001:8001"
depends_on:
- mysql
- redis
environment:
- ENV=prod
- DATABASE_URL=mysql+pymysql://root:123456@mysql:3306/fastapi_admin
- REDIS_URL=redis://redis:6379/0
volumes:
- ./backend:/app
restart: always
mysql:
image: mysql:8.0
environment:
MYSQL_ROOT_PASSWORD: 123456
MYSQL_DATABASE: fastapi_admin
volumes:
- mysql_data:/var/lib/mysql
restart: always
redis:
image: redis:7.0
restart: always
volumes:
mysql_data:📈 性能优化
数据库优化
索引优化
- 为常用查询字段添加索引
- 避免过多索引影响写入性能
- 使用复合索引优化多字段查询
查询优化
- 使用
selectinload或joinedload避免 N+1 查询 - 只查询必要字段，避免
SELECT * - 使用分页查询，避免一次性加载大量数据
- 使用
连接池优化
- 合理配置数据库连接池大小
- 使用
asyncpg或aiomysql等异步数据库驱动
缓存优化
Redis 缓存
- 缓存热点数据
- 使用合适的缓存过期策略
- 实现缓存预热
缓存键设计
- 使用前缀区分不同类型的缓存
- 包含相关参数，确保缓存唯一性
- 避免过长的缓存键
异步处理
异步路由
- 使用
async def定义异步路由 - 避免在异步函数中使用同步阻塞操作
- 使用
后台任务
- 对于耗时操作，使用
BackgroundTasks - 对于更复杂的异步任务，使用 Celery 或 APScheduler
- 对于耗时操作，使用
并发控制
- 使用
asyncio.Semaphore控制并发数量 - 避免过多并发导致系统负载过高
- 使用
代码优化
代码结构
- 合理使用装饰器提取重复代码
- 使用上下文管理器管理资源
- 避免过长的函数和类
异常处理
- 合理捕获和处理异常
- 避免裸
except语句 - 记录异常信息便于调试
依赖管理
- 定期更新依赖包
- 避免不必要的依赖
- 使用
poetry或pip-tools管理依赖
🚀 快速开始
环境要求
- Python: >= 3.10
- MySQL: >= 8.0
- Redis: >= 7.0
- Git: >= 2.0.0
项目启动步骤
克隆项目
bashgit clone https://gitee.com/fastapiadmin/FastapiAdmin.git进入后端目录
bashcd FastapiAdmin/backend安装依赖
bashpip install -r requirements.txt配置环境变量
bashcp env/.env.dev.example env/.env.dev # 编辑 env/.env.dev 文件，配置数据库和Redis连接信息初始化数据库
bash# 生成迁移文件 python main.py revision "初始化迁移" --env=dev # 应用迁移 python main.py upgrade --env=dev # 初始化系统数据 python main.py init启动开发服务器
bashpython main.py run --env=dev访问 API 文档
- Swagger UI: http://localhost:8001/api/v1/docs
- Redoc: http://localhost:8001/api/v1/redoc
默认账号
- 用户名:
admin - 密码:
123456
- 用户名:
🎯 开发规范
代码风格
命名规范
- 类名:
PascalCase - 函数名:
snake_case - 变量名:
snake_case - 常量:
UPPER_SNAKE_CASE
- 类名:
代码格式
- 使用
black格式化代码 - 使用
isort排序导入 - 每行不超过 88 个字符
- 使用
注释规范
- 使用文档字符串记录函数和类
- 复杂逻辑添加注释
- 避免冗余注释
架构规范
分层架构
- 严格遵循分层架构
- 控制器只处理 HTTP 请求和响应
- 业务逻辑放在服务层
- 数据访问放在 CRUD 层
依赖注入
- 使用 FastAPI 的依赖注入系统
- 避免硬编码依赖
- 便于测试和扩展
错误处理
- 使用统一的错误响应格式
- 合理使用 HTTP 状态码
- 记录错误日志
安全规范
密码安全
- 使用
bcrypt哈希密码 - 避免存储明文密码
- 实现密码强度检查
- 使用
认证授权
- 使用 JWT 进行无状态认证
- 实现细粒度的权限控制
- 定期轮换密钥
输入验证
- 使用 Pydantic 模型验证输入
- 避免 SQL 注入和 XSS 攻击
- 对用户输入进行清洗和验证
📚 学习资源
官方文档
推荐教程
工具推荐
- IDE: PyCharm 或 VS Code + Python 扩展
- 数据库工具: DBeaver 或 MySQL Workbench
- API 测试: Postman 或 Insomnia
- 性能分析: cProfile 或 py-spy
通过本指南，您应该能够快速上手并开发出功能完善的后端服务。在开发过程中，建议遵循项目规范，合理利用现有组件和工具，以提高开发效率和代码质量。
🎯 总结
FastApiAdmin 后端项目提供了一套完整的企业级后台管理系统后端解决方案，具有以下优势：
- 高性能异步：基于 FastAPI 异步框架，提供卓越的性能表现
- 安全可靠：JWT 认证机制和 RBAC 权限控制模型，保障系统安全
- 易于扩展：模块化设计，清晰的分层架构，便于功能扩展
- 开发友好：完善的类型检查、自动文档生成和丰富的开发工具
- 生产就绪：支持 Docker 部署、数据库迁移和性能监控
通过本指南，您应该能够快速上手并开发出功能完善的后端服务。在开发过程中，建议遵循项目规范，合理利用现有组件和工具，以提高开发效率和代码质量。
这个后端项目结构体现了现代 Python 后端项目的标准架构，具有良好的可维护性、可扩展性和开发体验。