Python脚本命令查询模式分离读写

本文目录导读：

1. 方案1：基础读写分离装饰器
2. 方案2：基于命令模式实现
3. 方案3：支持读写分离的数据库操作
4. 方案4：基于SQLAlchemy的读写分离
5. 使用建议

我来帮你实现一个Python命令查询模式下的读写分离功能,这里提供几种常见的实现方案：

方案1：基础读写分离装饰器

import functools
from enum import Enum
class Mode(Enum):
    READ = "read"
    WRITE = "write"
class ReadWriteSeparator:
    def __init__(self):
        self._current_mode = Mode.READ
    @property
    def current_mode(self):
        return self._current_mode
    @current_mode.setter
    def current_mode(self, mode):
        if mode not in Mode:
            raise ValueError(f"Invalid mode: {mode}")
        self._current_mode = mode
    def read(self, func):
        """装饰读操作"""
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            if self._current_mode != Mode.READ:
                raise RuntimeError("当前不是读模式")
            return func(*args, **kwargs)
        return wrapper
    def write(self, func):
        """装饰写操作"""
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            if self._current_mode != Mode.WRITE:
                raise RuntimeError("当前不是写模式")
            return func(*args, **kwargs)
        return wrapper
# 使用示例
separator = ReadWriteSeparator()
class DataManager:
    def __init__(self):
        self.data = {}
    @separator.read
    def get_data(self, key):
        return self.data.get(key)
    @separator.write
    def set_data(self, key, value):
        self.data[key] = value
    @separator.read
    def list_keys(self):
        return list(self.data.keys())
# 测试
dm = DataManager()
separator.current_mode = Mode.WRITE
dm.set_data("name", "Alice")  # 可以写入
separator.current_mode = Mode.READ
print(dm.get_data("name"))  # 可以读取
try:
    dm.set_data("age", 25)  # 会抛出异常
except RuntimeError as e:
    print(f"错误: {e}")

方案2：基于命令模式实现

from abc import ABC, abstractmethod
from typing import Any, Dict, List
class Command(ABC):
    """命令基类"""
    @abstractmethod
    def execute(self):
        pass
    @abstractmethod
    def is_readonly(self) -> bool:
        pass
class ReadCommand(Command):
    """读命令基类"""
    def is_readonly(self):
        return True
class WriteCommand(Command):
    """写命令基类"""
    def is_readonly(self):
        return False
class DataStore:
    def __init__(self):
        self._data: Dict[str, Any] = {}
        self._read_commands = []
        self._write_commands = []
    def execute(self, command: Command):
        """执行命令"""
        if command.is_readonly():
            self._read_commands.append(command)
            return command.execute()
        else:
            self._write_commands.append(command)
            return command.execute()
    def get_data(self, key: str) -> Any:
        """创建读命令"""
        class GetCommand(ReadCommand):
            def execute(self):
                return self._data.get(key)
        return self.execute(GetCommand())
    def set_data(self, key: str, value: Any):
        """创建写命令"""
        class SetCommand(WriteCommand):
            def execute(self):
                self._data[key] = value
                return True
        return self.execute(SetCommand())
    def get_command_history(self) -> Dict[str, List]:
        """获取命令历史"""
        return {
            "read": self._read_commands,
            "write": self._write_commands
        }
# 使用示例
store = DataStore()
store.set_data("name", "Bob")
store.set_data("age", 30)
print(store.get_data("name"))  # Bob
print(store.get_data("city"))  # None

方案3：支持读写分离的数据库操作

import sqlite3
from contextlib import contextmanager
from typing import Optional
class DatabaseReadWriteSeparator:
    """数据库读写分离管理器"""
    def __init__(self, read_db_path: str, write_db_path: str):
        self._read_db_path = read_db_path
        self._write_db_path = write_db_path
        self._connections = {}
    @contextmanager
    def read_connection(self):
        """获取读连接"""
        conn = sqlite3.connect(self._read_db_path)
        self._connections['read'] = conn
        try:
            yield conn
        finally:
            conn.close()
            self._connections.pop('read', None)
    @contextmanager
    def write_connection(self):
        """获取写连接"""
        conn = sqlite3.connect(self._write_db_path)
        self._connections['write'] = conn
        try:
            yield conn
        finally:
            conn.close()
            self._connections.pop('write', None)
    def execute_read(self, query: str, params: tuple = ()) -> Optional[list]:
        """执行读操作"""
        with self.read_connection() as conn:
            cursor = conn.cursor()
            cursor.execute(query, params)
            return cursor.fetchall()
    def execute_write(self, query: str, params: tuple = ()) -> bool:
        """执行写操作"""
        with self.write_connection() as conn:
            cursor = conn.cursor()
            cursor.execute(query, params)
            conn.commit()
            return True
    def sync_databases(self):
        """同步读写数据库（示例功能）"""
        # 这里可以实现主从同步逻辑
        pass
# 使用示例
class UserRepository:
    def __init__(self, db_separator: DatabaseReadWriteSeparator):
        self.db = db_separator
    def get_user(self, user_id: int):
        """读操作"""
        query = "SELECT * FROM users WHERE id = ?"
        return self.db.execute_read(query, (user_id,))
    def create_user(self, name: str, email: str):
        """写操作"""
        query = "INSERT INTO users (name, email) VALUES (?, ?)"
        return self.db.execute_write(query, (name, email))
    def update_user(self, user_id: int, name: str):
        """写操作"""
        query = "UPDATE users SET name = ? WHERE id = ?"
        return self.db.execute_write(query, (name, user_id))

方案4：基于SQLAlchemy的读写分离

from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker, Session
from typing import Optional
class SQLAlchemyReadWriteSeparator:
    """SQLAlchemy读写分离封装"""
    def __init__(self, read_url: str, write_url: str):
        self._read_engine = create_engine(read_url)
        self._write_engine = create_engine(write_url)
        self._ReadSession = sessionmaker(bind=self._read_engine)
        self._WriteSession = sessionmaker(bind=self._write_engine)
    @contextmanager
    def read_session(self) -> Session:
        """获取读会话"""
        session = self._ReadSession()
        try:
            yield session
        finally:
            session.close()
    @contextmanager
    def write_session(self) -> Session:
        """获取写会话"""
        session = self._WriteSession()
        try:
            yield session
            session.commit()
        except Exception:
            session.rollback()
            raise
        finally:
            session.close()
    def execute_read_query(self, sql: str, params: dict = None) -> list:
        """执行读SQL"""
        with self.read_session() as session:
            result = session.execute(text(sql), params or {})
            return result.fetchall()
    def execute_write_query(self, sql: str, params: dict = None) -> int:
        """执行写SQL"""
        with self.write_session() as session:
            result = session.execute(text(sql), params or {})
            return result.rowcount

使用建议

1. 选择方案：根据你的具体需求选择合适的方案
2. 异常处理：添加适当的异常处理机制
3. 日志记录：记录读写操作日志以便调试
4. 性能监控：监控读写分离的性能效果
5. 数据一致性：确保读写分离后的数据一致性

需要我详细解释某个特定方案或提供更多示例吗？