如何使用MariaDB Connector/Python简化Python连接与游标管理？

问题背景 #

在实现大量Python数据库交互功能时，连接与会话管理缺乏自动化方式，导致事务相关函数存在大量重复代码，主要陷入两种尴尬场景：

• 每个需要游标的函数必须接收游标参数，配合装饰器使用
• 显式声明连接（即使使用上下文管理器），仍需手动调用conn.close()和cursor.close()，冗余代码无法规避

已尝试两种方案：

1. 显式上下文管理器方案（存在冗余关闭代码）

import mariadb as db

def db_create_connection():
    return db.connect(
        user='...',
        password='...',
        host='...',
        database='...',
        port=...
    )

def db_some_fn():
    with db_create_connection() as conn:
        with conn.cursor() as cursor:
            ...
            conn.commit()
            cursor.close()
            conn.close()

2. 装饰器方案（需函数接收游标参数，担心后续误操作游标引发调试问题）

import mariadb as db
from functools import wraps

def with_db_cursor(func):
    @wraps(func)
    def wrapper(*args, **kwargs):

        conn, cursor = None, None

        try:
            conn = db.connect(
                user='...',
                password='...',
                host='...',
                database='...',
                port=...
            )

            cursor = conn.cursor()

            result = func(cursor, *args, **kwargs)
            conn.commit()
            return result

        except Exception as e:
            if conn:
                conn.rollback()
            print(f"An error occurred: {e}")
            raise

        finally:
            if cursor:
                cursor.close()
            if conn:
                conn.close()

    return wrapper

更优解决方案：自定义上下文管理器 #

可以封装一个同时管理连接和游标的上下文管理器，既无需手动关闭连接/游标，也不用让业务函数接收游标参数，同时内置事务提交/回滚逻辑，降低误操作风险。

实现代码 #

import mariadb as db

class DBContext:
    def __init__(self, user, password, host, database, port):
        self.user = user
        self.password = password
        self.host = host
        self.database = database
        self.port = port
        self.conn = None
        self.cursor = None

    def __enter__(self):
        # 建立连接并创建游标
        self.conn = db.connect(
            user=self.user,
            password=self.password,
            host=self.host,
            database=self.database,
            port=self.port
        )
        self.cursor = self.conn.cursor()
        # 返回游标供业务逻辑使用
        return self.cursor

    def __exit__(self, exc_type, exc_val, exc_tb):
        # 处理事务：有异常则回滚，否则提交
        if exc_type is None:
            self.conn.commit()
        else:
            self.conn.rollback()
            print(f"数据库操作异常: {exc_val}")
        
        # 自动关闭游标和连接
        if self.cursor:
            self.cursor.close()
        if self.conn:
            self.conn.close()
        
        # 返回False让异常继续向上抛出（如果需要）
        return False

# 使用示例
def db_some_fn():
    with DBContext(
        user='...',
        password='...',
        host='...',
        database='...',
        port=...
    ) as cursor:
        # 直接编写业务SQL逻辑
        cursor.execute("SELECT * FROM some_table")
        result = cursor.fetchall()
        # 无需手动提交、关闭游标/连接
        return result

方案优势 #

• 完全消除冗余代码：无需在业务函数中写close()和事务管理代码
• 降低误操作风险：连接和游标由上下文管理器统一管理，业务代码仅需关注SQL逻辑，不会误操作连接对象
• 事务自动处理：内置提交/回滚逻辑，异常时自动回滚，避免数据不一致
• 扩展性强：后续可在上下文管理器中添加连接池、日志记录等扩展功能

进阶优化：连接池集成 #

如果需要处理高并发场景，可以在上下文管理器中集成连接池，避免频繁创建/销毁连接的性能损耗：

import mariadb as db
from mariadb import ConnectionPool

class DBContext:
    _pool = None

    @classmethod
    def init_pool(cls, user, password, host, database, port, pool_size=5):
        # 初始化全局连接池
        cls._pool = ConnectionPool(
            user=user,
            password=password,
            host=host,
            database=database,
            port=port,
            pool_size=pool_size
        )

    def __enter__(self):
        if not self._pool:
            raise RuntimeError("请先调用DBContext.init_pool初始化连接池")
        # 从连接池获取连接
        self.conn = self._pool.get_connection()
        self.cursor = self.conn.cursor()
        return self.cursor

    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is None:
            self.conn.commit()
        else:
            self.conn.rollback()
            print(f"数据库操作异常: {exc_val}")
        
        self.cursor.close()
        # 将连接归还连接池而非关闭
        self.conn.close()
        
        return False

# 使用前初始化连接池
DBContext.init_pool(
    user='...',
    password='...',
    host='...',
    database='...',
    port=...
)

# 业务函数使用
def db_query_users():
    with DBContext() as cursor:
        cursor.execute("SELECT id, name FROM users")
        return cursor.fetchall()

内容的提问来源于stack exchange，提问作者DarkRemino