Python MySQL 进阶用法详解

1. 使用连接池

使用 DBUtils 实现连接池管理：

from dbutils.pooled_db import PooledDB
import pymysql

class DBConnectionPool:
    _pool = None
    
    @staticmethod
    def get_pool():
        if DBConnectionPool._pool is None:
            DBConnectionPool._pool = PooledDB(
                creator=pymysql,        # 使用pymysql作为连接器
                maxconnections=10,      # 连接池最大连接数
                mincached=2,           # 初始化时创建的空闲连接数
                maxcached=5,           # 连接池最大空闲连接数
                maxshared=3,           # 共享连接数
                blocking=True,         # 连接数达到最大时是否阻塞
                maxusage=None,         # 一个连接最多被使用的次数
                setsession=[],         # 开始会话前执行的命令
                ping=0,                # ping MySQL服务端确保连接有效
                host='localhost',
                port=3306,
                user='root',
                password='123456',
                database='test',
                charset='utf8mb4',
                cursorclass=pymysql.cursors.DictCursor
            )
        return DBConnectionPool._pool

    @staticmethod
    def get_conn():
        return DBConnectionPool.get_pool().connection()

# 使用示例
class UserDAO:
    def get_user(self, user_id):
        with DBConnectionPool.get_conn() as conn:
            with conn.cursor() as cursor:
                sql = "SELECT * FROM users WHERE id = %s"
                cursor.execute(sql, (user_id,))
                return cursor.fetchone()

2. 实现简单的 ORM 映射

from typing import Dict, Any, Type, TypeVar
from datetime import datetime

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

class Field:
    def __init__(self, field_type: str, primary_key: bool = False):
        self.field_type = field_type
        self.primary_key = primary_key

class BaseModel:
    _table_name: str = ''
    _fields: Dict[str, Field] = {}
    
    def __init__(self, **kwargs):
        for key, value in kwargs.items():
            setattr(self, key, value)
    
    @classmethod
    def from_db_dict(cls: Type[T], db_dict: Dict[str, Any]) -> T:
        """从数据库字典创建对象"""
        return cls(**db_dict)
    
    def to_db_dict(self) -> Dict[str, Any]:
        """转换为数据库字典"""
        result = {}
        for key in self._fields:
            if hasattr(self, key):
                result[key] = getattr(self, key)
        return result

class User(BaseModel):
    _table_name = 'users'
    _fields = {
        'id': Field('BIGINT', primary_key=True),
        'username': Field('VARCHAR(50)'),
        'password': Field('VARCHAR(100)'),
        'age': Field('INT'),
        'create_at': Field('TIMESTAMP')
    }
    
    def __init__(self, username: str, password: str, age: int, 
                 id: int = None, create_at: datetime = None):
        self.id = id
        self.username = username
        self.password = password
        self.age = age
        self.create_at = create_at

class UserRepository:
    def __init__(self, db_pool):
        self.db_pool = db_pool
    
    def save(self, user: User) -> User:
        with self.db_pool.get_conn() as conn:
            with conn.cursor() as cursor:
                if user.id is None:
                    # Insert
                    sql = """INSERT INTO users (username, password, age) 
                            VALUES (%s, %s, %s)"""
                    cursor.execute(sql, (user.username, user.password, user.age))
                    user.id = cursor.lastrowid
                else:
                    # Update
                    sql = """UPDATE users SET username=%s, password=%s, age=%s 
                            WHERE id=%s"""
                    cursor.execute(sql, (user.username, user.password, 
                                      user.age, user.id))
                conn.commit()
                return user

# 使用示例
user = User(username="张三", password="123456", age=25)
repo = UserRepository(DBConnectionPool())
saved_user = repo.save(user)

3. 读写分离实现

from enum import Enum
from typing import List
import random

class DBType(Enum):
    MASTER = "master"
    SLAVE = "slave"

class DBConfig:
    def __init__(self, host: str, port: int, db_type: DBType):
        self.host = host
        self.port = port
        self.db_type = db_type

class DBRouter:
    def __init__(self):
        self.master_config = DBConfig("master.mysql", 3306, DBType.MASTER)
        self.slave_configs: List[DBConfig] = [
            DBConfig("slave1.mysql", 3306, DBType.SLAVE),
            DBConfig("slave2.mysql", 3306, DBType.SLAVE),
        ]
        
        # 创建连接池
        self.master_pool = self._create_pool(self.master_config)
        self.slave_pools = [self._create_pool(cfg) for cfg in self.slave_configs]
    
    def _create_pool(self, config: DBConfig):
        return PooledDB(
            creator=pymysql,
            maxconnections=10,
            host=config.host,
            port=config.port,
            user='root',
            password='123456',
            database='test',
            charset='utf8mb4'
        )
    
    def get_connection(self, for_write: bool = False):
        if for_write:
            return self.master_pool.connection()
        # 随机选择一个从库
        slave_pool = random.choice(self.slave_pools)
        return slave_pool.connection()

class UserService:
    def __init__(self, db_router: DBRouter):
        self.db_router = db_router
    
    def get_user(self, user_id: int):
        # 读操作从从库获取
        with self.db_router.get_connection(for_write=False) as conn:
            with conn.cursor() as cursor:
                sql = "SELECT * FROM users WHERE id = %s"
                cursor.execute(sql, (user_id,))
                return cursor.fetchone()
    
    def create_user(self, user: User):
        # 写操作使用主库
        with self.db_router.get_connection(for_write=True) as conn:
            with conn.cursor() as cursor:
                sql = """INSERT INTO users (username, password, age) 
                        VALUES (%s, %s, %s)"""
                cursor.execute(sql, (user.username, user.password, user.age))
                conn.commit()
                return cursor.lastrowid

4. 分库分表实现

from hashlib import md5
from typing import Tuple

class ShardingConfig:
    DB_COUNT = 2      # 数据库数量
    TABLE_COUNT = 4   # 每个库中的表数量

class ShardingRouter:
    @staticmethod
    def get_db_table(user_id: int) -> Tuple[int, int]:
        """获取分库分表位置"""
        # 使用用户ID做hash
        hash_val = int(md5(str(user_id).encode()).hexdigest(), 16)
        db_index = hash_val % ShardingConfig.DB_COUNT
        table_index = (hash_val // ShardingConfig.DB_COUNT) % ShardingConfig.TABLE_COUNT
        return db_index, table_index

    def get_connection(self, db_index: int):
        """获取指定分库的连接"""
        # 这里简化处理，实际应该维护多个连接池
        config = {
            'host': f'mysql{db_index}.example.com',
            'port': 3306,
            'user': 'root',
            'password': '123456',
            'database': f'test_{db_index}'
        }
        return pymysql.connect(**config)

class ShardingUserRepository:
    def __init__(self):
        self.router = ShardingRouter()
    
    def get_user(self, user_id: int) -> Optional[Dict]:
        db_index, table_index = self.router.get_db_table(user_id)
        with self.router.get_connection(db_index) as conn:
            with conn.cursor() as cursor:
                sql = f"SELECT * FROM users_{table_index} WHERE id = %s"
                cursor.execute(sql, (user_id,))
                return cursor.fetchone()
    
    def create_user(self, user: User) -> int:
        # 这里使用用户名作为分片键
        hash_val = int(md5(user.username.encode()).hexdigest(), 16)
        db_index = hash_val % ShardingConfig.DB_COUNT
        table_index = (hash_val // ShardingConfig.DB_COUNT) % ShardingConfig.TABLE_COUNT
        
        with self.router.get_connection(db_index) as conn:
            with conn.cursor() as cursor:
                sql = f"""INSERT INTO users_{table_index} 
                         (username, password, age) VALUES (%s, %s, %s)"""
                cursor.execute(sql, (user.username, user.password, user.age))
                conn.commit()
                return cursor.lastrowid

5. 主从复制配置

5.1 主库配置 (my.cnf)

[mysqld]
server-id = 1
log-bin = mysql-bin
binlog_format = ROW
sync_binlog = 1

5.2 从库配置 (my.cnf)

[mysqld]
server-id = 2
relay-log = slave-relay-bin
read_only = 1

5.3 主从复制设置

在主库执行：

-- 创建复制用户
CREATE USER 'repl'@'%' IDENTIFIED BY 'password';
GRANT REPLICATION SLAVE ON *.* TO 'repl'@'%';
FLUSH PRIVILEGES;

-- 获取主库状态
SHOW MASTER STATUS;

在从库执行：

CHANGE MASTER TO
    MASTER_HOST='master_host',
    MASTER_USER='repl',
    MASTER_PASSWORD='password',
    MASTER_LOG_FILE='mysql-bin.000001',
    MASTER_LOG_POS=123;

-- 启动从库复制
START SLAVE;

-- 查看从库状态
SHOW SLAVE STATUS\G

5.4 Python 监控主从状态

class ReplicationMonitor:
    def __init__(self, master_pool, slave_pool):
        self.master_pool = master_pool
        self.slave_pool = slave_pool
    
    def check_replication_status(self) -> Dict:
        master_status = self._get_master_status()
        slave_status = self._get_slave_status()
        
        return {
            'master': master_status,
            'slave': slave_status,
            'delay': self._calculate_delay(master_status, slave_status)
        }
    
    def _get_master_status(self) -> Dict:
        with self.master_pool.get_conn() as conn:
            with conn.cursor() as cursor:
                cursor.execute("SHOW MASTER STATUS")
                return cursor.fetchone()
    
    def _get_slave_status(self) -> Dict:
        with self.slave_pool.get_conn() as conn:
            with conn.cursor() as cursor:
                cursor.execute("SHOW SLAVE STATUS")
                return cursor.fetchone()
    
    def _calculate_delay(self, master_status: Dict, slave_status: Dict) -> int:
        # 计算主从延迟
        if not master_status or not slave_status:
            return -1
        return slave_status.get('Seconds_Behind_Master', -1)

# 使用示例
monitor = ReplicationMonitor(master_pool, slave_pool)
status = monitor.check_replication_status()
print(f"主从延迟: {status['delay']} 秒")