# 創(chuàng)建項目目錄
mkdir data-storage-system && cd data-storage-system

# 創(chuàng)建虛擬環(huán)境
python -m venv venv

# 激活虛擬環(huán)境
# Windows
venv\Scripts\activate
# Linux/Mac
source venv/bin/activate

# 安裝核心依賴
pip install sqlalchemy redis pymongo leveldb lmdb pandas numpy

# 安裝開發(fā)工具
pip install black flake8 pytest python-dotenv

# 創(chuàng)建項目結構
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):
    """用戶數(shù)據(jù)模型"""
    __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):
    """文章數(shù)據(jù)模型"""
    __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):
    """文章標簽關聯(lián)表"""
    __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:
        """創(chuàng)建數(shù)據(jù)"""
        pass
    
    @abstractmethod
    def read(self, key: str) -> Optional[Dict[str, Any]]:
        """讀取數(shù)據(jù)"""
        pass
    
    @abstractmethod
    def update(self, key: str, data: Dict[str, Any]) -> bool:
        """更新數(shù)據(jù)"""
        pass
    
    @abstractmethod
    def delete(self, key: str) -> bool:
        """刪除數(shù)據(jù)"""
        pass
    
    @abstractmethod
    def query(self, conditions: Dict[str, Any]) -> List[Dict[str, Any]]:
        """查詢數(shù)據(jù)"""
        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數(shù)據(jù)庫"""
        try:
            self.connection = sqlite3.connect(self.db_path)
            self.connection.row_factory = sqlite3.Row
            # 啟用外鍵約束
            self.connection.execute("PRAGMA foreign_keys = ON")
            # 啟用WAL模式提高并發(fā)性能
            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:
        """創(chuàng)建數(shù)據(jù)"""
        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"創(chuàng)建數(shù)據(jù)失敗: {e}")
            return False
    
    def read(self, key: str) -> Optional[Dict[str, Any]]:
        """讀取數(shù)據(jù)"""
        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"讀取數(shù)據(jù)失敗: {e}")
            return None
    
    def update(self, key: str, data: Dict[str, Any]) -> bool:
        """更新數(shù)據(jù)"""
        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"更新數(shù)據(jù)失敗: {e}")
            return False
    
    def delete(self, key: str) -> bool:
        """刪除數(shù)據(jù)"""
        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"刪除數(shù)據(jù)失敗: {e}")
            return False
    
    def query(self, conditions: Dict[str, Any]) -> List[Dict[str, Any]]:
        """查詢數(shù)據(jù)"""
        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"查詢數(shù)據(jù)失敗: {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:
        """創(chuàng)建數(shù)據(jù)"""
        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創(chuàng)建數(shù)據(jù)失敗: {e}")
            return False
    
    def read(self, key: str) -> Optional[Dict[str, Any]]:
        """讀取數(shù)據(jù)"""
        try:
            data = self.client.get(key)
            return json.loads(data) if data else None
        except (redis.RedisError, json.JSONDecodeError) as e:
            print(f"Redis讀取數(shù)據(jù)失敗: {e}")
            return None
    
    def update(self, key: str, data: Dict[str, Any]) -> bool:
        """更新數(shù)據(jù) - Redis中set操作會自動覆蓋"""
        return self.create(key, data)
    
    def delete(self, key: str) -> bool:
        """刪除數(shù)據(jù)"""
        try:
            return bool(self.client.delete(key))
        except redis.RedisError as e:
            print(f"Redis刪除數(shù)據(jù)失敗: {e}")
            return False
    
    def query(self, conditions: Dict[str, Any]) -> List[Dict[str, Any]]:
        """查詢數(shù)據(jù) - Redis需要根據(jù)具體數(shù)據(jù)結構實現(xiàn)"""
        # 簡化實現(xiàn)，實際中可能需要使用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:
        """創(chuàng)建存儲引擎實例"""
        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:
    """數(shù)據(jù)庫連接管理器"""
    
    _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):
        """初始化數(shù)據(jù)庫連接"""
        try:
            # 創(chuàng)建數(shù)據(jù)庫引擎
            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
            )
            
            # 創(chuàng)建會話工廠
            self.session_factory = sessionmaker(
                bind=self.engine,
                autocommit=False,
                autoflush=False,
                expire_on_commit=False
            )
            
            # 創(chuàng)建線程安全的scoped session
            self.ScopedSession = scoped_session(self.session_factory)
            
            print("數(shù)據(jù)庫連接初始化成功")
            return True
            
        except Exception as e:
            print(f"數(shù)據(jù)庫連接初始化失敗: {e}")
            return False
    
    @contextmanager
    def get_session(self):
        """獲取數(shù)據(jù)庫會話上下文管理器"""
        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):
        """獲取數(shù)據(jù)庫引擎"""
        return self.engine
    
    def close(self):
        """關閉所有數(shù)據(jù)庫連接"""
        if self.engine:
            self.engine.dispose()
        print("數(shù)據(jù)庫連接已關閉")

# 全局數(shù)據(jù)庫管理器實例
db_manager = DatabaseManager()

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

load_dotenv()

@dataclass
class Settings:
    """應用配置設置"""
    
    # 數(shù)據(jù)庫配置
    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):
    """數(shù)據(jù)訪問對象基類"""
    
    def __init__(self, model_class: Type[T]):
        self.model_class = model_class
    
    @abstractmethod
    def create(self, obj: T) -> T:
        """創(chuàng)建對象"""
        pass
    
    @abstractmethod
    def get_by_id(self, id: int) -> Optional[T]:
        """根據(jù)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的數(shù)據(jù)訪問對象"""
    
    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實現(xiàn)
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:
        """生成緩存鍵"""
        # 基于函數(shù)名和參數(shù)生成唯一的緩存鍵
        key_parts = [func.__module__, func.__name__]
        
        # 添加參數(shù)信息
        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]:
        """獲取緩存數(shù)據(jù)"""
        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:
        """設置緩存數(shù)據(jù)"""
        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:
        """刪除緩存數(shù)據(jù)"""
        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:
            # 這里需要根據(jù)具體的存儲引擎實現(xiàn)模式刪除
            # 對于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
            
            # 執(zhí)行函數(shù)并緩存結果
            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:
    """查詢優(yōu)化器"""
    
    @staticmethod
    def optimize_query(query: Query, options: List[Any] = None) -> Query:
        """優(yōu)化SQL查詢"""
        if options:
            query = query.options(*options)
        
        # 添加其他優(yōu)化策略
        return query
    
    @staticmethod
    def eager_load_relationships(query: Query, model_class, relationships: List[str]) -> Query:
        """預加載關聯(lián)關系"""
        options = []
        for rel in relationships:
            if hasattr(model_class, rel):
                # 根據(jù)關系類型選擇合適的加載策略
                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

# 優(yōu)化后的DAO方法
class OptimizedPostDAO(PostDAO):
    
    @cached(expire=300)  # 緩存5分鐘
    def get_published_posts_with_authors(self, skip: int = 0, limit: int = 100) -> List[Post]:
        """獲取已發(fā)布文章及其作者信息（優(yōu)化版）"""
        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]:
        """處理批量數(shù)據(jù)"""
        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]:
        """并發(fā)處理單個批次"""
        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}")
                    # 可以根據(jù)需要記錄失敗的項目
        
        return results
    
    def batch_insert(self, items: List[Any], dao: Any) -> List[Any]:
        """批量插入數(shù)據(jù)"""
        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]:
        """批量更新數(shù)據(jù)"""
        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():
    """初始化數(shù)據(jù)庫"""
    print("初始化數(shù)據(jù)庫...")
    
    # 創(chuàng)建數(shù)據(jù)庫表
    engine = db_manager.get_engine()
    Base.metadata.create_all(engine)
    print("數(shù)據(jù)庫表創(chuàng)建完成")
    
    # 插入示例數(shù)據(jù)
    insert_sample_data()

def insert_sample_data():
    """插入示例數(shù)據(jù)"""
    print("插入示例數(shù)據(jù)...")
    
    user_dao = UserDAO()
    post_dao = PostDAO()
    batch_processor = BatchProcessor()
    
    # 創(chuàng)建示例用戶
    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"創(chuàng)建了 {len(created_users)} 個用戶")
    
    # 創(chuàng)建示例文章
    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} 篇文章內(nèi)容",
                    status='published' if j % 2 == 0 else 'draft'
                )
            )
    
    created_posts = batch_processor.batch_insert(posts, post_dao)
    print(f"創(chuàng)建了 {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():
    """主函數(shù)"""
    settings = Settings()
    
    try:
        # 初始化應用
        print("啟動數(shù)據(jù)存儲系統(tǒng)...")
        print(f"環(huán)境: {settings.ENVIRONMENT}")
        print(f"調(diào)試模式: {settings.DEBUG}")
        
        # 初始化數(shù)據(jù)庫連接
        if not db_manager.init_app():
            print("數(shù)據(jù)庫初始化失敗，退出應用")
            return
        
        # 初始化數(shù)據(jù)庫表和數(shù)據(jù)
        init_database()
        
        # 性能測試
        benchmark_performance()
        
        # 演示數(shù)據(jù)訪問
        print("\n演示數(shù)據(jù)訪問...")
        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"已發(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

# 安裝系統(tǒng)依賴
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 . .

# 創(chuàng)建數(shù)據(jù)目錄
RUN mkdir -p data

# 創(chuàng)建非root用戶
RUN useradd -m -u 1000 appuser && chown -R appuser:appuser /app
USER appuser

# 暴露端口
EXPOSE 8000

# 啟動命令
CMD ["python", "app/main.py"]

# .env
# 數(shù)據(jù)庫配置
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