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