在Python中没有ORM的http服务器（Flask）中处理sql连接的最佳方法是什么？

Question

我正在将 Flask 与 MySQL (MariaDB) 数据库一起使用。为了处理 sql 连接和游标，我使用自制的上下文管理器。我在每个 Flask http 请求 hadling 函数中打开和关闭连接，所以我可以确定到 db 的连接数不会超过特定数量，但它会产生开销。我确信其他用户可以使用相同的 mysql 连接，如果我不使用 ORM，我可以使用什么其他方法来处理 sql 连接和游标？用于挂起光标和连接的上下文管理器：from contextlib import contextmanagerimport mysql.connectorfrom mysql.connector.errors import Error@contextmanagerdef mysql_connection(user, password, host, database, auth_plugin):    _conn = mysql.connector.connect(user=user, password=password, host=host, database=database, auth_plugin=auth_plugin)    try:        yield _conn    except (Exception, Error) as ex:        # if error happened all made changes during the connection will be rolled back:        _conn.rollback()        # this statement re-raise error to let it be handled in outer scope:        raise    else:        # if everything is fine commit all changes to save them in db:        _conn.commit()    finally:        # close connection to db, do not wait for timeout release:        _conn.close()@contextmanagerdef mysql_curs(user, password, host, database, auth_plugin) -> "curs":    with mysql_connection(user=user, password=password, host=host, database=database, auth_plugin=auth_plugin) as _conn:        _curs = _conn.cursor()        try:            yield _curs        finally:            _curs.close()  # close cursor when everything is done编辑：我想补充一点，我发现以下文章讨论了如何将 Flask 与 PostgreSQL 一起使用并创建自定义的 sql 连接上下文管理器，但我有疑问在 Flask 中我应该在哪里声明 sql 连接器池：在 Flask 中管理 RAW 数据库连接池

Answer 1

尝试池连接

来自官方文档：

在向请求者提供连接时，池会打开多个连接并处理线程安全

实现连接池，可以复用已有的连接

dbconfig = {

  "database": "test",

  "user":     "joe"

}

cnxpool = mysql.connector.connect(pool_name = "mypool",

                                  pool_size = 3,    # or any number to suit your need

                                  **dbconfig)

# then to get a connection from pool use

cnx = cnxpool.get_connection()

有关更多信息，请参阅：https ://dev.mysql.com/doc/connector-python/en/connector-python-connection-pooling.html

Answer 2

如果有人对在没有 ORM 的情况下处理 sql 连接的方法感兴趣，我做了以下步骤来结合 MySQL Connections Pool、上下文管理器和 Flask：

SQL_CONN_POOL = pooling.MySQLConnectionPool(

    pool_name="mysqlpool",

    pool_size=10,

    user=DB_USER,

    password=DB_PASS,

    host=DB_HOST,

    database=DATABASE,

    auth_plugin=DB_PLUGIN

)

@contextmanager

def mysql_connection_from_pool() -> "conn":

    conn_pool = SQL_CONN_POOL  # get connection from the pool, all the rest is the same

    # you can add print(conn_pool) here to be sure that pool

    # is the same for each http request

    _conn = conn_pool.get_connection()

    try:

        yield _conn

    except (Exception, Error) as ex:

        # if error happened all made changes during the connection will be rolled back:

        _conn.rollback()

        # this statement re-raise error to let it be handled in outer scope:

        raise

    else:

        # if everything is fine commit all changes to save them in db:

        _conn.commit()

    finally:

        # actually it returns cursor to the pool, rather than close it

        _conn.close()

@contextmanager

def mysql_curs_from_pool() -> "curs":

    with mysql_connection_from_pool() as _conn:

        _curs = _conn.cursor()

        try:

            yield _curs

        finally:

            _curs.close()