Каков наилучший способ обработки соединения sql на http-сервере (Flask) без ORM в Python?

Question

Я использую Flask с базой данных MySQL (MariaDB). Для обработки sql соединения и курсора я использую самодельный контекстный менеджер. Я открываю и закрываю соединение внутри каждой функции хедлинга запроса Flask http, поэтому я могу быть уверен, что количество соединений с БД не превысит определенного числа, но это создает накладные расходы. Я уверен, что те же самые соединения mysql могут использоваться другими пользователями, какой другой подход для обработки соединения sql и курсора я могу использовать, если не использую ORM?

Менеджеры контекста чтобы повесить курсор и соединение:

from contextlib import contextmanager
import mysql.connector
from mysql.connector.errors import Error

@contextmanager
def 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()


@contextmanager
def 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 функция http-обработчика:

@app.route('/admin_panel/repair', methods=["GET"])
def repair_show_all_menu_webpages():
    """The page exists to repair menu if not existent flask function was added"""
    try:
        with common_db_ops.mysql_curs() as curs:
            left_side_menu = []
            webpages = admin_ops.show_all_menu_webpages_to_repair(curs)
    except (Exception, Error) as err:
        app.logger.error(f"Failed to repair website: {err}")
        abort(500)

    return render_template('repair_menu.html', webpages=webpages, left_side_menu=left_side_menu)

Редактировать: Я хотел бы добавим, что я нашел следующую статью, в которой обсуждается, как использовать Flask с PostgreSQL и создать ваш настроенный менеджер контекста соединения sql, но у меня есть вопрос, где в Flask я должен объявить sql Пул соединителей:

Управление пулом соединений с базами данных RAW в Flask

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

Если кого-то интересует подход обработки sql соединения без ORM, я сделал следующие шаги для объединения MySQL Пул соединений, менеджер контекста и 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()

Я использовал следующие ссылки для ответа на вопрос:

Управление пулом соединений с базами данных RAW в Flask

MySQL документы