# 全栈项目实战:任务管理系统
在当今敏捷开发与远程协作盛行的时代,任务管理系统是每个团队必备的核心工具。本课程旨在带你从0到1,使用现代全栈技术栈(FastAPI + SQLAlchemy + Vue3 + Pinia)构建一个功能完备的任务管理系统。你将深入理解前后端分离架构,掌握RESTful API设计、用户认证(JWT)、数据持久化与前端状态管理等关键技能。学完本课程,你将具备独立开发中小型全栈应用的能力,并能将这套模式迁移至其他业务场景。
原理说明: FastAPI基于Python的异步特性(async/await)和类型提示(Type Hints)构建,能自动生成交互式API文档(Swagger UI)。RESTful API通过HTTP动词(GET、POST、PUT、DELETE)对资源进行操作,是前后端通信的标准模式。
典型示例: 为任务资源设计一个获取列表的API端点。
关键代码:
# app/main.py
from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel
from typing import List, Optional
from datetime import datetime
app = FastAPI(title="Task Manager API")
# Pydantic模型用于请求/响应验证
class TaskCreate(BaseModel):
title: str
description: Optional[str] = None
due_date: Optional[datetime] = None
class Task(TaskCreate):
id: int
completed: bool = False
owner_id: int
class Config:
orm_mode = True # 允许从ORM模型转换
# 模拟内存数据库
fake_db = {}
task_id_counter = 0
@app.get("/tasks", response_model=List[Task])
async def get_tasks():
"""获取所有任务列表"""
return list(fake_db.values())
@app.post("/tasks", response_model=Task, status_code=201)
async def create_task(task: TaskCreate):
"""创建新任务"""
global task_id_counter
task_id_counter += 1
new_task = Task(id=task_id_counter, **task.dict(), owner_id=1)
fake_db[task_id_counter] = new_task
return new_task
@app.get("/tasks/{task_id}", response_model=Task)
async def get_task(task_id: int):
"""获取单个任务"""
if task_id not in fake_db:
raise HTTPException(status_code=404, detail="Task not found")
return fake_db[task_id]
2. SQLAlchemy ORM与数据库迁移
原理说明: SQLAlchemy是Python最强大的ORM(对象关系映射)库,它将数据库表映射为Python类,将行记录映射为对象。Alembic是其配套的数据库迁移工具,用于管理数据库结构的版本变更,避免手动修改表结构。
典型示例: 定义User和Task模型,并建立一对多关系。
关键代码:
# app/models.py
from sqlalchemy import Column, Integer, String, Boolean, DateTime, ForeignKey
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime
Base = declarative_base()
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
username = Column(String(50), unique=True, index=True, nullable=False)
hashed_password = Column(String(255), nullable=False)
tasks = relationship("Task", back_populates="owner") # 建立关系
class Task(Base):
__tablename__ = "tasks"
id = Column(Integer, primary_key=True, index=True)
title = Column(String(200), nullable=False)
description = Column(String(1000))
completed = Column(Boolean, default=False)
due_date = Column(DateTime)
owner_id = Column(Integer, ForeignKey("users.id"), nullable=False)
owner = relationship("User", back_populates="tasks")
3. JWT用户认证与授权
原理说明: JWT(JSON Web Token)是一种无状态的认证机制。用户登录成功后,服务端签发一个包含用户信息的加密Token,客户端在后续请求中携带此Token,服务端验证Token的有效性即可识别用户身份。
典型示例: 实现登录接口和Token验证依赖项。
关键代码:
# app/auth.py
from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
SECRET_KEY = "your-secret-key-keep-it-secret"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
def create_access_token(data: dict):
to_encode = data.copy()
expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
to_encode.update({"exp": expire})
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
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"},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub")
if username is None:
raise credentials_exception
except JWTError:
raise credentials_exception
# 从数据库查询用户
user = await get_user_by_username(username) # 假设已实现
if user is None:
raise credentials_exception
return user
4. Vue3组合式API与Pinia状态管理
原理说明: Vue3的组合式API(Composition API)允许将逻辑按功能组织成可复用的函数(composables),替代了Vue2的Options API。Pinia是Vue3官方推荐的状态管理库,比Vuex更简洁,天然支持TypeScript和组合式API。
典型示例: 创建任务模块的Pinia Store。
关键代码:
// src/stores/taskStore.js
import { defineStore } from 'pinia'
import { ref } from 'vue'
import axios from 'axios'
export const useTaskStore = defineStore('task', () => {
// 状态
const tasks = ref([])
const loading = ref(false)
const error = ref(null)
// 获取任务列表
async function fetchTasks() {
loading.value = true
try {
const response = await axios.get('/api/tasks')
tasks.value = response.data
} catch (err) {
error.value = err.message
} finally {
loading.value = false
}
}
// 创建任务
async function createTask(taskData) {
try {
const response = await axios.post('/api/tasks', taskData)
tasks.value.push(response.data)
return response.data
} catch (err) {
error.value = err.message
throw err
}
}
// 更新任务
async function updateTask(taskId, taskData) {
try {
const response = await axios.put(/api/tasks/${taskId}, taskData)
const index = tasks.value.findIndex(t => t.id === taskId)
if (index !== -1) {
tasks.value[index] = response.data
}
return response.data
} catch (err) {
error.value = err.message
throw err
}
}
return { tasks, loading, error, fetchTasks, createTask, updateTask }
})
5. 前后端联调与跨域问题
原理说明: 前后端分离架构中,前端(Vue3)和后端(FastAPI)运行在不同端口或域名下,浏览器出于安全考虑会阻止跨域HTTP请求。CORS(跨域资源共享)通过HTTP头允许服务器声明哪些源可以访问资源。
典型示例: 在FastAPI中配置CORS中间件。
关键代码:
# app/main.py (续)
from fastapi.middleware.cors import CORSMiddleware
# 允许的源列表
origins = [
"http://localhost:5173", # Vue3开发服务器默认端口
"http://127.0.0.1:5173",
"http://localhost:3000", # 生产环境可能的前端端口
]
app.add_middleware(
CORSMiddleware,
allow_origins=origins, # 生产环境应设置为具体域名
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
二、实操步骤
步骤1:项目初始化与依赖安装
1. 创建后端项目:
mkdir task-manager && cd task-manager
python -m venv venv
source venv/bin/activate # Windows: venv\Scripts\activate
pip install fastapi uvicorn sqlalchemy alembic python-jose passlib[bcrypt] python-multipart
2. 创建前端项目:
npm create vue@latest frontend
cd frontend
npm install pinia axios vue-router@4
步骤2:搭建后端API基础架构
1. 创建目录结构:app/、app/models.py、app/schemas.py、app/database.py
2. 配置数据库连接(SQLite用于开发):
# app/database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from .models import Base
SQLALCHEMY_DATABASE_URL = "sqlite:///./tasks.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
def init_db():
Base.metadata.create_all(bind=engine)
3. 实现完整的CRUD API(参考核心知识部分),添加JWT认证中间件。
步骤3:开发Vue3前端应用
1. 使用Vue Router配置路由:/login、/tasks、/tasks/new
2. 创建Pinia Store管理用户认证和任务数据
3. 实现登录页面、任务列表页面、任务创建/编辑页面
4. 使用axios.interceptors自动添加JWT Token到请求头:
// src/utils/axios.js
import axios from 'axios'
import { useAuthStore } from '@/stores/authStore'
const instance = axios.create({ baseURL: 'http://localhost:8000/api' })
instance.interceptors.request.use(config => {
const authStore = useAuthStore()
if (authStore.token) {
config.headers.Authorization = Bearer ${authStore.token}
}
return config
})
步骤4:前后端联调测试
1. 启动后端:uvicorn app.main:app --reload
2. 启动前端:npm run dev
3. 测试完整流程:注册/登录 → 创建任务 → 修改任务状态 → 删除任务
三、常见问题与故障排查
问题1:数据库迁移冲突
现象: Alembic执行alembic upgrade head时报错,提示表已存在或字段不匹配。
解决方法: 删除alembic/versions/目录下的旧迁移文件,重新生成:alembic revision --autogenerate -m "initial",然后执行升级。
问题2:CORS跨域请求被阻止
现象: 前端调用API时浏览器控制台报CORS错误。
解决方法: 检查FastAPI的origins列表是否包含前端URL,注意端口号(Vue3开发默认5173)。如果使用代理,确保后端正确返回Access-Control-Allow-Origin头。
问题3:JWT Token过期后未自动刷新
现象: 用户操作一段时间后,所有请求返回401未授权。
解决方法: 实现Token刷新机制:在登录时返回access_token和refresh_token,使用refresh_token在后台静默获取新的access_token。在axios拦截器中捕获401错误并尝试刷新。
问题4:Pinia状态在页面刷新后丢失
现象: 刷新浏览器后,已登录状态消失,任务列表数据丢失。
解决方法: 使用Pinia的persist插件(如pinia-plugin-persistedstate)或手动将关键状态(如Token)保存到localStorage,在Store初始化时恢复。
问题5:FastAPI异步操作阻塞
现象: 多个并发请求处理缓慢,甚至超时。
解决方法: 确保所有数据库操作使用异步SQLAlchemy(AsyncSession),或使用async def配合await。对于同步ORM操作,使用run_in_executor在后台线程池执行。
四、总结与扩展学习
核心要点总结
- **全栈架构**:FastAPI负责API层、SQLAlchemy负责数据层、Vue3负责表现层、Pinia负责状态层,形成了清晰的关注点分离。
- **认证流程**:JWT无状态认证结合OAuth2密码流,实现了安全、可扩展的用户管理。
- **状态管理**:Pinia的组合式API让状态逻辑更集中、更易测试,与Vue3的组合式API天然契合。
- **联调技巧**:CORS配置、axios拦截器、统一的错误处理是前后端协作的关键基础设施。
扩展学习方向
1. 部署与运维:学习使用Docker容器化应用,部署到云服务器(如阿里云ECS)或使用Vercel/Netlify托管前端。
2. 性能优化:为API添加缓存(Redis)、数据库索引优化、前端代码分割与懒加载。
3. 功能增强:添加任务分类、标签系统、文件附件上传、邮件通知、WebSocket实时协作。
4. 测试驱动开发:使用pytest编写后端API测试,使用Vitest和Cypress进行前端单元测试和E2E测试。
5. 微服务演进:将认证、任务管理、通知拆分为独立服务,使用消息队列(RabbitMQ)进行服务间通信。
推荐学习资源:FastAPI官方文档、Vue3官方教程、SQLAlchemy 2.0文档、《全栈应用开发:使用Vue和FastAPI》书籍。