Python模块变量定义最优方案？FastAPI+SQLAlchemy数据库场景问询

1. 模块级变量（当前实现） #

实现代码 #

db.py:

from sqlalchemy import create_engine, MetaData

def get_conn():
    engine = create_engine("your-db-url")
    return engine.connect()

def get_metadata(conn):
    meta = MetaData()
    meta.reflect(bind=conn)
    return meta

conn = get_conn()
meta = get_metadata(conn)

other_file.py:

from db import conn, meta

# 使用conn和meta操作数据库

优点 #

• 实现简单，无需额外依赖或复杂逻辑
• 模块加载时仅初始化一次，全局复用同一个对象
• 调用方直接导入变量，代码简洁

缺点 #

• 无法动态修改配置：如果需要切换数据库（比如测试/生产环境），必须重启应用
• 缺乏生命周期管理：连接不会自动关闭，除非手动调用conn.close()，可能导致连接泄漏
• 测试不友好：单元测试时难以替换为测试数据库连接，只能修改模块变量或重新加载模块

2. 函数缓存（functools.lru_cache） #

实现代码 #

db.py:

from sqlalchemy import create_engine, MetaData
from functools import lru_cache

@lru_cache(maxsize=None)
def get_conn():
    engine = create_engine("your-db-url")
    return engine.connect()

@lru_cache(maxsize=None)
def get_metadata():
    conn = get_conn()
    meta = MetaData()
    meta.reflect(bind=conn)
    return meta

other_file.py:

from db import get_conn, get_metadata

conn = get_conn()
meta = get_metadata()

优点 #

• 保留函数调用的灵活性，同时确保返回同一个对象
• 相比模块变量，支持在运行时清空缓存（get_conn.cache_clear()），方便测试或切换配置
• 无需全局变量，符合函数式编程风格

缺点 #

• 缓存的对象是全局的，依然存在生命周期管理问题，连接不会自动释放
• 如果函数依赖参数（比如动态传入数据库URL），需要确保参数可哈希，否则无法缓存
• 调用方每次需要调用函数，相比直接导入变量多了一步操作

3. 单例类 #

实现代码 #

db.py:

from sqlalchemy import create_engine, MetaData

class DBManager:
    _instance = None
    conn = None
    meta = None

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            # 初始化连接和元数据
            cls._instance.conn = create_engine("your-db-url").connect()
            cls._instance.meta = MetaData()
            cls._instance.meta.reflect(bind=cls._instance.conn)
        return cls._instance

    def close(self):
        if self.conn:
            self.conn.close()

other_file.py:

from db import DBManager

db = DBManager()
conn = db.conn
meta = db.meta

优点 #

• 封装了数据库资源的生命周期管理，可提供close()方法手动释放连接
• 单例模式确保全局只有一个实例，避免重复初始化
• 可扩展其他数据库操作方法（比如执行查询、事务管理），代码结构更清晰

缺点 #

• 实现比模块变量和缓存函数复杂，需要维护单例逻辑
• 依然存在全局状态，测试时替换实例需要额外处理（比如修改_instance属性）
• 如果应用多线程/多进程，需要考虑线程安全（SQLAlchemy连接本身是线程不安全的，需额外处理连接池）

4. FastAPI依赖注入+单例模式 #

实现代码 #

db.py:

from sqlalchemy import create_engine, MetaData
from fastapi import Depends

engine = create_engine("your-db-url")
meta = MetaData()
meta.reflect(bind=engine)

def get_db_conn():
    conn = engine.connect()
    try:
        yield conn
    finally:
        conn.close()

def get_db_meta():
    return meta

main.py:

from fastapi import FastAPI, Depends
from sqlalchemy import text
from db import get_db_conn, get_db_meta

app = FastAPI()

@app.get("/test")
def test_db(conn = Depends(get_db_conn), meta = Depends(get_db_meta)):
    result = conn.execute(text("SELECT 1"))
    return {"result": result.scalar(), "tables": list(meta.tables.keys())}

优点 #

• 符合FastAPI的设计理念，依赖注入自动管理连接生命周期（请求结束自动关闭）
• 元数据通过依赖注入返回单例，避免重复反射
• 测试友好：可以轻松替换依赖为测试数据库的连接和元数据
• 支持连接池（SQLAlchemy引擎默认自带连接池），适合高并发场景

缺点 #

• 仅适用于FastAPI请求上下文，非请求场景（比如脚本、定时任务）需要手动管理连接
• 依赖FastAPI的依赖注入系统，代码耦合度较高
• 元数据初始化在模块加载时完成，若需要动态修改数据库URL，需调整初始化逻辑

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