Python模块全局变量my_connection初始化后值丢失，引发属性错误

在Python中尝试创建全局变量my_connection，在mysql_example模块中定义并在MysqlExample类的__init__方法中初始化，但后续调用类方法时该变量始终为None，引发错误：AttributeError: 'NoneType' object has no attribute 'cursor'。

模块代码：

import datetime
import mysql.connector as mysql

my_connection = None


class MysqlExample(object):

    def __init__(self, config : dict ):
        my_connection = mysql.Connect( **config )
        my_connection.autocommit = True
    
    def close(self):
        self._endExtraction()
        my_connection.close()
    
    def __enter__(self):
        self._id_extraction = self._startExtraction()
        return self
        
    def __exit__(self, type, value, tb ):
        self.close()
    
    def _startExtraction(self):
        cur = my_connection.cursor()
        cur.execute( '''select * from simple_test''' )
        return cur.lastrowid
        
    def _endExtraction(self):
        self._lock.acquire()
        cur = my_connection.cursor()
        cur.execute('''select * from simple_test''', 
            { 'id_extraction' : self._id_extraction, 
              'end_date' : datetime.datetime.now() } )
        self._lock.release()

主调用代码：

with MysqlExample({ "host" : "XXXX", 
                              "database" : "XXXXX",
                              "user" : "XXXXX", 
                              "password" : "super-secret-password" }) as my_example: 
    print("hello")

原因分析 #

这是Python的常规行为。当你在函数（这里是__init__方法）内部对变量赋值时，Python会默认将该变量视为局部变量，而非全局变量。所以你在__init__里的my_connection = mysql.Connect(**config)并没有修改模块级的全局my_connection，而是创建了一个仅在__init__方法内有效的局部变量，方法执行完毕后这个局部变量就被销毁了，模块级的my_connection仍然是初始的None。

修正方案 #

方案1：使用global声明修改全局变量

在__init__方法开头添加global关键字，明确告诉Python你要操作的是全局变量：

def __init__(self, config : dict ):
    global my_connection
    my_connection = mysql.Connect( **config )
    my_connection.autocommit = True

但这种方法不推荐，因为全局变量会导致多个MysqlExample实例共享同一个连接，容易引发并发问题，也不符合面向对象的设计思想。

方案2：将连接作为类的实例属性（推荐）

把数据库连接存储为实例的属性，每个实例拥有独立的连接，更安全且符合类的封装特性：

import datetime
import mysql.connector as mysql
from threading import Lock  # 导入Lock类

class MysqlExample(object):

    def __init__(self, config : dict ):
        self.my_connection = mysql.Connect( **config )
        self.my_connection.autocommit = True
        self._lock = Lock()  # 初始化锁
    
    def close(self):
        self._endExtraction()
        self.my_connection.close()
    
    def __enter__(self):
        self._id_extraction = self._startExtraction()
        return self
        
    def __exit__(self, type, value, tb ):
        self.close()
    
    def _startExtraction(self):
        cur = self.my_connection.cursor()
        cur.execute( '''select * from simple_test''' )
        return cur.lastrowid
        
    def _endExtraction(self):
        self._lock.acquire()
        try:  # 添加try-finally确保锁被释放
            cur = self.my_connection.cursor()
            # 修正原SQL语句参数不匹配问题，改为合理的更新操作示例
            cur.execute('''UPDATE simple_test SET end_date = %s WHERE id_extraction = %s''', 
                (datetime.datetime.now(), self._id_extraction) )
        finally:
            self._lock.release()

同时修正了原代码中_lock未初始化的问题，以及_endExtraction中SQL语句参数与占位符不匹配的问题。

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