Python多模块与函数间数据库连接的正确处理方式问询

嘿，我完全懂你的困惑——数据库连接这事儿，对非专业开发者来说，光是术语就够头疼的，更别说还要兼顾性能和正确性。咱们先拆解你当前的问题，再一步步给你捋清楚正确的做法。

先说说你现有代码的问题 #

你现在的Database类，在__init__里直接就创建了connection——这意味着每次实例化Database，都会新建一个数据库连接。比如PlayerData里建一个，save_calculations里又建一个，要是你的stats模块有一堆函数都这么干，很快就会出现大量闲置连接，既拖慢程序，还可能把数据库的连接配额占满，导致新请求连不上。

至于你之前直接全局导入连接的做法，确实有隐患：如果你的程序是单线程的，可能暂时没问题，但一旦涉及多线程、多进程，或者程序运行时间久了，很容易出现连接失效、并发操作冲突的问题，而且全局连接的生命周期不好管控，比如程序退出时可能没正确关闭，造成连接泄漏。

正确的思路：利用SQLAlchemy的连接池 #

其实你已经在用SQLAlchemy的create_engine了——它本身就自带连接池功能！连接池会自动帮你维护一批空闲连接，需要的时候取出来用，用完再放回池里，不用每次都新建连接，这才是高效又安全的做法。

咱们来重构你的代码，一步步优化：

第一步：重构db.py，让连接池帮你干活 #

把engine和会话管理器全局初始化（因为engine本身就是连接池的管理者，全局单例就够了），然后用上下文管理器来安全获取/释放连接：

from sqlalchemy import create_engine, MetaData
from sqlalchemy.engine import URL
from sqlalchemy.orm import sessionmaker

# 全局初始化engine，它会自动维护连接池
db_config = {
    'drivername': 'mysql+pymysql',
    'host': DB_HOST,
    'username': DB_USER,
    'password': DB_PASS,
    'port': DB_PORT
}
engine = create_engine(URL(**db_config), echo=False, strategy='threadlocal')
# 会话管理器，用于ORM操作（如果用原生SQL也可以用engine的连接）
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
metadata = MetaData(bind=engine)

# 用上下文管理器封装会话（推荐用ORM的话用这个）
class DatabaseSession:
    def __enter__(self):
        self.session = SessionLocal()
        return self.session
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        # 如果有异常，自动回滚；正常则提交（也可以手动在业务代码里提交）
        if exc_type is None:
            self.session.commit()
        else:
            self.session.rollback()
        self.session.close()

# 如果习惯用原生SQL，也可以用这个连接生成器
def get_db_connection():
    with engine.connect() as conn:
        yield conn

第二步：修改PlayerData类，用上下文管理器获取连接 #

现在不需要在__init__里实例化Database了，而是在需要查询的时候，临时从连接池拿连接：

class PlayerData():
    def load_player_data(self, player_id=None):
        # 用ORM会话的方式（推荐）
        with DatabaseSession() as session:
            # 如果是ORM查询：result = session.query(PlayerModel).filter_by(id=player_id).all()
            # 如果是原生SQL：
            result = session.execute(s)
            # 处理结果...
        
        # 或者用原生SQL的连接方式：
        # for conn in get_db_connection():
        #     result = conn.execute(s)
        #     # 处理结果...

第三步：修改stats.py的保存函数，同样用上下文管理器 #

def save_calculations(data):
    with DatabaseSession() as session:
        session.execute('insert into ... values (...)', data)
        # 这里如果在DatabaseSession的__exit__里自动提交了，就不用手动写commit了
        # 当然你也可以手动在这提交，更灵活

为什么这样做更好？ #

• 连接复用：连接池会自动维护空闲连接，每次用with获取的都是池里的闲置连接，不会每次新建，大大降低连接开销。
• 安全可靠：上下文管理器确保连接/会话被正确关闭、归还到池里，不会出现连接泄漏。
• 线程安全：SQLAlchemy的连接池和会话管理器已经帮你处理了线程安全问题，不用担心多线程下的冲突。

额外的小建议 #

• 如果你的业务代码用ORM更多，优先用Session的方式，SQLAlchemy会帮你处理事务、缓存等细节。
• 可以调整连接池参数，比如pool_size（默认5，设置成适合你并发量的数值）、max_overflow（临时超出池大小的连接数），比如：

  engine = create_engine(URL(**db_config), echo=False, strategy='threadlocal', pool_size=10, max_overflow=20)
  

• 调试的时候可以打开echo=True，看看SQLAlchemy的日志，能清楚看到连接的获取、使用、归还过程，帮你理解连接池的工作原理。

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