Python快速搭建一个高效的数据存储系统的实战指南

# 创建项目目录
mkdir data-storage-system && cd data-storage-system

# 创建虚拟环境
python -m venv venv

# 激活虚拟环境
# Windows
venv\Scripts\activate
# Linux/Mac
source venv/bin/activate

# 安装核心依赖
pip install sqlalchemy redis pymongo leveldb lmdb pandas numpy

# 安装开发工具
pip install black flake8 pytest python-dotenv

# 创建项目结构
mkdir -p app/{models,services,utils,config} tests data

# app/models/base.py
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, DateTime, Text, ForeignKey
from sqlalchemy.orm import relationship
import datetime

Base = declarative_base()

class User(Base):
    """用户数据模型"""
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True, autoincrement=True)
    username = Column(String(50), unique=True, nullable=False, index=True)
    email = Column(String(100), unique=True, nullable=False, index=True)
    password_hash = Column(String(255), nullable=False)
    created_at = Column(DateTime, default=datetime.datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.datetime.utcnow, 
                       onupdate=datetime.datetime.utcnow)
    
    # 关系定义
    posts = relationship("Post", back_populates="user")
    profiles = relationship("UserProfile", back_populates="user", uselist=False)

class UserProfile(Base):
    """用户配置文件模型"""
    __tablename__ = 'user_profiles'
    
    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey('users.id'), unique=True)
    full_name = Column(String(100))
    avatar_url = Column(String(255))
    bio = Column(Text)
    
    # 关系定义
    user = relationship("User", back_populates="profiles")

class Post(Base):
    """文章数据模型"""
    __tablename__ = 'posts'
    
    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey('users.id'))
    title = Column(String(200), nullable=False, index=True)
    content = Column(Text, nullable=False)
    status = Column(String(20), default='draft')  # draft, published, archived
    created_at = Column(DateTime, default=datetime.datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.datetime.utcnow, 
                       onupdate=datetime.datetime.utcnow)
    
    # 关系定义
    user = relationship("User", back_populates="posts")
    tags = relationship("Tag", secondary="post_tags", back_populates="posts")

class Tag(Base):
    """标签模型"""
    __tablename__ = 'tags'
    
    id = Column(Integer, primary_key=True)
    name = Column(String(50), unique=True, nullable=False, index=True)
    slug = Column(String(50), unique=True, nullable=False)
    
    posts = relationship("Post", secondary="post_tags", back_populates="tags")

class PostTag(Base):
    """文章标签关联表"""
    __tablename__ = 'post_tags'
    
    post_id = Column(Integer, ForeignKey('posts.id'), primary_key=True)
    tag_id = Column(Integer, ForeignKey('tags.id'), primary_key=True)
    created_at = Column(DateTime, default=datetime.datetime.utcnow)

# app/models/indexes.py
from sqlalchemy import Index

# 定义复合索引
user_email_index = Index('idx_user_email', User.email)
user_username_index = Index('idx_user_username', User.username)
post_title_index = Index('idx_post_title', Post.title)
post_user_status_index = Index('idx_post_user_status', Post.user_id, Post.status)
post_created_at_index = Index('idx_post_created_at', Post.created_at)

# 唯一索引约束
user_email_unique = Index('uq_user_email', User.email, unique=True)
user_username_unique = Index('uq_user_username', User.username, unique=True)
tag_name_unique = Index('uq_tag_name', Tag.name, unique=True)
tag_slug_unique = Index('uq_tag_slug', Tag.slug, unique=True)

# app/services/storage_engine.py
from abc import ABC, abstractmethod
from typing import Dict, Any, List, Optional
import json
import sqlite3
import leveldb
import redis
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.config.settings import Settings

class StorageEngine(ABC):
    """存储引擎抽象基类"""
    
    @abstractmethod
    def connect(self):
        """连接存储引擎"""
        pass
    
    @abstractmethod
    def disconnect(self):
        """断开连接"""
        pass
    
    @abstractmethod
    def create(self, key: str, data: Dict[str, Any]) -> bool:
        """创建数据"""
        pass
    
    @abstractmethod
    def read(self, key: str) -> Optional[Dict[str, Any]]:
        """读取数据"""
        pass
    
    @abstractmethod
    def update(self, key: str, data: Dict[str, Any]) -> bool:
        """更新数据"""
        pass
    
    @abstractmethod
    def delete(self, key: str) -> bool:
        """删除数据"""
        pass
    
    @abstractmethod
    def query(self, conditions: Dict[str, Any]) -> List[Dict[str, Any]]:
        """查询数据"""
        pass

class SQLiteStorage(StorageEngine):
    """SQLite存储引擎"""
    
    def __init__(self, db_path: str = "data/app.db"):
        self.db_path = db_path
        self.connection = None
    
    def connect(self):
        """连接SQLite数据库"""
        try:
            self.connection = sqlite3.connect(self.db_path)
            self.connection.row_factory = sqlite3.Row
            # 启用外键约束
            self.connection.execute("PRAGMA foreign_keys = ON")
            # 启用WAL模式提高并发性能
            self.connection.execute("PRAGMA journal_mode = WAL")
            return True
        except sqlite3.Error as e:
            print(f"SQLite连接失败: {e}")
            return False
    
    def disconnect(self):
        """断开连接"""
        if self.connection:
            self.connection.close()
    
    def create(self, key: str, data: Dict[str, Any]) -> bool:
        """创建数据"""
        try:
            table_name = key.split(':')[0]
            columns = ', '.join(data.keys())
            placeholders = ', '.join(['?'] * len(data))
            values = tuple(data.values())
            
            query = f"INSERT INTO {table_name} ({columns}) VALUES ({placeholders})"
            self.connection.execute(query, values)
            self.connection.commit()
            return True
        except sqlite3.Error as e:
            print(f"创建数据失败: {e}")
            return False
    
    def read(self, key: str) -> Optional[Dict[str, Any]]:
        """读取数据"""
        try:
            table_name, id_value = key.split(':')
            query = f"SELECT * FROM {table_name} WHERE id = ?"
            cursor = self.connection.execute(query, (id_value,))
            row = cursor.fetchone()
            return dict(row) if row else None
        except (sqlite3.Error, ValueError) as e:
            print(f"读取数据失败: {e}")
            return None
    
    def update(self, key: str, data: Dict[str, Any]) -> bool:
        """更新数据"""
        try:
            table_name, id_value = key.split(':')
            set_clause = ', '.join([f"{k} = ?" for k in data.keys()])
            values = tuple(data.values()) + (id_value,)
            
            query = f"UPDATE {table_name} SET {set_clause} WHERE id = ?"
            self.connection.execute(query, values)
            self.connection.commit()
            return True
        except (sqlite3.Error, ValueError) as e:
            print(f"更新数据失败: {e}")
            return False
    
    def delete(self, key: str) -> bool:
        """删除数据"""
        try:
            table_name, id_value = key.split(':')
            query = f"DELETE FROM {table_name} WHERE id = ?"
            self.connection.execute(query, (id_value,))
            self.connection.commit()
            return True
        except (sqlite3.Error, ValueError) as e:
            print(f"删除数据失败: {e}")
            return False
    
    def query(self, conditions: Dict[str, Any]) -> List[Dict[str, Any]]:
        """查询数据"""
        try:
            table_name = conditions.get('table')
            where_conditions = conditions.get('where', {})
            
            where_clause = ' AND '.join([f"{k} = ?" for k in where_conditions.keys()])
            values = tuple(where_conditions.values())
            
            query = f"SELECT * FROM {table_name}"
            if where_clause:
                query += f" WHERE {where_clause}"
            
            cursor = self.connection.execute(query, values)
            return [dict(row) for row in cursor.fetchall()]
        except sqlite3.Error as e:
            print(f"查询数据失败: {e}")
            return []

class RedisStorage(StorageEngine):
    """Redis存储引擎（用于缓存和高速访问）"""
    
    def __init__(self, host: str = 'localhost', port: int = 6379, db: int = 0):
        self.host = host
        self.port = port
        self.db = db
        self.client = None
    
    def connect(self):
        """连接Redis"""
        try:
            self.client = redis.Redis(
                host=self.host,
                port=self.port,
                db=self.db,
                decode_responses=True
            )
            # 测试连接
            return self.client.ping()
        except redis.RedisError as e:
            print(f"Redis连接失败: {e}")
            return False
    
    def disconnect(self):
        """断开连接"""
        if self.client:
            self.client.close()
    
    def create(self, key: str, data: Dict[str, Any], expire: int = None) -> bool:
        """创建数据"""
        try:
            serialized_data = json.dumps(data)
            if expire:
                return bool(self.client.setex(key, expire, serialized_data))
            else:
                return bool(self.client.set(key, serialized_data))
        except (redis.RedisError, TypeError) as e:
            print(f"Redis创建数据失败: {e}")
            return False
    
    def read(self, key: str) -> Optional[Dict[str, Any]]:
        """读取数据"""
        try:
            data = self.client.get(key)
            return json.loads(data) if data else None
        except (redis.RedisError, json.JSONDecodeError) as e:
            print(f"Redis读取数据失败: {e}")
            return None
    
    def update(self, key: str, data: Dict[str, Any]) -> bool:
        """更新数据 - Redis中set操作会自动覆盖"""
        return self.create(key, data)
    
    def delete(self, key: str) -> bool:
        """删除数据"""
        try:
            return bool(self.client.delete(key))
        except redis.RedisError as e:
            print(f"Redis删除数据失败: {e}")
            return False
    
    def query(self, conditions: Dict[str, Any]) -> List[Dict[str, Any]]:
        """查询数据 - Redis需要根据具体数据结构实现"""
        # 简化实现，实际中可能需要使用Redis的SCAN、SET等操作
        pattern = conditions.get('pattern', '*')
        try:
            keys = self.client.keys(pattern)
            results = []
            for key in keys:
                data = self.read(key)
                if data:
                    results.append(data)
            return results
        except redis.RedisError as e:
            print(f"Redis查询失败: {e}")
            return []

class StorageFactory:
    """存储引擎工厂"""
    
    @staticmethod
    def create_engine(engine_type: str, **kwargs) -> StorageEngine:
        """创建存储引擎实例"""
        engines = {
            'sqlite': SQLiteStorage,
            'redis': RedisStorage,
            # 可以扩展其他存储引擎
        }
        
        engine_class = engines.get(engine_type.lower())
        if not engine_class:
            raise ValueError(f"不支持的存储引擎类型: {engine_type}")
        
        return engine_class(**kwargs)

# app/config/database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session
from contextlib import contextmanager
import threading
from app.config.settings import Settings

class DatabaseManager:
    """数据库连接管理器"""
    
    _instance = None
    _lock = threading.Lock()
    
    def __new__(cls):
        with cls._lock:
            if cls._instance is None:
                cls._instance = super().__new__(cls)
                cls._instance._initialized = False
            return cls._instance
    
    def __init__(self):
        if self._initialized:
            return
        
        self.settings = Settings()
        self.engine = None
        self.session_factory = None
        self._initialized = True
    
    def init_app(self):
        """初始化数据库连接"""
        try:
            # 创建数据库引擎
            self.engine = create_engine(
                self.settings.DATABASE_URL,
                pool_size=self.settings.DB_POOL_SIZE,
                max_overflow=self.settings.DB_MAX_OVERFLOW,
                pool_timeout=self.settings.DB_POOL_TIMEOUT,
                pool_recycle=self.settings.DB_POOL_RECYCLE,
                echo=self.settings.DB_ECHO
            )
            
            # 创建会话工厂
            self.session_factory = sessionmaker(
                bind=self.engine,
                autocommit=False,
                autoflush=False,
                expire_on_commit=False
            )
            
            # 创建线程安全的scoped session
            self.ScopedSession = scoped_session(self.session_factory)
            
            print("数据库连接初始化成功")
            return True
            
        except Exception as e:
            print(f"数据库连接初始化失败: {e}")
            return False
    
    @contextmanager
    def get_session(self):
        """获取数据库会话上下文管理器"""
        session = self.ScopedSession()
        try:
            yield session
            session.commit()
        except Exception as e:
            session.rollback()
            raise e
        finally:
            session.close()
            self.ScopedSession.remove()
    
    def get_engine(self):
        """获取数据库引擎"""
        return self.engine
    
    def close(self):
        """关闭所有数据库连接"""
        if self.engine:
            self.engine.dispose()
        print("数据库连接已关闭")

# 全局数据库管理器实例
db_manager = DatabaseManager()

# app/config/settings.py
import os
from dataclasses import dataclass
from dotenv import load_dotenv

load_dotenv()

@dataclass
class Settings:
    """应用配置设置"""
    
    # 数据库配置
    DATABASE_URL: str = os.getenv('DATABASE_URL', 'sqlite:///data/app.db')
    DB_POOL_SIZE: int = int(os.getenv('DB_POOL_SIZE', 5))
    DB_MAX_OVERFLOW: int = int(os.getenv('DB_MAX_OVERFLOW', 10))
    DB_POOL_TIMEOUT: int = int(os.getenv('DB_POOL_TIMEOUT', 30))
    DB_POOL_RECYCLE: int = int(os.getenv('DB_POOL_RECYCLE', 3600))
    DB_ECHO: bool = os.getenv('DB_ECHO', 'False').lower() == 'true'
    
    # Redis配置
    REDIS_HOST: str = os.getenv('REDIS_HOST', 'localhost')
    REDIS_PORT: int = int(os.getenv('REDIS_PORT', 6379))
    REDIS_DB: int = int(os.getenv('REDIS_DB', 0))
    REDIS_PASSWORD: str = os.getenv('REDIS_PASSWORD', '')
    
    # 缓存配置
    CACHE_ENABLED: bool = os.getenv('CACHE_ENABLED', 'True').lower() == 'true'
    CACHE_TTL: int = int(os.getenv('CACHE_TTL', 300))  # 5分钟
    
    # 性能配置
    BATCH_SIZE: int = int(os.getenv('BATCH_SIZE', 1000))
    MAX_CONNECTIONS: int = int(os.getenv('MAX_CONNECTIONS', 100))
    
    # 应用配置
    DEBUG: bool = os.getenv('DEBUG', 'False').lower() == 'true'
    ENVIRONMENT: str = os.getenv('ENVIRONMENT', 'development')

# app/services/data_access.py
from typing import Type, TypeVar, List, Optional, Dict, Any
from abc import ABC, abstractmethod
from app.config.database import db_manager
from app.models.base import Base

T = TypeVar('T', bound=Base)

class BaseDAO(ABC):
    """数据访问对象基类"""
    
    def __init__(self, model_class: Type[T]):
        self.model_class = model_class
    
    @abstractmethod
    def create(self, obj: T) -> T:
        """创建对象"""
        pass
    
    @abstractmethod
    def get_by_id(self, id: int) -> Optional[T]:
        """根据ID获取对象"""
        pass
    
    @abstractmethod
    def get_all(self, skip: int = 0, limit: int = 100) -> List[T]:
        """获取所有对象"""
        pass
    
    @abstractmethod
    def update(self, obj: T) -> T:
        """更新对象"""
        pass
    
    @abstractmethod
    def delete(self, id: int) -> bool:
        """删除对象"""
        pass
    
    @abstractmethod
    def query(self, filters: Dict[str, Any], skip: int = 0, limit: int = 100) -> List[T]:
        """条件查询"""
        pass

class SQLAlchemyDAO(BaseDAO):
    """基于SQLAlchemy的数据访问对象"""
    
    def create(self, obj: T) -> T:
        with db_manager.get_session() as session:
            session.add(obj)
            session.flush()
            session.refresh(obj)
            return obj
    
    def get_by_id(self, id: int) -> Optional[T]:
        with db_manager.get_session() as session:
            return session.query(self.model_class).get(id)
    
    def get_all(self, skip: int = 0, limit: int = 100) -> List[T]:
        with db_manager.get_session() as session:
            return session.query(self.model_class).offset(skip).limit(limit).all()
    
    def update(self, obj: T) -> T:
        with db_manager.get_session() as session:
            session.merge(obj)
            session.flush()
            session.refresh(obj)
            return obj
    
    def delete(self, id: int) -> bool:
        with db_manager.get_session() as session:
            obj = session.query(self.model_class).get(id)
            if obj:
                session.delete(obj)
                return True
            return False
    
    def query(self, filters: Dict[str, Any], skip: int = 0, limit: int = 100) -> List[T]:
        with db_manager.get_session() as session:
            query = session.query(self.model_class)
            
            for field, value in filters.items():
                if hasattr(self.model_class, field):
                    if isinstance(value, (list, tuple)):
                        query = query.filter(getattr(self.model_class, field).in_(value))
                    else:
                        query = query.filter(getattr(self.model_class, field) == value)
            
            return query.offset(skip).limit(limit).all()

# 具体DAO实现
class UserDAO(SQLAlchemyDAO):
    def __init__(self):
        super().__init__(User)
    
    def get_by_email(self, email: str) -> Optional[User]:
        with db_manager.get_session() as session:
            return session.query(User).filter(User.email == email).first()
    
    def get_by_username(self, username: str) -> Optional[User]:
        with db_manager.get_session() as session:
            return session.query(User).filter(User.username == username).first()

class PostDAO(SQLAlchemyDAO):
    def __init__(self):
        super().__init__(Post)
    
    def get_published_posts(self, skip: int = 0, limit: int = 100) -> List[Post]:
        with db_manager.get_session() as session:
            return session.query(Post).filter(
                Post.status == 'published'
            ).order_by(
                Post.created_at.desc()
            ).offset(skip).limit(limit).all()
    
    def get_posts_by_user(self, user_id: int, skip: int = 0, limit: int = 100) -> List[Post]:
        with db_manager.get_session() as session:
            return session.query(Post).filter(
                Post.user_id == user_id
            ).order_by(
                Post.created_at.desc()
            ).offset(skip).limit(limit).all()

# app/services/cache_service.py
from typing import Optional, Any, Callable
import pickle
import hashlib
import json
from functools import wraps
from app.config.settings import Settings
from app.services.storage_engine import StorageFactory

class CacheService:
    """缓存服务"""
    
    def __init__(self):
        self.settings = Settings()
        self.engine = None
        self._init_cache_engine()
    
    def _init_cache_engine(self):
        """初始化缓存引擎"""
        if self.settings.CACHE_ENABLED:
            try:
                self.engine = StorageFactory.create_engine(
                    'redis',
                    host=self.settings.REDIS_HOST,
                    port=self.settings.REDIS_PORT,
                    db=self.settings.REDIS_DB
                )
                if not self.engine.connect():
                    print("缓存连接失败，将禁用缓存功能")
                    self.engine = None
            except Exception as e:
                print(f"缓存初始化失败: {e}")
                self.engine = None
    
    def generate_cache_key(self, func: Callable, *args, **kwargs) -> str:
        """生成缓存键"""
        # 基于函数名和参数生成唯一的缓存键
        key_parts = [func.__module__, func.__name__]
        
        # 添加参数信息
        if args:
            key_parts.append(str(args))
        if kwargs:
            key_parts.append(str(sorted(kwargs.items())))
        
        # 生成MD5哈希
        key_string = ':'.join(key_parts)
        return f"cache:{hashlib.md5(key_string.encode()).hexdigest()}"
    
    def get(self, key: str) -> Optional[Any]:
        """获取缓存数据"""
        if not self.engine or not self.settings.CACHE_ENABLED:
            return None
        
        try:
            data = self.engine.read(key)
            if data:
                return pickle.loads(data.encode('latin1')) if isinstance(data, str) else data
        except Exception as e:
            print(f"缓存获取失败: {e}")
        return None
    
    def set(self, key: str, value: Any, expire: int = None) -> bool:
        """设置缓存数据"""
        if not self.engine or not self.settings.CACHE_ENABLED:
            return False
        
        try:
            # 使用pickle序列化复杂对象
            serialized_value = pickle.dumps(value).decode('latin1')
            ttl = expire or self.settings.CACHE_TTL
            return self.engine.create(key, serialized_value, ttl)
        except Exception as e:
            print(f"缓存设置失败: {e}")
            return False
    
    def delete(self, key: str) -> bool:
        """删除缓存数据"""
        if not self.engine or not self.settings.CACHE_ENABLED:
            return False
        
        try:
            return self.engine.delete(key)
        except Exception as e:
            print(f"缓存删除失败: {e}")
            return False
    
    def clear_pattern(self, pattern: str) -> int:
        """清除匹配模式的缓存"""
        if not self.engine or not self.settings.CACHE_ENABLED:
            return 0
        
        try:
            # 这里需要根据具体的存储引擎实现模式删除
            # 对于Redis，可以使用keys+delete组合操作
            results = self.engine.query({'pattern': pattern})
            count = 0
            for item in results:
                if self.delete(item.get('key')):
                    count += 1
            return count
        except Exception as e:
            print(f"缓存清除失败: {e}")
            return 0

def cached(expire: int = None):
    """缓存装饰器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            cache_service = CacheService()
            cache_key = cache_service.generate_cache_key(func, *args, **kwargs)
            
            # 尝试从缓存获取
            cached_result = cache_service.get(cache_key)
            if cached_result is not None:
                return cached_result
            
            # 执行函数并缓存结果
            result = func(*args, **kwargs)
            cache_service.set(cache_key, result, expire)
            
            return result
        return wrapper
    return decorator

# 全局缓存服务实例
cache_service = CacheService()

# app/utils/query_optimizer.py
from sqlalchemy.orm import Query, joinedload, selectinload
from typing import List, Dict, Any
from app.services.cache_service import cached

class QueryOptimizer:
    """查询优化器"""
    
    @staticmethod
    def optimize_query(query: Query, options: List[Any] = None) -> Query:
        """优化SQL查询"""
        if options:
            query = query.options(*options)
        
        # 添加其他优化策略
        return query
    
    @staticmethod
    def eager_load_relationships(query: Query, model_class, relationships: List[str]) -> Query:
        """预加载关联关系"""
        options = []
        for rel in relationships:
            if hasattr(model_class, rel):
                # 根据关系类型选择合适的加载策略
                options.append(joinedload(getattr(model_class, rel)))
        
        return query.options(*options)
    
    @staticmethod
    def apply_filters(query: Query, filters: Dict[str, Any]) -> Query:
        """应用过滤条件"""
        for field, value in filters.items():
            if hasattr(query.column_descriptions[0]['type'], field):
                if isinstance(value, (list, tuple)):
                    query = query.filter(getattr(query.column_descriptions[0]['type'], field).in_(value))
                else:
                    query = query.filter(getattr(query.column_descriptions[0]['type'], field) == value)
        return query

# 优化后的DAO方法
class OptimizedPostDAO(PostDAO):
    
    @cached(expire=300)  # 缓存5分钟
    def get_published_posts_with_authors(self, skip: int = 0, limit: int = 100) -> List[Post]:
        """获取已发布文章及其作者信息（优化版）"""
        with db_manager.get_session() as session:
            query = session.query(Post).filter(Post.status == 'published')
            query = QueryOptimizer.eager_load_relationships(query, Post, ['user'])
            query = query.order_by(Post.created_at.desc())
            query = query.offset(skip).limit(limit)
            
            return query.all()

# app/utils/batch_processor.py
from typing import List, Callable, Any
from concurrent.futures import ThreadPoolExecutor, as_completed
from app.config.settings import Settings

class BatchProcessor:
    """批量处理器"""
    
    def __init__(self, max_workers: int = None):
        self.settings = Settings()
        self.max_workers = max_workers or self.settings.MAX_CONNECTIONS
    
    def process_batch(self, items: List[Any], process_func: Callable, 
                     batch_size: int = None) -> List[Any]:
        """处理批量数据"""
        batch_size = batch_size or self.settings.BATCH_SIZE
        results = []
        
        for i in range(0, len(items), batch_size):
            batch = items[i:i + batch_size]
            batch_results = self._process_batch_concurrently(batch, process_func)
            results.extend(batch_results)
        
        return results
    
    def _process_batch_concurrently(self, batch: List[Any], process_func: Callable) -> List[Any]:
        """并发处理单个批次"""
        results = []
        
        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            # 提交所有任务
            future_to_item = {
                executor.submit(process_func, item): item for item in batch
            }
            
            # 收集结果
            for future in as_completed(future_to_item):
                try:
                    result = future.result()
                    results.append(result)
                except Exception as e:
                    print(f"处理失败: {e}")
                    # 可以根据需要记录失败的项目
        
        return results
    
    def batch_insert(self, items: List[Any], dao: Any) -> List[Any]:
        """批量插入数据"""
        def insert_item(item):
            return dao.create(item)
        
        return self.process_batch(items, insert_item)
    
    def batch_update(self, items: List[Any], dao: Any) -> List[Any]:
        """批量更新数据"""
        def update_item(item):
            return dao.update(item)
        
        return self.process_batch(items, update_item)

# app/main.py
from app.config.database import db_manager
from app.config.settings import Settings
from app.models.base import Base
from app.services.data_access import UserDAO, PostDAO
from app.utils.batch_processor import BatchProcessor
from app.services.cache_service import cache_service
import time

def init_database():
    """初始化数据库"""
    print("初始化数据库...")
    
    # 创建数据库表
    engine = db_manager.get_engine()
    Base.metadata.create_all(engine)
    print("数据库表创建完成")
    
    # 插入示例数据
    insert_sample_data()

def insert_sample_data():
    """插入示例数据"""
    print("插入示例数据...")
    
    user_dao = UserDAO()
    post_dao = PostDAO()
    batch_processor = BatchProcessor()
    
    # 创建示例用户
    users = [
        User(username=f"user{i}", email=f"user{i}@example.com", 
             password_hash=f"hash{i}") for i in range(1, 6)
    ]
    
    created_users = batch_processor.batch_insert(users, user_dao)
    print(f"创建了 {len(created_users)} 个用户")
    
    # 创建示例文章
    posts = []
    for user in created_users:
        for j in range(1, 4):
            posts.append(
                Post(
                    user_id=user.id,
                    title=f"文章 {j} by {user.username}",
                    content=f"这是 {user.username} 的第 {j} 篇文章内容",
                    status='published' if j % 2 == 0 else 'draft'
                )
            )
    
    created_posts = batch_processor.batch_insert(posts, post_dao)
    print(f"创建了 {len(created_posts)} 篇文章")

def benchmark_performance():
    """性能基准测试"""
    print("\n性能基准测试...")
    
    post_dao = PostDAO()
    
    # 测试无缓存性能
    start_time = time.time()
    for _ in range(10):
        posts = post_dao.get_published_posts()
    uncached_time = time.time() - start_time
    print(f"无缓存查询时间: {uncached_time:.4f}秒")
    
    # 测试有缓存性能
    start_time = time.time()
    for _ in range(10):
        posts = post_dao.get_published_posts_with_authors()
    cached_time = time.time() - start_time
    print(f"有缓存查询时间: {cached_time:.4f}秒")
    
    print(f"性能提升: {uncached_time/cached_time:.2f}倍")

def main():
    """主函数"""
    settings = Settings()
    
    try:
        # 初始化应用
        print("启动数据存储系统...")
        print(f"环境: {settings.ENVIRONMENT}")
        print(f"调试模式: {settings.DEBUG}")
        
        # 初始化数据库连接
        if not db_manager.init_app():
            print("数据库初始化失败，退出应用")
            return
        
        # 初始化数据库表和数据
        init_database()
        
        # 性能测试
        benchmark_performance()
        
        # 演示数据访问
        print("\n演示数据访问...")
        user_dao = UserDAO()
        post_dao = PostDAO()
        
        # 查询用户
        users = user_dao.get_all(limit=3)
        print(f"前3个用户: {[u.username for u in users]}")
        
        # 查询文章
        posts = post_dao.get_published_posts(limit=5)
        print(f"已发布文章: {[p.title for p in posts]}")
        
        print("\n应用运行完成！")
        
    except Exception as e:
        print(f"应用运行出错: {e}")
    finally:
        # 清理资源
        db_manager.close()
        if cache_service.engine:
            cache_service.engine.disconnect()

if __name__ == "__main__":
    main()

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

# 安装系统依赖
RUN apt-get update && apt-get install -y \
    gcc \
    libsqlite3-dev \
    && rm -rf /var/lib/apt/lists/*

# 复制依赖文件
COPY requirements.txt .

# 安装Python依赖
RUN pip install --no-cache-dir -r requirements.txt

# 复制应用代码
COPY . .

# 创建数据目录
RUN mkdir -p data

# 创建非root用户
RUN useradd -m -u 1000 appuser && chown -R appuser:appuser /app
USER appuser

# 暴露端口
EXPOSE 8000

# 启动命令
CMD ["python", "app/main.py"]

# .env
# 数据库配置
DATABASE_URL=sqlite:///data/app.db
DB_POOL_SIZE=10
DB_MAX_OVERFLOW=20
DB_POOL_TIMEOUT=30
DB_POOL_RECYCLE=3600
DB_ECHO=False

# Redis配置
REDIS_HOST=localhost
REDIS_PORT=6379
REDIS_DB=0
REDIS_PASSWORD=

# 缓存配置
CACHE_ENABLED=True
CACHE_TTL=300

# 性能配置
BATCH_SIZE=1000
MAX_CONNECTIONS=50

# 应用配置
DEBUG=False
ENVIRONMENT=production