python - sqlalchemy 数据库连接无法关闭？

Question

刚才发现每次调用完接口都有一堆连接没关掉（ 如下图 ）：

于是写了一段本地代码无限循环跑发现单独执行一次是可以立刻关闭连接的即便循环跑也一直是 10-30 来回跳我估计就是每次循环好就关闭了问题不大（ 代码如下 ）：

from sqlalchemy import create_engine, text, select
from typing import List
from typing import Optional
from sqlalchemy import ForeignKey
from sqlalchemy import String
from sqlalchemy.orm import DeclarativeBase, Session
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship
from sqlalchemy.pool import NullPool


class Base(DeclarativeBase):
    pass


class User(Base):
     __tablename__ = "test"

     id: Mapped[int] = mapped_column(primary_key=True)
     app_name: Mapped[str] = mapped_column(String(30))

while True:
    # 创建数据库引擎
    engine = create_engine('mysql+pymysql://root:admin4ph@192.168.3.232:33306/test', echo=True, poolclass=NullPool, isolation_level="AUTOCOMMIT")
    session = Session(engine)
    stmt = select(User)
    for user in session.scalars(stmt):
        print(user.app_name)

但是我的 flask 应用中每次请求完连接都在一直没搞懂什么问题（ 代码如下 ）：

from flask import current_app
from lib.database import Database
from core.database.base import Base
from lib.type import Type
from flask import g


class Platform(Database.Easy):

    @property
    def database(self):
        """
        @ 数据库[缓存]
        """
        return Database.container(**self.database_conf)["session"]

    @property
    def database_conf(self):
        """
        @ [平台-数据库]：配置
        """
        return Base.setting(current_app.config["Database"])

from sqlalchemy import create_engine, URL, delete, update, select, exists, inspect, MetaData, Table, func, text
from sqlalchemy.orm import sessionmaker, scoped_session, Session
from sqlalchemy.dialects import mysql
from core.database.base import Base
from lib.type import Type
from typing import Any
from sqlalchemy.pool import NullPool

import importlib
import re


class Database:

    @classmethod
    def container(cls, **config):
        """
        @ 创建容器：节约开销
        """
        engine = cls.create_engine(**config)
        session = Session(engine)
        # return Type.Database(engine=engine, session=session)
        return {"engine": engine, "session": session}

    @classmethod
    def create_engine(cls, **kwargs):
        """
        @ 创建连接：引擎
        """
        return create_engine(URL.create("mysql+pymysql", **kwargs), poolclass=NullPool, isolation_level="AUTOCOMMIT")

class Easy:


    def table_data_query_one(self, model: Any, condition: list = None,design:bool=False) -> dict:
        """
        @ 查询：单条

        :param model: 模型
        :param condition: 条件
        """
        condition.append(model.status >= -1 if design else model.status == 1)
        result = self.database.execute(select(model).filter(*condition).limit(1)).scalar_one_or_none()
        return None if result is None else result.dict()

请求指点分析什么问题

Answer 1

在 Flask 应用中，你遇到的数据库连接未关闭的问题很可能是由于你的数据库会话（session）管理不当。在你的 Database 类中，你为每个请求创建了一个新的引擎和会话，但是它们并没有被适当地关闭或回收。

在 Flask 应用中，通常的做法是使用应用上下文（application context）或请求上下文（request context）来管理数据库会话的生命周期。这样，你可以确保在每个请求开始时创建一个新的会话，并在请求结束时关闭它。

以下是一些建议的改进方法：

1. 使用 Flask-SQLAlchemy 或类似的扩展：这些扩展已经为你处理了会话管理的大部分细节。
2. 使用 scoped_session：scoped_session 可以确保在请求期间只有一个会话实例，并在请求结束时自动关闭。
3. 在请求开始时创建会话，在请求结束时关闭：你可以使用 Flask 的 @app.teardown_appcontext 装饰器来确保在每个请求结束时关闭会话。

以下是一个使用 scoped_session 和 Flask 的上下文管理器的示例：

from flask import Flask, g
from sqlalchemy.orm import scoped_session, sessionmaker
from sqlalchemy import create_engine
from sqlalchemy.pool import NullPool

app = Flask(__name__)

# 创建引擎（通常只需要一次）
engine = create_engine('mysql+pymysql://root:admin4ph@192.168.3.232:33306/test', poolclass=NullPool, isolation_level="AUTOCOMMIT")

# 创建 scoped_session
Session = scoped_session(sessionmaker(bind=engine))

@app.before_request
def before_request():
    # 在请求开始时，将 session 绑定到 g
    g.db = Session()

@app.teardown_appcontext
def teardown_appcontext(exception):
    # 在请求结束时，关闭 session
    if hasattr(g, 'db'):
        g.db.close()

# 在你的视图函数或模型中，使用 g.db 来访问数据库
# 例如：
# result = g.db.query(User).all()

注意：在上面的示例中，我使用了 NullPool，这意味着每个数据库连接在使用后都会被立即关闭。这可能在某些情况下是有效的，但它也限制了连接池的大小和可能的性能优势。根据你的应用需求，你可能想要使用一个具有连接池的引擎。

最后，确保你的 Flask 应用正确地配置了线程和进程安全，特别是如果你打算在生产环境中使用它。