AutoGPT-09-最佳实践与实战经验

# 遵循PEP 8规范
# 使用类型注解提高代码可读性和IDE支持

from typing import Dict, List, Optional, Union
from datetime import datetime
from pydantic import BaseModel, Field

class UserProfile(BaseModel):
    """
    用户配置模型
    
    遵循以下规范：
    1. 类名使用PascalCase
    2. 函数和变量使用snake_case
    3. 常量使用UPPER_CASE
    4. 私有成员使用下划线前缀
    """
    
    id: str = Field(..., description="用户唯一标识符")
    email: str = Field(..., description="用户邮箱地址")
    created_at: datetime = Field(..., description="创建时间")
    preferences: Dict[str, Union[str, int, bool]] = Field(
        default_factory=dict, 
        description="用户偏好设置"
    )
    
    def get_display_name(self) -> str:
        """
        获取用户显示名称
        
        Returns:
            str: 用户显示名称
        """
        return self.email.split('@')[0]
    
    def _validate_preferences(self) -> bool:
        """私有方法：验证偏好设置"""
        # 实现验证逻辑
        return True

# 常量定义
DEFAULT_PAGE_SIZE = 20
MAX_RETRY_ATTEMPTS = 3
CACHE_EXPIRY_SECONDS = 3600

import logging
from typing import Optional
from contextlib import asynccontextmanager

logger = logging.getLogger(__name__)

class AutoGPTError(Exception):
    """AutoGPT基础异常类"""
    
    def __init__(self, message: str, error_code: Optional[str] = None):
        self.message = message
        self.error_code = error_code
        super().__init__(self.message)

class ValidationError(AutoGPTError):
    """数据验证异常"""
    pass

class ExecutionError(AutoGPTError):
    """执行异常"""
    pass

async def execute_graph_safely(graph_id: str, inputs: Dict) -> Dict:
    """
    安全执行图的示例
    
    最佳实践：
    1. 明确的异常类型
    2. 详细的错误日志
    3. 优雅的错误处理
    4. 资源清理
    """
    try:
        logger.info(f"开始执行图: {graph_id}")
        
        # 验证输入
        if not graph_id:
            raise ValidationError("图ID不能为空", "INVALID_GRAPH_ID")
        
        # 执行逻辑
        result = await _execute_graph_internal(graph_id, inputs)
        
        logger.info(f"图执行成功: {graph_id}")
        return result
        
    except ValidationError as e:
        logger.error(f"输入验证失败: {e.message}")
        raise
    except ExecutionError as e:
        logger.error(f"图执行失败: {e.message}")
        raise
    except Exception as e:
        logger.error(f"未知错误: {str(e)}", exc_info=True)
        raise ExecutionError(f"图执行异常: {str(e)}", "EXECUTION_FAILED")

@asynccontextmanager
async def database_transaction():
    """数据库事务上下文管理器"""
    transaction = None
    try:
        transaction = await db.begin()
        yield transaction
        await transaction.commit()
    except Exception as e:
        if transaction:
            await transaction.rollback()
        logger.error(f"数据库事务失败: {e}")
        raise

// 使用TypeScript提供类型安全
import React, { useState, useEffect, useCallback } from 'react';
import { useQuery, useMutation } from '@tanstack/react-query';

// 接口定义
interface GraphExecutionProps {
  graphId: string;
  onExecutionComplete?: (result: ExecutionResult) => void;
  className?: string;
}

interface ExecutionResult {
  id: string;
  status: 'completed' | 'failed' | 'cancelled';
  outputs: Record<string, unknown>;
  error?: string;
}

// 组件实现
export const GraphExecution: React.FC<GraphExecutionProps> = ({
  graphId,
  onExecutionComplete,
  className = ''
}) => {
  // 状态管理
  const [isExecuting, setIsExecuting] = useState(false);
  const [executionId, setExecutionId] = useState<string | null>(null);
  
  // 数据获取
  const { data: graph, isLoading } = useQuery({
    queryKey: ['graph', graphId],
    queryFn: () => fetchGraph(graphId),
    enabled: !!graphId
  });
  
  // 执行变更
  const executeMutation = useMutation({
    mutationFn: executeGraph,
    onSuccess: (result) => {
      setIsExecuting(false);
      onExecutionComplete?.(result);
    },
    onError: (error) => {
      setIsExecuting(false);
      console.error('执行失败:', error);
    }
  });
  
  // 事件处理
  const handleExecute = useCallback(async () => {
    if (!graph || isExecuting) return;
    
    setIsExecuting(true);
    try {
      const result = await executeMutation.mutateAsync({
        graphId,
        inputs: {}
      });
      setExecutionId(result.id);
    } catch (error) {
      // 错误已在mutation中处理
    }
  }, [graph, isExecuting, executeMutation, graphId]);
  
  // 副作用
  useEffect(() => {
    // 清理逻辑
    return () => {
      if (executionId) {
        // 取消执行
        cancelExecution(executionId);
      }
    };
  }, [executionId]);
  
  if (isLoading) {
    return <div className="loading">加载中...</div>;
  }
  
  return (
    <div className={`graph-execution ${className}`}>
      <h3>{graph?.name}</h3>
      <button 
        onClick={handleExecute}
        disabled={isExecuting}
        className="execute-button"
      >
        {isExecuting ? '执行中...' : '执行'}
      </button>
    </div>
  );
};

// 工具函数
async function fetchGraph(graphId: string): Promise<Graph> {
  const response = await fetch(`/api/graphs/${graphId}`);
  if (!response.ok) {
    throw new Error(`获取图失败: ${response.statusText}`);
  }
  return response.json();
}

async function executeGraph(params: {
  graphId: string;
  inputs: Record<string, unknown>;
}): Promise<ExecutionResult> {
  const response = await fetch('/api/executions', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify(params)
  });
  
  if (!response.ok) {
    throw new Error(`执行失败: ${response.statusText}`);
  }
  
  return response.json();
}

# 好的例子：职责单一的服务类
class UserAuthenticationService:
    """用户认证服务 - 只负责认证相关逻辑"""
    
    def __init__(self, jwt_service: JWTService, user_repository: UserRepository):
        self.jwt_service = jwt_service
        self.user_repository = user_repository
    
    async def authenticate_user(self, email: str, password: str) -> Optional[User]:
        """认证用户"""
        user = await self.user_repository.get_by_email(email)
        if user and self._verify_password(password, user.password_hash):
            return user
        return None
    
    def _verify_password(self, password: str, password_hash: str) -> bool:
        """验证密码"""
        # 密码验证逻辑
        pass

class UserProfileService:
    """用户配置服务 - 只负责配置管理"""
    
    def __init__(self, user_repository: UserRepository):
        self.user_repository = user_repository
    
    async def update_profile(self, user_id: str, profile_data: Dict) -> User:
        """更新用户配置"""
        # 配置更新逻辑
        pass

# 避免的例子：职责混乱的服务类
class UserService:  # 不好的设计
    """用户服务 - 职责过多"""
    
    async def authenticate_user(self, email: str, password: str):
        pass
    
    async def update_profile(self, user_id: str, profile_data: Dict):
        pass
    
    async def send_notification(self, user_id: str, message: str):
        pass
    
    async def calculate_usage_cost(self, user_id: str):
        pass

from abc import ABC, abstractmethod
from typing import Protocol

# 定义接口
class EmailServiceProtocol(Protocol):
    async def send_email(self, to: str, subject: str, body: str) -> bool:
        ...

class UserRepositoryProtocol(Protocol):
    async def get_by_id(self, user_id: str) -> Optional[User]:
        ...
    
    async def save(self, user: User) -> User:
        ...

# 实现类
class SMTPEmailService:
    """SMTP邮件服务实现"""
    
    def __init__(self, smtp_config: SMTPConfig):
        self.smtp_config = smtp_config
    
    async def send_email(self, to: str, subject: str, body: str) -> bool:
        # SMTP发送实现
        pass

class DatabaseUserRepository:
    """数据库用户仓储实现"""
    
    def __init__(self, db_connection: DatabaseConnection):
        self.db = db_connection
    
    async def get_by_id(self, user_id: str) -> Optional[User]:
        # 数据库查询实现
        pass

# 服务类使用依赖注入
class NotificationService:
    """通知服务"""
    
    def __init__(
        self, 
        email_service: EmailServiceProtocol,
        user_repository: UserRepositoryProtocol
    ):
        self.email_service = email_service
        self.user_repository = user_repository
    
    async def notify_user(self, user_id: str, message: str) -> bool:
        """通知用户"""
        user = await self.user_repository.get_by_id(user_id)
        if not user:
            return False
        
        return await self.email_service.send_email(
            to=user.email,
            subject="AutoGPT通知",
            body=message
        )

# 依赖注入容器配置
def create_notification_service() -> NotificationService:
    """创建通知服务实例"""
    smtp_config = SMTPConfig.from_env()
    email_service = SMTPEmailService(smtp_config)
    
    db_connection = create_db_connection()
    user_repository = DatabaseUserRepository(db_connection)
    
    return NotificationService(email_service, user_repository)

import asyncio
from typing import List, Dict, Any
from contextlib import asynccontextmanager

class GraphExecutionService:
    """图执行服务 - 异步编程最佳实践"""
    
    def __init__(self, max_concurrent_executions: int = 10):
        self.semaphore = asyncio.Semaphore(max_concurrent_executions)
        self.active_executions: Dict[str, asyncio.Task] = {}
    
    async def execute_graph(self, graph_id: str, inputs: Dict) -> str:
        """执行图 - 返回执行ID"""
        execution_id = generate_execution_id()
        
        # 使用信号量控制并发数
        async with self.semaphore:
            task = asyncio.create_task(
                self._execute_graph_internal(execution_id, graph_id, inputs)
            )
            self.active_executions[execution_id] = task
            
            # 设置任务完成回调
            task.add_done_callback(
                lambda t: self.active_executions.pop(execution_id, None)
            )
        
        return execution_id
    
    async def _execute_graph_internal(
        self, 
        execution_id: str, 
        graph_id: str, 
        inputs: Dict
    ) -> Dict:
        """内部执行逻辑"""
        try:
            # 获取图定义
            graph = await self._get_graph(graph_id)
            
            # 并发执行节点
            results = await self._execute_nodes_concurrently(
                execution_id, 
                graph.nodes, 
                inputs
            )
            
            return {"execution_id": execution_id, "results": results}
            
        except Exception as e:
            logger.error(f"图执行失败 {execution_id}: {e}")
            raise
    
    async def _execute_nodes_concurrently(
        self, 
        execution_id: str, 
        nodes: List[Node], 
        inputs: Dict
    ) -> Dict:
        """并发执行节点"""
        # 构建依赖图
        dependency_graph = self._build_dependency_graph(nodes)
        
        # 按拓扑顺序执行
        results = {}
        executed_nodes = set()
        
        while len(executed_nodes) < len(nodes):
            # 找到可以执行的节点（依赖已满足）
            ready_nodes = [
                node for node in nodes
                if node.id not in executed_nodes
                and all(dep in executed_nodes for dep in dependency_graph.get(node.id, []))
            ]
            
            if not ready_nodes:
                raise ExecutionError("检测到循环依赖")
            
            # 并发执行就绪的节点
            tasks = [
                self._execute_node(execution_id, node, inputs, results)
                for node in ready_nodes
            ]
            
            node_results = await asyncio.gather(*tasks, return_exceptions=True)
            
            # 处理结果
            for node, result in zip(ready_nodes, node_results):
                if isinstance(result, Exception):
                    raise ExecutionError(f"节点执行失败 {node.id}: {result}")
                
                results[node.id] = result
                executed_nodes.add(node.id)
        
        return results
    
    async def cancel_execution(self, execution_id: str) -> bool:
        """取消执行"""
        task = self.active_executions.get(execution_id)
        if task and not task.done():
            task.cancel()
            try:
                await task
            except asyncio.CancelledError:
                pass
            return True
        return False
    
    @asynccontextmanager
    async def execution_context(self, execution_id: str):
        """执行上下文管理器"""
        logger.info(f"开始执行 {execution_id}")
        start_time = asyncio.get_event_loop().time()
        
        try:
            yield
        finally:
            end_time = asyncio.get_event_loop().time()
            duration = end_time - start_time
            logger.info(f"执行完成 {execution_id}, 耗时: {duration:.2f}s")

import pytest
from unittest.mock import Mock, AsyncMock, patch
from typing import Dict, Any

class TestUserAuthenticationService:
    """用户认证服务测试类
    
    测试命名规范：test_<method_name>_<scenario>_<expected_result>
    """
    
    @pytest.fixture
    def mock_jwt_service(self):
        """JWT服务模拟对象"""
        return Mock(spec=JWTService)
    
    @pytest.fixture
    def mock_user_repository(self):
        """用户仓储模拟对象"""
        return AsyncMock(spec=UserRepository)
    
    @pytest.fixture
    def auth_service(self, mock_jwt_service, mock_user_repository):
        """认证服务实例"""
        return UserAuthenticationService(mock_jwt_service, mock_user_repository)
    
    async def test_authenticate_user_valid_credentials_returns_user(
        self, 
        auth_service, 
        mock_user_repository
    ):
        """测试：有效凭据认证返回用户"""
        # Arrange
        email = "test@example.com"
        password = "password123"
        expected_user = User(id="user1", email=email)
        
        mock_user_repository.get_by_email.return_value = expected_user
        
        # Act
        result = await auth_service.authenticate_user(email, password)
        
        # Assert
        assert result == expected_user
        mock_user_repository.get_by_email.assert_called_once_with(email)
    
    async def test_authenticate_user_invalid_credentials_returns_none(
        self, 
        auth_service, 
        mock_user_repository
    ):
        """测试：无效凭据认证返回None"""
        # Arrange
        email = "test@example.com"
        password = "wrong_password"
        
        mock_user_repository.get_by_email.return_value = None
        
        # Act
        result = await auth_service.authenticate_user(email, password)
        
        # Assert
        assert result is None
    
    async def test_authenticate_user_database_error_raises_exception(
        self, 
        auth_service, 
        mock_user_repository
    ):
        """测试：数据库错误抛出异常"""
        # Arrange
        email = "test@example.com"
        password = "password123"
        
        mock_user_repository.get_by_email.side_effect = DatabaseError("连接失败")
        
        # Act & Assert
        with pytest.raises(DatabaseError):
            await auth_service.authenticate_user(email, password)

# 参数化测试
@pytest.mark.parametrize("email,password,expected", [
    ("valid@example.com", "password123", True),
    ("", "password123", False),
    ("valid@example.com", "", False),
    ("invalid-email", "password123", False),
])
async def test_validate_credentials_various_inputs(email, password, expected):
    """测试：各种输入的凭据验证"""
    result = validate_credentials(email, password)
    assert result == expected

# 集成测试
@pytest.mark.integration
class TestGraphExecutionIntegration:
    """图执行集成测试"""
    
    @pytest.fixture
    async def test_database(self):
        """测试数据库"""
        # 创建测试数据库连接
        db = await create_test_database()
        yield db
        await cleanup_test_database(db)
    
    async def test_complete_graph_execution_flow(self, test_database):
        """测试：完整的图执行流程"""
        # 创建测试图
        graph = await create_test_graph(test_database)
        
        # 执行图
        execution_service = GraphExecutionService(test_database)
        execution_id = await execution_service.execute_graph(
            graph.id, 
            {"input": "test_value"}
        )
        
        # 等待执行完成
        result = await wait_for_execution_completion(execution_id)
        
        # 验证结果
        assert result.status == "completed"
        assert "output" in result.outputs

// tests/e2e/graph-execution.spec.ts
import { test, expect } from '@playwright/test';

test.describe('图执行功能', () => {
  test.beforeEach(async ({ page }) => {
    // 登录测试用户
    await page.goto('/login');
    await page.fill('[data-testid="email-input"]', 'test@example.com');
    await page.fill('[data-testid="password-input"]', 'password123');
    await page.click('[data-testid="login-button"]');
    
    // 等待跳转到主页
    await expect(page).toHaveURL('/dashboard');
  });

  test('应该能够创建和执行简单图', async ({ page }) => {
    // 导航到图编辑器
    await page.click('[data-testid="create-graph-button"]');
    await expect(page).toHaveURL('/graphs/new');
    
    // 添加节点
    await page.click('[data-testid="add-node-button"]');
    await page.selectOption('[data-testid="node-type-select"]', 'text-input');
    await page.click('[data-testid="confirm-add-node"]');
    
    // 配置节点
    await page.fill('[data-testid="node-input-text"]', 'Hello, World!');
    
    // 保存图
    await page.fill('[data-testid="graph-name-input"]', '测试图');
    await page.click('[data-testid="save-graph-button"]');
    
    // 等待保存成功
    await expect(page.locator('[data-testid="success-message"]')).toBeVisible();
    
    // 执行图
    await page.click('[data-testid="execute-graph-button"]');
    
    // 等待执行完成
    await expect(page.locator('[data-testid="execution-status"]')).toHaveText('已完成');
    
    // 验证输出
    const output = await page.textContent('[data-testid="execution-output"]');
    expect(output).toContain('Hello, World!');
  });

  test('应该能够处理执行错误', async ({ page }) => {
    // 创建会失败的图
    await page.goto('/graphs/new');
    await page.click('[data-testid="add-node-button"]');
    await page.selectOption('[data-testid="node-type-select"]', 'api-call');
    
    // 配置无效的API调用
    await page.fill('[data-testid="api-url-input"]', 'invalid-url');
    
    // 保存并执行
    await page.fill('[data-testid="graph-name-input"]', '错误测试图');
    await page.click('[data-testid="save-graph-button"]');
    await page.click('[data-testid="execute-graph-button"]');
    
    // 验证错误处理
    await expect(page.locator('[data-testid="execution-status"]')).toHaveText('失败');
    await expect(page.locator('[data-testid="error-message"]')).toBeVisible();
  });
});

-- 用户表优化
CREATE INDEX CONCURRENTLY idx_users_email ON users(email);
CREATE INDEX CONCURRENTLY idx_users_created_at ON users(created_at);
CREATE INDEX CONCURRENTLY idx_users_active ON users(is_active) WHERE is_active = true;

-- 图表优化
CREATE INDEX CONCURRENTLY idx_graphs_user_id ON graphs(user_id);
CREATE INDEX CONCURRENTLY idx_graphs_user_active ON graphs(user_id, is_active) 
  WHERE is_active = true;

-- 执行表优化
CREATE INDEX CONCURRENTLY idx_executions_user_status ON graph_executions(user_id, status);
CREATE INDEX CONCURRENTLY idx_executions_created_at ON graph_executions(created_at);
CREATE INDEX CONCURRENTLY idx_executions_graph_id ON graph_executions(graph_id);

-- 复合索引优化
CREATE INDEX CONCURRENTLY idx_executions_user_created ON graph_executions(user_id, created_at DESC);

class OptimizedGraphRepository:
    """优化的图仓储实现"""
    
    async def get_user_graphs_with_stats(
        self, 
        user_id: str, 
        limit: int = 20,
        offset: int = 0
    ) -> List[Dict]:
        """
        获取用户图列表及统计信息
        
        优化策略：
        1. 使用单个查询获取所有需要的数据
        2. 避免N+1查询问题
        3. 使用适当的索引
        """
        query = """
        SELECT 
            g.id,
            g.name,
            g.description,
            g.created_at,
            g.updated_at,
            COUNT(ge.id) as execution_count,
            MAX(ge.created_at) as last_execution_at,
            AVG(CASE WHEN ge.status = 'completed' THEN 1.0 ELSE 0.0 END) as success_rate
        FROM graphs g
        LEFT JOIN graph_executions ge ON g.id = ge.graph_id
        WHERE g.user_id = $1 AND g.is_active = true
        GROUP BY g.id, g.name, g.description, g.created_at, g.updated_at
        ORDER BY g.updated_at DESC
        LIMIT $2 OFFSET $3
        """
        
        return await self.db.fetch_all(query, user_id, limit, offset)
    
    async def get_execution_history_paginated(
        self,
        user_id: str,
        page: int = 1,
        page_size: int = 20,
        status_filter: Optional[str] = None
    ) -> Dict:
        """
        分页获取执行历史
        
        优化策略：
        1. 使用游标分页提高大数据集性能
        2. 并行执行数据查询和计数查询
        3. 使用适当的WHERE条件和索引
        """
        offset = (page - 1) * page_size
        
        # 构建WHERE条件
        where_conditions = ["user_id = $1"]
        params = [user_id]
        
        if status_filter:
            where_conditions.append("status = $2")
            params.append(status_filter)
        
        where_clause = " AND ".join(where_conditions)
        
        # 并行执行数据查询和计数查询
        data_query = f"""
        SELECT id, graph_id, status, created_at, started_at, ended_at
        FROM graph_executions
        WHERE {where_clause}
        ORDER BY created_at DESC
        LIMIT ${len(params) + 1} OFFSET ${len(params) + 2}
        """
        
        count_query = f"""
        SELECT COUNT(*) as total
        FROM graph_executions
        WHERE {where_clause}
        """
        
        # 使用asyncio.gather并行执行
        data_task = self.db.fetch_all(data_query, *params, page_size, offset)
        count_task = self.db.fetch_one(count_query, *params)
        
        data_results, count_result = await asyncio.gather(data_task, count_task)
        
        return {
            "data": data_results,
            "pagination": {
                "page": page,
                "page_size": page_size,
                "total": count_result["total"],
                "total_pages": (count_result["total"] + page_size - 1) // page_size
            }
        }

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import QueuePool

class DatabaseManager:
    """数据库管理器 - 连接池优化"""
    
    def __init__(self, database_url: str):
        # 连接池配置
        self.engine = create_async_engine(
            database_url,
            # 连接池设置
            poolclass=QueuePool,
            pool_size=20,          # 连接池大小
            max_overflow=30,       # 最大溢出连接数
            pool_timeout=30,       # 获取连接超时时间
            pool_recycle=3600,     # 连接回收时间（秒）
            pool_pre_ping=True,    # 连接前ping测试
            
            # 连接参数
            connect_args={
                "server_settings": {
                    "application_name": "autogpt_backend",
                    "jit": "off"  # 关闭JIT以提高小查询性能
                }
            },
            
            # 执行选项
            execution_options={
                "isolation_level": "READ_COMMITTED"
            }
        )
        
        self.async_session = sessionmaker(
            self.engine,
            class_=AsyncSession,
            expire_on_commit=False
        )
    
    async def get_session(self) -> AsyncSession:
        """获取数据库会话"""
        return self.async_session()
    
    async def close(self):
        """关闭数据库连接"""
        await self.engine.dispose()

from typing import Optional, Any, Dict
import json
import hashlib
from datetime import timedelta

class MultiLevelCache:
    """多层缓存实现"""
    
    def __init__(
        self,
        redis_client,
        local_cache_size: int = 1000,
        default_ttl: int = 3600
    ):
        self.redis = redis_client
        self.local_cache = {}  # 简化的本地缓存
        self.local_cache_size = local_cache_size
        self.default_ttl = default_ttl
    
    async def get(self, key: str) -> Optional[Any]:
        """
        获取缓存值
        
        查找顺序：本地缓存 -> Redis -> 数据库
        """
        # 1. 检查本地缓存
        if key in self.local_cache:
            return self.local_cache[key]
        
        # 2. 检查Redis缓存
        redis_value = await self.redis.get(key)
        if redis_value:
            try:
                value = json.loads(redis_value)
                # 更新本地缓存
                self._update_local_cache(key, value)
                return value
            except json.JSONDecodeError:
                pass
        
        return None
    
    async def set(
        self, 
        key: str, 
        value: Any, 
        ttl: Optional[int] = None
    ) -> bool:
        """
        设置缓存值
        
        同时更新本地缓存和Redis
        """
        ttl = ttl or self.default_ttl
        
        try:
            # 序列化值
            serialized_value = json.dumps(value, default=str)
            
            # 更新Redis
            await self.redis.setex(key, ttl, serialized_value)
            
            # 更新本地缓存
            self._update_local_cache(key, value)
            
            return True
        except Exception as e:
            logger.error(f"缓存设置失败: {e}")
            return False
    
    async def delete(self, key: str) -> bool:
        """删除缓存"""
        # 删除本地缓存
        self.local_cache.pop(key, None)
        
        # 删除Redis缓存
        return await self.redis.delete(key) > 0
    
    def _update_local_cache(self, key: str, value: Any):
        """更新本地缓存"""
        # 简单的LRU实现
        if len(self.local_cache) >= self.local_cache_size:
            # 删除最旧的条目
            oldest_key = next(iter(self.local_cache))
            del self.local_cache[oldest_key]
        
        self.local_cache[key] = value

class CacheService:
    """缓存服务 - 业务层缓存封装"""
    
    def __init__(self, cache: MultiLevelCache):
        self.cache = cache
    
    async def get_user_profile(self, user_id: str) -> Optional[Dict]:
        """获取用户配置（带缓存）"""
        cache_key = f"user_profile:{user_id}"
        
        # 尝试从缓存获取
        cached_profile = await self.cache.get(cache_key)
        if cached_profile:
            return cached_profile
        
        # 从数据库获取
        profile = await self._fetch_user_profile_from_db(user_id)
        if profile:
            # 缓存用户配置（1小时）
            await self.cache.set(cache_key, profile, ttl=3600)
        
        return profile
    
    async def invalidate_user_cache(self, user_id: str):
        """使用户缓存失效"""
        patterns = [
            f"user_profile:{user_id}",
            f"user_graphs:{user_id}",
            f"user_executions:{user_id}",
        ]
        
        for pattern in patterns:
            await self.cache.delete(pattern)
    
    def cache_key_for_graph_list(
        self, 
        user_id: str, 
        page: int, 
        filters: Dict
    ) -> str:
        """生成图列表的缓存键"""
        # 创建过滤器的哈希
        filter_hash = hashlib.md5(
            json.dumps(filters, sort_keys=True).encode()
        ).hexdigest()[:8]
        
        return f"graph_list:{user_id}:{page}:{filter_hash}"

class CacheWarmupService:
    """缓存预热服务"""
    
    def __init__(self, cache_service: CacheService):
        self.cache_service = cache_service
    
    async def warmup_user_data(self, user_id: str):
        """预热用户数据"""
        tasks = [
            self._warmup_user_profile(user_id),
            self._warmup_user_graphs(user_id),
            self._warmup_recent_executions(user_id),
        ]
        
        await asyncio.gather(*tasks, return_exceptions=True)
    
    async def _warmup_user_profile(self, user_id: str):
        """预热用户配置"""
        await self.cache_service.get_user_profile(user_id)
    
    async def _warmup_user_graphs(self, user_id: str):
        """预热用户图列表"""
        # 预热前几页的图列表
        for page in range(1, 4):
            await self.cache_service.get_user_graphs(
                user_id, 
                page=page, 
                filters={}
            )
    
    async def schedule_cache_refresh(self):
        """定期刷新缓存"""
        while True:
            try:
                # 获取活跃用户列表
                active_users = await self._get_active_users()
                
                # 并发预热活跃用户数据
                semaphore = asyncio.Semaphore(10)  # 限制并发数
                
                async def warmup_with_semaphore(user_id: str):
                    async with semaphore:
                        await self.warmup_user_data(user_id)
                
                tasks = [
                    warmup_with_semaphore(user_id) 
                    for user_id in active_users
                ]
                
                await asyncio.gather(*tasks, return_exceptions=True)
                
                # 等待下次刷新
                await asyncio.sleep(300)  # 5分钟
                
            except Exception as e:
                logger.error(f"缓存刷新失败: {e}")
                await asyncio.sleep(60)  # 错误时等待1分钟

import React, { memo, useMemo, useCallback, useState } from 'react';
import { useVirtualizer } from '@tanstack/react-virtual';

// 使用memo优化组件重渲染
const GraphListItem = memo<{
  graph: Graph;
  onExecute: (graphId: string) => void;
  onEdit: (graphId: string) => void;
}>(({ graph, onExecute, onEdit }) => {
  // 使用useCallback缓存事件处理函数
  const handleExecute = useCallback(() => {
    onExecute(graph.id);
  }, [graph.id, onExecute]);
  
  const handleEdit = useCallback(() => {
    onEdit(graph.id);
  }, [graph.id, onEdit]);
  
  return (
    <div className="graph-item">
      <h3>{graph.name}</h3>
      <p>{graph.description}</p>
      <div className="actions">
        <button onClick={handleExecute}>执行</button>
        <button onClick={handleEdit}>编辑</button>
      </div>
    </div>
  );
});

// 虚拟滚动优化大列表
const VirtualizedGraphList: React.FC<{
  graphs: Graph[];
  onExecute: (graphId: string) => void;
  onEdit: (graphId: string) => void;
}> = ({ graphs, onExecute, onEdit }) => {
  const parentRef = React.useRef<HTMLDivElement>(null);
  
  const virtualizer = useVirtualizer({
    count: graphs.length,
    getScrollElement: () => parentRef.current,
    estimateSize: () => 120, // 估算每项高度
    overscan: 5, // 预渲染项数
  });
  
  return (
    <div
      ref={parentRef}
      className="graph-list-container"
      style={{ height: '600px', overflow: 'auto' }}
    >
      <div
        style={{
          height: `${virtualizer.getTotalSize()}px`,
          width: '100%',
          position: 'relative',
        }}
      >
        {virtualizer.getVirtualItems().map((virtualItem) => {
          const graph = graphs[virtualItem.index];
          return (
            <div
              key={virtualItem.key}
              style={{
                position: 'absolute',
                top: 0,
                left: 0,
                width: '100%',
                height: `${virtualItem.size}px`,
                transform: `translateY(${virtualItem.start}px)`,
              }}
            >
              <GraphListItem
                graph={graph}
                onExecute={onExecute}
                onEdit={onEdit}
              />
            </div>
          );
        })}
      </div>
    </div>
  );
};

// 使用useMemo优化计算
const GraphDashboard: React.FC = () => {
  const [graphs, setGraphs] = useState<Graph[]>([]);
  const [filter, setFilter] = useState('');
  
  // 使用useMemo缓存过滤结果
  const filteredGraphs = useMemo(() => {
    if (!filter) return graphs;
    
    return graphs.filter(graph =>
      graph.name.toLowerCase().includes(filter.toLowerCase()) ||
      graph.description.toLowerCase().includes(filter.toLowerCase())
    );
  }, [graphs, filter]);
  
  // 使用useMemo缓存统计数据
  const stats = useMemo(() => {
    return {
      total: graphs.length,
      active: graphs.filter(g => g.isActive).length,
      templates: graphs.filter(g => g.isTemplate).length,
    };
  }, [graphs]);
  
  const handleExecute = useCallback((graphId: string) => {
    // 执行逻辑
  }, []);
  
  const handleEdit = useCallback((graphId: string) => {
    // 编辑逻辑
  }, []);
  
  return (
    <div className="dashboard">
      <div className="stats">
        <div>总计: {stats.total}</div>
        <div>活跃: {stats.active}</div>
        <div>模板: {stats.templates}</div>
      </div>
      
      <input
        type="text"
        placeholder="搜索图..."
        value={filter}
        onChange={(e) => setFilter(e.target.value)}
      />
      
      <VirtualizedGraphList
        graphs={filteredGraphs}
        onExecute={handleExecute}
        onEdit={handleEdit}
      />
    </div>
  );
};

import { lazy, Suspense } from 'react';
import { Routes, Route } from 'react-router-dom';

// 懒加载组件
const GraphEditor = lazy(() => import('./components/GraphEditor'));
const GraphList = lazy(() => import('./components/GraphList'));
const Settings = lazy(() => import('./components/Settings'));
const Analytics = lazy(() => import('./components/Analytics'));

// 加载组件
const LoadingSpinner = () => (
  <div className="loading-spinner">
    <div className="spinner" />
    <p>加载中...</p>
  </div>
);

// 错误边界组件
class ErrorBoundary extends React.Component<
  { children: React.ReactNode },
  { hasError: boolean }
> {
  constructor(props: { children: React.ReactNode }) {
    super(props);
    this.state = { hasError: false };
  }
  
  static getDerivedStateFromError(error: Error) {
    return { hasError: true };
  }
  
  componentDidCatch(error: Error, errorInfo: React.ErrorInfo) {
    console.error('组件错误:', error, errorInfo);
  }
  
  render() {
    if (this.state.hasError) {
      return (
        <div className="error-boundary">
          <h2>出错了</h2>
          <p>页面加载失败，请刷新重试。</p>
        </div>
      );
    }
    
    return this.props.children;
  }
}

// 应用路由
const App: React.FC = () => {
  return (
    <ErrorBoundary>
      <Routes>
        <Route path="/" element={
          <Suspense fallback={<LoadingSpinner />}>
            <GraphList />
          </Suspense>
        } />
        
        <Route path="/editor/:graphId?" element={
          <Suspense fallback={<LoadingSpinner />}>
            <GraphEditor />
          </Suspense>
        } />
        
        <Route path="/settings" element={
          <Suspense fallback={<LoadingSpinner />}>
            <Settings />
          </Suspense>
        } />
        
        <Route path="/analytics" element={
          <Suspense fallback={<LoadingSpinner />}>
            <Analytics />
          </Suspense>
        } />
      </Routes>
    </ErrorBoundary>
  );
};

// 预加载关键路由
const preloadRoutes = () => {
  // 在空闲时间预加载常用组件
  if ('requestIdleCallback' in window) {
    requestIdleCallback(() => {
      import('./components/GraphEditor');
      import('./components/Settings');
    });
  }
};

// 在应用启动时预加载
export { App, preloadRoutes };

import jwt
import secrets
from datetime import datetime, timedelta
from typing import Dict, Optional

class SecureJWTService:
    """安全的JWT服务实现"""
    
    def __init__(self, secret_key: str, algorithm: str = "HS256"):
        if len(secret_key) < 32:
            raise ValueError("JWT密钥长度至少32字符")
        
        self.secret_key = secret_key
        self.algorithm = algorithm
        self.access_token_expire = timedelta(minutes=15)  # 短期访问令牌
        self.refresh_token_expire = timedelta(days=7)     # 长期刷新令牌
    
    def create_access_token(self, user_id: str, permissions: List[str]) -> str:
        """创建访问令牌"""
        now = datetime.utcnow()
        payload = {
            "sub": user_id,
            "iat": now,
            "exp": now + self.access_token_expire,
            "type": "access",
            "permissions": permissions,
            "jti": secrets.token_urlsafe(16),  # JWT ID防止重放攻击
        }
        
        return jwt.encode(payload, self.secret_key, algorithm=self.algorithm)
    
    def create_refresh_token(self, user_id: str) -> str:
        """创建刷新令牌"""
        now = datetime.utcnow()
        payload = {
            "sub": user_id,
            "iat": now,
            "exp": now + self.refresh_token_expire,
            "type": "refresh",
            "jti": secrets.token_urlsafe(16),
        }
        
        return jwt.encode(payload, self.secret_key, algorithm=self.algorithm)
    
    def verify_token(self, token: str, token_type: str = "access") -> Optional[Dict]:
        """验证令牌"""
        try:
            payload = jwt.decode(
                token,
                self.secret_key,
                algorithms=[self.algorithm],
                options={
                    "verify_exp": True,
                    "verify_iat": True,
                    "verify_signature": True,
                }
            )
            
            # 验证令牌类型
            if payload.get("type") != token_type:
                return None
            
            # 检查令牌是否在黑名单中
            if await self._is_token_blacklisted(payload.get("jti")):
                return None
            
            return payload
            
        except jwt.ExpiredSignatureError:
            logger.warning("JWT令牌已过期")
            return None
        except jwt.InvalidTokenError as e:
            logger.warning(f"JWT令牌无效: {e}")
            return None
    
    async def blacklist_token(self, jti: str):
        """将令牌加入黑名单"""
        # 使用Redis存储黑名单，设置过期时间
        await redis_client.setex(
            f"blacklist:{jti}",
            self.refresh_token_expire.total_seconds(),
            "1"
        )
    
    async def _is_token_blacklisted(self, jti: str) -> bool:
        """检查令牌是否在黑名单中"""
        return await redis_client.exists(f"blacklist:{jti}")

# 权限装饰器
def require_permissions(*required_permissions: str):
    """权限检查装饰器"""
    def decorator(func):
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            # 从请求上下文获取用户权限
            user_permissions = get_current_user_permissions()
            
            # 检查权限
            if not all(perm in user_permissions for perm in required_permissions):
                raise HTTPException(
                    status_code=403,
                    detail="权限不足"
                )
            
            return await func(*args, **kwargs)
        return wrapper
    return decorator

# 使用示例
@require_permissions("graph:execute", "user:read")
async def execute_user_graph(graph_id: str, user_id: str):
    """执行用户图 - 需要图执行和用户读取权限"""
    pass