解密FastAPI与React前后端分离架构如何助力开发者快速构建高性能现代化Web应用从技术选型到项目实战全解析

1. 引言：前后端分离架构的崛起

在当今快速发展的Web开发领域，前后端分离架构已成为构建现代化Web应用的主流选择。这种架构模式通过将前端和后端的职责明确分离，使开发团队能够并行工作、独立部署，并最终实现更高效、更灵活的开发流程。本文将深入探讨FastAPI与React这一强大组合如何助力开发者快速构建高性能的现代化Web应用，从技术选型到项目实战进行全面解析。

2. 技术选型：为什么选择FastAPI与React

2.1 FastAPI的优势

FastAPI是一个现代、快速（高性能）的Web框架，用于构建API，基于Python 3.6+的类型提示。它的主要优势包括：

高性能：FastAPI的性能与NodeJS和Go相当，这得益于Starlette（用于Web部分）和Pydantic（用于数据部分）的底层设计。
快速开发：提供直观的编辑器支持，减少约40%的开发错误，并提高开发速度。
自动文档生成：基于OpenAPI（以前称为Swagger）和JSON Schema，自动生成交互式API文档。
类型提示：利用Python的类型提示进行数据验证和序列化，减少代码错误。

# FastAPI简单示例 from fastapi import FastAPI from pydantic import BaseModel app = FastAPI() class Item(BaseModel): name: str description: str = None price: float tax: float = 10.5 @app.post("/items/") async def create_item(item: Item): return {"item": item}

2.2 React的优势

React是一个用于构建用户界面的JavaScript库，由Facebook开发并维护。它的主要优势包括：

组件化：通过组件化开发，使UI构建更加模块化和可重用。
虚拟DOM：通过虚拟DOM提高应用性能，减少实际DOM操作。
单向数据流：使数据管理更加可预测，便于调试。
丰富的生态系统：拥有庞大的社区和丰富的第三方库支持。

// React简单示例 import React, { useState } from 'react'; function ItemForm() { const [item, setItem] = useState({ name: '', description: '', price: 0, tax: 10.5 }); const handleSubmit = async (e) => { e.preventDefault(); const response = await fetch('/items/', { method: 'POST', headers: { 'Content-Type': 'application/json', }, body: JSON.stringify(item), }); const data = await response.json(); console.log('Success:', data); }; return ( <form onSubmit={handleSubmit}> <input type="text" value={item.name} onChange={(e) => setItem({...item, name: e.target.value})} placeholder="Item name" /> <input type="text" value={item.description} onChange={(e) => setItem({...item, description: e.target.value})} placeholder="Description" /> <input type="number" value={item.price} onChange={(e) => setItem({...item, price: parseFloat(e.target.value)})} placeholder="Price" /> <button type="submit">Submit</button> </form> ); }

2.3 FastAPI与React的协同效应

FastAPI和React的结合提供了以下协同效应：

开发效率：前后端可以并行开发，API文档自动生成，前端开发者可以提前了解API结构。
性能优化：FastAPI提供高性能后端服务，React通过虚拟DOM优化前端渲染。
类型安全：FastAPI的类型提示可以与React的TypeScript无缝集成，提供端到端的类型安全。
现代化开发体验：两者都支持热重载，提供快速的开发反馈循环。

3. 项目架构设计

3.1 整体架构

一个典型的FastAPI与React前后端分离架构如下：

├── backend/ # FastAPI后端 │ ├── app/ │ │ ├── __init__.py │ │ ├── main.py # FastAPI应用入口 │ │ ├── models/ # 数据模型 │ │ ├── schemas/ # Pydantic模型 │ │ ├── api/ # API路由 │ │ ├── core/ # 核心配置 │ │ └── db/ # 数据库相关 │ ├── requirements.txt # Python依赖 │ └── alembic/ # 数据库迁移 │ ├── frontend/ # React前端 │ ├── public/ │ ├── src/ │ │ ├── components/ # React组件 │ │ ├── pages/ # 页面组件 │ │ ├── services/ # API服务 │ │ ├── hooks/ # 自定义Hooks │ │ ├── contexts/ # React Context │ │ └── utils/ # 工具函数 │ ├── package.json # Node.js依赖 │ └── ... # 其他配置文件 │ └── docker-compose.yml # 容器编排

3.2 后端架构设计

FastAPI后端采用分层架构，主要包括：

API层：处理HTTP请求，调用业务逻辑层。
业务逻辑层：实现具体的业务逻辑。
数据访问层：与数据库交互，执行CRUD操作。
模型层：定义数据结构和关系。

# backend/app/main.py from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from app.api.api_v1.api import api_router from app.core.config import settings app = FastAPI( title=settings.PROJECT_NAME, openapi_url=f"{settings.API_V1_STR}/openapi.json" ) # 设置CORS app.add_middleware( CORSMiddleware, allow_origins=["http://localhost:3000"], # React前端地址 allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) app.include_router(api_router, prefix=settings.API_V1_STR) @app.get("/") async def root(): return {"message": "Hello World"}

3.3 前端架构设计

React前端采用组件化架构，主要包括：

页面组件：对应应用的不同页面。
UI组件：可重用的UI元素。
服务层：与后端API交互。
状态管理：使用Context API或Redux等管理应用状态。

// frontend/src/App.js import React from 'react'; import { BrowserRouter as Router, Routes, Route } from 'react-router-dom'; import { AuthProvider } from './contexts/AuthContext'; import Layout from './components/Layout'; import HomePage from './pages/HomePage'; import LoginPage from './pages/LoginPage'; import DashboardPage from './pages/DashboardPage'; import ProtectedRoute from './components/ProtectedRoute'; function App() { return ( <AuthProvider> <Router> <Layout> <Routes> <Route path="/" element={<HomePage />} /> <Route path="/login" element={<LoginPage />} /> <Route path="/dashboard" element={ <ProtectedRoute> <DashboardPage /> </ProtectedRoute> } /> </Routes> </Layout> </Router> </AuthProvider> ); } export default App;

4. 项目实战：构建一个任务管理应用

让我们通过构建一个任务管理应用来演示FastAPI与React的实战应用。

4.1 后端实现

4.1.1 数据模型

首先，我们定义任务的数据模型：

# backend/app/models/task.py from sqlalchemy import Column, Integer, String, Boolean, DateTime, ForeignKey from sqlalchemy.orm import relationship from sqlalchemy.sql import func from app.db.base_class import Base class Task(Base): __tablename__ = "tasks" id = Column(Integer, primary_key=True, index=True) title = Column(String, index=True) description = Column(String, index=True) completed = Column(Boolean, default=False) owner_id = Column(Integer, ForeignKey("users.id")) created_at = Column(DateTime(timezone=True), server_default=func.now()) updated_at = Column(DateTime(timezone=True), onupdate=func.now()) owner = relationship("User", back_populates="tasks")

4.1.2 Pydantic模型

定义用于数据验证和序列化的Pydantic模型：

# backend/app/schemas/task.py from typing import Optional from datetime import datetime from pydantic import BaseModel class TaskBase(BaseModel): title: str description: Optional[str] = None completed: bool = False class TaskCreate(TaskBase): pass class TaskUpdate(TaskBase): pass class TaskInDBBase(TaskBase): id: int owner_id: int created_at: datetime updated_at: Optional[datetime] = None class Config: orm_mode = True class Task(TaskInDBBase): pass class TaskInDB(TaskInDBBase): pass

4.1.3 API路由

创建任务相关的API路由：

# backend/app/api/api_v1/endpoints/tasks.py from typing import List from fastapi import APIRouter, Depends, HTTPException from sqlalchemy.orm import Session from app.api import deps from app.schemas.task import Task, TaskCreate, TaskUpdate from app.crud import crud_task router = APIRouter() @router.get("/", response_model=List[Task]) def read_tasks( skip: int = 0, limit: int = 100, db: Session = Depends(deps.get_db), current_user: int = Depends(deps.get_current_active_user), ): """ 检索任务列表 """ tasks = crud_task.get_multi_by_owner( db=db, owner_id=current_user.id, skip=skip, limit=limit ) return tasks @router.post("/", response_model=Task) def create_task( *, db: Session = Depends(deps.get_db), task_in: TaskCreate, current_user: int = Depends(deps.get_current_active_user), ): """ 创建新任务 """ task = crud_task.create_with_owner(db=db, obj_in=task_in, owner_id=current_user.id) return task @router.put("/{id}", response_model=Task) def update_task( *, db: Session = Depends(deps.get_db), id: int, task_in: TaskUpdate, current_user: int = Depends(deps.get_current_active_user), ): """ 更新任务 """ task = crud_task.get(db=db, id=id) if not task: raise HTTPException(status_code=404, detail="Task not found") if task.owner_id != current_user.id: raise HTTPException(status_code=400, detail="Not enough permissions") task = crud_task.update(db=db, db_obj=task, obj_in=task_in) return task @router.delete("/{id}", response_model=Task) def delete_task( *, db: Session = Depends(deps.get_db), id: int, current_user: int = Depends(deps.get_current_active_user), ): """ 删除任务 """ task = crud_task.get(db=db, id=id) if not task: raise HTTPException(status_code=404, detail="Task not found") if task.owner_id != current_user.id: raise HTTPException(status_code=400, detail="Not enough permissions") task = crud_task.remove(db=db, id=id) return task

4.1.4 CRUD操作

实现基本的CRUD操作：

# backend/app/crud/crud_task.py from typing import Any, Dict, List, Optional, Union from sqlalchemy.orm import Session from app.crud.base import CRUDBase from app.models.task import Task from app.schemas.task import TaskCreate, TaskUpdate class CRUDTask(CRUDBase[Task, TaskCreate, TaskUpdate]): def create_with_owner( self, db: Session, *, obj_in: TaskCreate, owner_id: int ) -> Task: obj_in_data = obj_in.dict() db_obj = Task(**obj_in_data, owner_id=owner_id) db.add(db_obj) db.commit() db.refresh(db_obj) return db_obj def get_multi_by_owner( self, db: Session, *, owner_id: int, skip: int = 0, limit: int = 100 ) -> List[Task]: return ( db.query(self.model) .filter(Task.owner_id == owner_id) .offset(skip) .limit(limit) .all() ) crud_task = CRUDTask(Task)

4.2 前端实现

4.2.1 API服务

创建与后端API交互的服务：

// frontend/src/services/taskService.js import axios from 'axios'; const API_URL = 'http://localhost:8000/api/v1'; // 获取所有任务 export const getTasks = async (token) => { try { const response = await axios.get(`${API_URL}/tasks/`, { headers: { 'Authorization': `Bearer ${token}` } }); return response.data; } catch (error) { console.error('Error fetching tasks:', error); throw error; } }; // 创建新任务 export const createTask = async (task, token) => { try { const response = await axios.post(`${API_URL}/tasks/`, task, { headers: { 'Authorization': `Bearer ${token}`, 'Content-Type': 'application/json' } }); return response.data; } catch (error) { console.error('Error creating task:', error); throw error; } }; // 更新任务 export const updateTask = async (id, task, token) => { try { const response = await axios.put(`${API_URL}/tasks/${id}`, task, { headers: { 'Authorization': `Bearer ${token}`, 'Content-Type': 'application/json' } }); return response.data; } catch (error) { console.error('Error updating task:', error); throw error; } }; // 删除任务 export const deleteTask = async (id, token) => { try { const response = await axios.delete(`${API_URL}/tasks/${id}`, { headers: { 'Authorization': `Bearer ${token}` } }); return response.data; } catch (error) { console.error('Error deleting task:', error); throw error; } };

4.2.2 任务列表组件

创建显示任务列表的组件：

// frontend/src/components/TaskList.js import React, { useState, useEffect, useContext } from 'react'; import { AuthContext } from '../contexts/AuthContext'; import { getTasks, deleteTask } from '../services/taskService'; import TaskItem from './TaskItem'; import TaskForm from './TaskForm'; import './TaskList.css'; const TaskList = () => { const [tasks, setTasks] = useState([]); const [loading, setLoading] = useState(true); const [error, setError] = useState(null); const [showForm, setShowForm] = useState(false); const { auth } = useContext(AuthContext); useEffect(() => { fetchTasks(); }, []); const fetchTasks = async () => { try { setLoading(true); const data = await getTasks(auth.token); setTasks(data); setError(null); } catch (err) { setError('Failed to fetch tasks. Please try again later.'); console.error(err); } finally { setLoading(false); } }; const handleDelete = async (id) => { try { await deleteTask(id, auth.token); setTasks(tasks.filter(task => task.id !== id)); } catch (err) { setError('Failed to delete task. Please try again later.'); console.error(err); } }; const handleTaskCreated = (newTask) => { setTasks([...tasks, newTask]); setShowForm(false); }; const handleTaskUpdated = (updatedTask) => { setTasks(tasks.map(task => task.id === updatedTask.id ? updatedTask : task)); }; return ( <div className="task-list-container"> <div className="task-list-header"> <h2>My Tasks</h2> <button className="btn btn-primary" onClick={() => setShowForm(!showForm)} > {showForm ? 'Cancel' : 'Add New Task'} </button> </div> {showForm && ( <TaskForm onTaskCreated={handleTaskCreated} /> )} {loading ? ( <p>Loading tasks...</p> ) : error ? ( <div className="error-message">{error}</div> ) : tasks.length === 0 ? ( <p>No tasks found. Create your first task!</p> ) : ( <div className="task-list"> {tasks.map(task => ( <TaskItem key={task.id} task={task} onDelete={handleDelete} onTaskUpdated={handleTaskUpdated} /> ))} </div> )} </div> ); }; export default TaskList;

4.2.3 任务项组件

创建单个任务项的组件：

// frontend/src/components/TaskItem.js import React, { useState } from 'react'; import { updateTask } from '../services/taskService'; import './TaskItem.css'; const TaskItem = ({ task, onDelete, onTaskUpdated }) => { const [isEditing, setIsEditing] = useState(false); const [title, setTitle] = useState(task.title); const [description, setDescription] = useState(task.description); const [completed, setCompleted] = useState(task.completed); const [loading, setLoading] = useState(false); const handleSave = async () => { try { setLoading(true); const updatedTask = await updateTask(task.id, { title, description, completed }, localStorage.getItem('token')); onTaskUpdated(updatedTask); setIsEditing(false); } catch (error) { console.error('Error updating task:', error); alert('Failed to update task. Please try again.'); } finally { setLoading(false); } }; const handleToggleComplete = async () => { try { setLoading(true); const updatedTask = await updateTask(task.id, { ...task, completed: !completed }, localStorage.getItem('token')); onTaskUpdated(updatedTask); setCompleted(!completed); } catch (error) { console.error('Error updating task:', error); alert('Failed to update task. Please try again.'); } finally { setLoading(false); } }; return ( <div className={`task-item ${completed ? 'completed' : ''}`}> {isEditing ? ( <div className="task-edit-form"> <input type="text" value={title} onChange={(e) => setTitle(e.target.value)} className="task-title-input" placeholder="Task title" /> <textarea value={description} onChange={(e) => setDescription(e.target.value)} className="task-description-input" placeholder="Task description" /> <div className="task-actions"> <button onClick={handleSave} disabled={loading} className="btn btn-save" > {loading ? 'Saving...' : 'Save'} </button> <button onClick={() => setIsEditing(false)} className="btn btn-cancel" > Cancel </button> </div> </div> ) : ( <div className="task-content"> <div className="task-header"> <h3 className="task-title">{task.title}</h3> <input type="checkbox" checked={completed} onChange={handleToggleComplete} disabled={loading} className="task-checkbox" /> </div> {task.description && ( <p className="task-description">{task.description}</p> )} <div className="task-meta"> <span className="task-date"> Created: {new Date(task.created_at).toLocaleDateString()} </span> {task.updated_at && ( <span className="task-date"> Updated: {new Date(task.updated_at).toLocaleDateString()} </span> )} </div> <div className="task-actions"> <button onClick={() => setIsEditing(true)} className="btn btn-edit" > Edit </button> <button onClick={() => onDelete(task.id)} className="btn btn-delete" > Delete </button> </div> </div> )} </div> ); }; export default TaskItem;

4.2.4 任务表单组件

创建用于添加新任务的表单组件：

// frontend/src/components/TaskForm.js import React, { useState } from 'react'; import { createTask } from '../services/taskService'; import './TaskForm.css'; const TaskForm = ({ onTaskCreated }) => { const [title, setTitle] = useState(''); const [description, setDescription] = useState(''); const [loading, setLoading] = useState(false); const [error, setError] = useState(null); const handleSubmit = async (e) => { e.preventDefault(); if (!title.trim()) { setError('Title is required'); return; } try { setLoading(true); setError(null); const newTask = await createTask({ title, description, completed: false }, localStorage.getItem('token')); onTaskCreated(newTask); setTitle(''); setDescription(''); } catch (err) { setError('Failed to create task. Please try again.'); console.error(err); } finally { setLoading(false); } }; return ( <div className="task-form-container"> <h3>Create New Task</h3> {error && <div className="error-message">{error}</div>} <form onSubmit={handleSubmit} className="task-form"> <div className="form-group"> <label htmlFor="title">Title *</label> <input type="text" id="title" value={title} onChange={(e) => setTitle(e.target.value)} placeholder="Enter task title" disabled={loading} /> </div> <div className="form-group"> <label htmlFor="description">Description</label> <textarea id="description" value={description} onChange={(e) => setDescription(e.target.value)} placeholder="Enter task description" disabled={loading} rows={3} /> </div> <div className="form-actions"> <button type="submit" disabled={loading} className="btn btn-primary" > {loading ? 'Creating...' : 'Create Task'} </button> </div> </form> </div> ); }; export default TaskForm;

5. 性能优化策略

5.1 后端性能优化

5.1.1 异步处理

FastAPI支持异步请求处理，可以显著提高并发性能：

# backend/app/main.py from fastapi import FastAPI import asyncio import time app = FastAPI() @app.get("/sync") def sync_endpoint(): time.sleep(1) # 阻塞操作 return {"message": "Sync endpoint"} @app.get("/async") async def async_endpoint(): await asyncio.sleep(1) # 非阻塞操作 return {"message": "Async endpoint"}

5.1.2 数据库优化

使用数据库连接池和适当的索引可以提高数据库性能：

# backend/app/core/config.py from pydantic import BaseSettings class Settings(BaseSettings): PROJECT_NAME: str = "Task Manager" DATABASE_URL: str = "postgresql://user:password@postgresserver/db" # 数据库连接池配置 POOL_SIZE: int = 10 MAX_OVERFLOW: int = 20 POOL_TIMEOUT: int = 30 POOL_RECYCLE: int = 3600 class Config: env_file = ".env" # backend/app/db/session.py from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker from app.core.config import settings engine = create_engine( settings.DATABASE_URL, pool_size=settings.POOL_SIZE, max_overflow=settings.MAX_OVERFLOW, pool_timeout=settings.POOL_TIMEOUT, pool_recycle=settings.POOL_RECYCLE, ) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) Base = declarative_base()

5.1.3 缓存策略

实现缓存可以减少数据库查询，提高响应速度：

# backend/app/core/redis_client.py import redis from app.core.config import settings redis_client = redis.Redis( host=settings.REDIS_HOST, port=settings.REDIS_PORT, db=settings.REDIS_DB, decode_responses=True ) # backend/app/api/api_v1/endpoints/tasks.py from fastapi import APIRouter, Depends, HTTPException from sqlalchemy.orm import Session from app.api import deps from app.core.redis_client import redis_client from app.crud import crud_task import json @router.get("/{task_id}", response_model=Task) def read_task( task_id: int, db: Session = Depends(deps.get_db), current_user: int = Depends(deps.get_current_active_user), ): # 尝试从缓存获取 cache_key = f"task:{task_id}:user:{current_user.id}" cached_task = redis_client.get(cache_key) if cached_task: return json.loads(cached_task) # 缓存未命中，从数据库获取 task = crud_task.get(db=db, id=task_id) if not task: raise HTTPException(status_code=404, detail="Task not found") if task.owner_id != current_user.id: raise HTTPException(status_code=400, detail="Not enough permissions") # 存入缓存，设置过期时间 redis_client.setex( cache_key, 3600, # 1小时过期 json.dumps(task.dict(), default=str) ) return task

5.2 前端性能优化

5.2.1 代码分割

使用React.lazy和Suspense实现代码分割，减少初始加载时间：

// frontend/src/App.js import React, { Suspense, lazy } from 'react'; import { BrowserRouter as Router, Routes, Route } from 'react-router-dom'; import { AuthProvider } from './contexts/AuthContext'; import Layout from './components/Layout'; import HomePage from './pages/HomePage'; import LoginPage from './pages/LoginPage'; import ProtectedRoute from './components/ProtectedRoute'; // 懒加载组件 const DashboardPage = lazy(() => import('./pages/DashboardPage')); const ProfilePage = lazy(() => import('./pages/ProfilePage')); function App() { return ( <AuthProvider> <Router> <Layout> <Suspense fallback={<div>Loading...</div>}> <Routes> <Route path="/" element={<HomePage />} /> <Route path="/login" element={<LoginPage />} /> <Route path="/dashboard" element={ <ProtectedRoute> <DashboardPage /> </ProtectedRoute> } /> <Route path="/profile" element={ <ProtectedRoute> <ProfilePage /> </ProtectedRoute> } /> </Routes> </Suspense> </Layout> </Router> </AuthProvider> ); } export default App;

5.2.2 虚拟化长列表

对于长列表，使用虚拟化技术只渲染可见区域的元素：

// frontend/src/components/VirtualizedTaskList.js import React, { useState, useEffect, useContext } from 'react'; import { FixedSizeList as List } from 'react-window'; import { AuthContext } from '../contexts/AuthContext'; import { getTasks } from '../services/taskService'; import TaskItem from './TaskItem'; import './VirtualizedTaskList.css'; const Row = ({ index, style, data }) => ( <div style={style}> <TaskItem task={data[index]} onDelete={data.onDelete} onTaskUpdated={data.onTaskUpdated} /> </div> ); const VirtualizedTaskList = () => { const [tasks, setTasks] = useState([]); const [loading, setLoading] = useState(true); const [error, setError] = useState(null); const { auth } = useContext(AuthContext); useEffect(() => { fetchTasks(); }, []); const fetchTasks = async () => { try { setLoading(true); const data = await getTasks(auth.token); setTasks(data); setError(null); } catch (err) { setError('Failed to fetch tasks. Please try again later.'); console.error(err); } finally { setLoading(false); } }; const handleDelete = (id) => { setTasks(tasks.filter(task => task.id !== id)); }; const handleTaskUpdated = (updatedTask) => { setTasks(tasks.map(task => task.id === updatedTask.id ? updatedTask : task)); }; const itemData = { tasks, onDelete: handleDelete, onTaskUpdated: handleTaskUpdated }; if (loading) return <div>Loading tasks...</div>; if (error) return <div className="error-message">{error}</div>; if (tasks.length === 0) return <div>No tasks found.</div>; return ( <div className="virtualized-task-list"> <List height={600} itemCount={tasks.length} itemSize={150} itemData={tasks} width="100%" > {Row} </List> </div> ); }; export default VirtualizedTaskList;

5.2.3 请求优化

使用React Query或SWR进行数据获取和缓存，优化API请求：

// frontend/src/services/api.js import axios from 'axios'; const API_URL = 'http://localhost:8000/api/v1'; const api = axios.create({ baseURL: API_URL, headers: { 'Content-Type': 'application/json', }, }); // 请求拦截器，添加认证token api.interceptors.request.use( (config) => { const token = localStorage.getItem('token'); if (token) { config.headers.Authorization = `Bearer ${token}`; } return config; }, (error) => { return Promise.reject(error); } ); // 响应拦截器，处理错误 api.interceptors.response.use( (response) => response, (error) => { if (error.response && error.response.status === 401) { // 处理未授权错误，例如重定向到登录页面 localStorage.removeItem('token'); window.location.href = '/login'; } return Promise.reject(error); } ); export default api; // frontend/src/hooks/useTasks.js import { useQuery, useMutation, useQueryClient } from 'react-query'; import api from '../services/api'; const fetchTasks = async () => { const response = await api.get('/tasks/'); return response.data; }; const createTask = async (task) => { const response = await api.post('/tasks/', task); return response.data; }; const updateTask = async ({ id, ...task }) => { const response = await api.put(`/tasks/${id}`, task); return response.data; }; const deleteTask = async (id) => { const response = await api.delete(`/tasks/${id}`); return response.data; }; export const useTasks = () => { return useQuery('tasks', fetchTasks); }; export const useCreateTask = () => { const queryClient = useQueryClient(); return useMutation(createTask, { onSuccess: () => { queryClient.invalidateQueries('tasks'); }, }); }; export const useUpdateTask = () => { const queryClient = useQueryClient(); return useMutation(updateTask, { onSuccess: () => { queryClient.invalidateQueries('tasks'); }, }); }; export const useDeleteTask = () => { const queryClient = useQueryClient(); return useMutation(deleteTask, { onSuccess: () => { queryClient.invalidateQueries('tasks'); }, }); };

6. 部署与运维

6.1 容器化部署

使用Docker和Docker Compose进行容器化部署：

# backend/Dockerfile FROM python:3.9 WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

# frontend/Dockerfile FROM node:16 as build WORKDIR /app COPY package*.json ./ RUN npm install COPY . . RUN npm run build FROM nginx:alpine COPY --from=build /app/build /usr/share/nginx/html COPY nginx.conf /etc/nginx/conf.d/default.conf EXPOSE 80 CMD ["nginx", "-g", "daemon off;"]

# docker-compose.yml version: '3.8' services: backend: build: ./backend ports: - "8000:8000" environment: - DATABASE_URL=postgresql://postgres:password@db:5432/taskdb - REDIS_HOST=redis depends_on: - db - redis volumes: - ./backend:/app command: uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload frontend: build: ./frontend ports: - "3000:80" depends_on: - backend db: image: postgres:13 environment: - POSTGRES_DB=taskdb - POSTGRES_USER=postgres - POSTGRES_PASSWORD=password volumes: - postgres_data:/var/lib/postgresql/data ports: - "5432:5432" redis: image: redis:6-alpine ports: - "6379:6379" volumes: postgres_data:

6.2 CI/CD流程

使用GitHub Actions实现CI/CD流程：

# .github/workflows/ci-cd.yml name: CI/CD Pipeline on: push: branches: [ main ] pull_request: branches: [ main ] jobs: test: runs-on: ubuntu-latest services: postgres: image: postgres:13 env: POSTGRES_PASSWORD: postgres options: >- --health-cmd pg_isready --health-interval 10s --health-timeout 5s --health-retries 5 ports: - 5432:5432 steps: - uses: actions/checkout@v2 - name: Set up Python uses: actions/setup-python@v2 with: python-version: 3.9 - name: Install dependencies run: | python -m pip install --upgrade pip pip install -r backend/requirements.txt - name: Run tests run: | cd backend python -m pytest env: DATABASE_URL: postgresql://postgres:postgres@localhost:5432/postgres - name: Set up Node.js uses: actions/setup-node@v2 with: node-version: '16' - name: Install frontend dependencies run: | cd frontend npm install - name: Run frontend tests run: | cd frontend npm test deploy: needs: test runs-on: ubuntu-latest if: github.ref == 'refs/heads/main' steps: - uses: actions/checkout@v2 - name: Set up Docker Buildx uses: docker/setup-buildx-action@v1 - name: Login to DockerHub uses: docker/login-action@v1 with: username: ${{ secrets.DOCKERHUB_USERNAME }} password: ${{ secrets.DOCKERHUB_TOKEN }} - name: Build and push backend uses: docker/build-push-action@v2 with: context: ./backend push: true tags: user/task-manager-backend:latest - name: Build and push frontend uses: docker/build-push-action@v2 with: context: ./frontend push: true tags: user/task-manager-frontend:latest - name: Deploy to production uses: appleboy/ssh-action@master with: host: ${{ secrets.PRODUCTION_HOST }} username: ${{ secrets.PRODUCTION_USER }} key: ${{ secrets.PRODUCTION_SSH_KEY }} script: | cd /opt/task-manager docker-compose pull docker-compose up -d

6.3 监控与日志

实现应用监控和日志收集：

# backend/app/core/logging.py import logging import sys from pathlib import Path def setup_logging(): # 创建日志目录 log_dir = Path("logs") log_dir.mkdir(exist_ok=True) # 配置日志格式 formatter = logging.Formatter( '%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) # 配置根日志记录器 root_logger = logging.getLogger() root_logger.setLevel(logging.INFO) # 控制台处理器 console_handler = logging.StreamHandler(sys.stdout) console_handler.setFormatter(formatter) root_logger.addHandler(console_handler) # 文件处理器 file_handler = logging.FileHandler("logs/app.log") file_handler.setFormatter(formatter) root_logger.addHandler(file_handler) # 错误文件处理器 error_file_handler = logging.FileHandler("logs/error.log") error_file_handler.setLevel(logging.ERROR) error_file_handler.setFormatter(formatter) root_logger.addHandler(error_file_handler) # backend/app/main.py from fastapi import FastAPI, Request from fastapi.responses import JSONResponse from fastapi.middleware.cors import CORSMiddleware import time import logging from app.core.logging import setup_logging # 设置日志 setup_logging() logger = logging.getLogger(__name__) app = FastAPI() # 请求日志中间件 @app.middleware("http") async def log_requests(request: Request, call_next): start_time = time.time() response = await call_next(request) process_time = time.time() - start_time logger.info( f"Method: {request.method}, Path: {request.url.path}, " f"Status: {response.status_code}, Process time: {process_time:.4f}s" ) return response # 全局异常处理 @app.exception_handler(Exception) async def global_exception_handler(request: Request, exc: Exception): logger.error(f"Unhandled exception: {exc}", exc_info=True) return JSONResponse( status_code=500, content={"detail": "Internal server error"} )

// frontend/src/services/loggingService.js class LoggingService { static logError(error, context = {}) { const errorData = { message: error.message, stack: error.stack, context, timestamp: new Date().toISOString(), userAgent: navigator.userAgent, url: window.location.href }; // 在开发环境中打印到控制台 if (process.env.NODE_ENV === 'development') { console.error('Error logged:', errorData); } // 发送到错误跟踪服务 this.sendToErrorTracking(errorData); } static logInfo(message, context = {}) { const infoData = { message, context, timestamp: new Date().toISOString(), userAgent: navigator.userAgent, url: window.location.href }; // 发送到日志服务 this.sendToLoggingService(infoData); } static sendToErrorTracking(errorData) { // 实现发送到错误跟踪服务（如Sentry、LogRocket等） if (process.env.NODE_ENV === 'production') { // 例如：Sentry.captureException(errorData); } } static sendToLoggingService(infoData) { // 实现发送到日志服务 if (process.env.NODE_ENV === 'production') { fetch('/api/v1/logs', { method: 'POST', headers: { 'Content-Type': 'application/json', }, body: JSON.stringify(infoData), }).catch(err => { console.error('Failed to send log:', err); }); } } } export default LoggingService; // 在React组件中使用 import React, { useEffect } from 'react'; import LoggingService from '../services/loggingService'; function MyComponent() { useEffect(() => { try { // 组件逻辑 } catch (error) { LoggingService.logError(error, { component: 'MyComponent' }); } LoggingService.logInfo('Component mounted', { component: 'MyComponent' }); }, []); return <div>My Component</div>; }