08 - Codex 实战案例集

from typing import List, Dict, TypedDict

class UserInput(TypedDict):
    """用户输入数据结构"""
    n: str  # name
    e: str  # email
    a: int  # age

class ProcessedUser(TypedDict):
    """处理后的用户数据结构"""
    n: str
    e: str
    a: int

ADULT_AGE_THRESHOLD = 18

def process_adult_users(users: List[UserInput]) -> List[ProcessedUser]:
    """
    处理成年用户数据
    
    过滤出年龄大于18岁的用户，并标准化其姓名和邮箱格式：
    - 姓名：转换为大写
    - 邮箱：转换为小写
    
    Args:
        users: 用户数据列表，每个用户包含姓名、邮箱和年龄
    
    Returns:
        处理后的成年用户列表
    
    Raises:
        ValueError: 如果用户数据格式无效
    
    Example:
        >>> users = [
        ...     {'n': 'Alice', 'e': 'ALICE@EXAMPLE.COM', 'a': 25},
        ...     {'n': 'Bob', 'e': 'bob@example.com', 'a': 17}
        ... ]
        >>> process_adult_users(users)
        [{'n': 'ALICE', 'e': 'alice@example.com', 'a': 25}]
    """
    if not isinstance(users, list):
        raise ValueError("users must be a list")
    
    processed_users = []
    
    for user in users:
        # 验证必需字段
        if not all(key in user for key in ['n', 'e', 'a']):
            raise ValueError(f"Invalid user data: {user}")
        
        # 过滤成年用户
        if user['a'] > ADULT_AGE_THRESHOLD:
            processed_users.append({
                'n': user['n'].upper(),
                'e': user['e'].lower(),
                'a': user['a']
            })
    
    return processed_users

# 或者使用列表推导式的简洁版本
def process_adult_users_compact(users: List[UserInput]) -> List[ProcessedUser]:
    """处理成年用户数据（简洁版本）"""
    return [
        {
            'n': user['n'].upper(),
            'e': user['e'].lower(),
            'a': user['a']
        }
        for user in users
        if user['a'] > ADULT_AGE_THRESHOLD
    ]

def process_users(users: List[dict]) -> List[dict]:
    """处理用户数据"""
    results = []
    
    for user in users:
        # 查询数据库获取额外信息
        profile = db.query(UserProfile).filter_by(
            user_id=user['id']
        ).first()
        
        # 调用外部 API
        score = requests.get(
            f"https://api.example.com/score/{user['id']}"
        ).json()['score']
        
        # 复杂计算
        result = calculate_user_metrics(user, profile, score)
        results.append(result)
    
    return results

import asyncio
import aiohttp
from sqlalchemy import select
from typing import List, Dict
from concurrent.futures import ProcessPoolExecutor

async def process_users_optimized(
    users: List[dict],
    batch_size: int = 1000
) -> List[dict]:
    """优化后的用户数据处理"""
    
    # 优化1：批量查询数据库（解决 N+1 问题）
    user_ids = [u['id'] for u in users]
    profiles = await fetch_profiles_batch(user_ids)
    profiles_map = {p.user_id: p for p in profiles}
    
    # 优化2：并发调用外部 API
    async with aiohttp.ClientSession() as session:
        scores = await fetch_scores_batch(session, user_ids, batch_size)
    
    # 优化3：并行计算（使用多进程）
    with ProcessPoolExecutor() as executor:
        tasks = [
            executor.submit(
                calculate_user_metrics,
                user,
                profiles_map.get(user['id']),
                scores.get(user['id'])
            )
            for user in users
        ]
        
        results = [task.result() for task in tasks]
    
    return results

async def fetch_profiles_batch(user_ids: List[int]) -> List[UserProfile]:
    """批量获取用户资料"""
    return await db.execute(
        select(UserProfile)
        .filter(UserProfile.user_id.in_(user_ids))
    ).scalars().all()

async def fetch_scores_batch(
    session: aiohttp.ClientSession,
    user_ids: List[int],
    batch_size: int
) -> Dict[int, float]:
    """批量获取评分（并发）"""
    scores = {}
    
    # 分批并发请求
    for i in range(0, len(user_ids), batch_size):
        batch = user_ids[i:i+batch_size]
        
        # 并发请求当前批次
        tasks = [
            fetch_score_single(session, uid)
            for uid in batch
        ]
        
        batch_scores = await asyncio.gather(*tasks)
        
        # 合并结果
        scores.update({
            uid: score
            for uid, score in zip(batch, batch_scores)
        })
    
    return scores

async def fetch_score_single(
    session: aiohttp.ClientSession,
    user_id: int
) -> float:
    """获取单个用户评分"""
    async with session.get(
        f"https://api.example.com/score/{user_id}"
    ) as response:
        data = await response.json()
        return data['score']

# 性能对比
"""
优化前：
- 数据库查询：100,000 × 10ms = 1000秒
- API 调用：100,000 × 50ms = 5000秒
- 计算：100,000 × 1ms = 100秒
- 总计：6100秒 (~102分钟)

优化后：
- 数据库查询：1次批量查询 = 1秒
- API 调用：100,000 / 100(并发) × 50ms = 50秒
- 计算：100,000 / 8(核心) × 1ms = 12.5秒
- 总计：63.5秒

提升：96倍！
"""