全栈项目实战:任务管理系统

🏷️ L3 📊 advanced ⏱️ 60分钟 🏷️ 全栈,项目实战,FastAPI,Vue3,CRUD,前沿

# 全栈项目实战:任务管理系统

概述

在当今敏捷开发与远程协作盛行的时代,任务管理系统是每个团队必备的核心工具。本课程旨在带你从0到1,使用现代全栈技术栈(FastAPI + SQLAlchemy + Vue3 + Pinia)构建一个功能完备的任务管理系统。你将深入理解前后端分离架构,掌握RESTful API设计、用户认证(JWT)、数据持久化与前端状态管理等关键技能。学完本课程,你将具备独立开发中小型全栈应用的能力,并能将这套模式迁移至其他业务场景。

一、核心知识讲解

1. FastAPI异步框架与RESTful API设计

原理说明: 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.pyapp/schemas.pyapp/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_tokenrefresh_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》书籍。

在博海学习网开始学习 →