我正在將 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 數據庫連接池
2 回答
冉冉說
TA貢獻1877條經驗 獲得超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
小唯快跑啊
TA貢獻1863條經驗 獲得超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()
添加回答
舉報
0/150
提交
取消