重试失败的 sqlalchemy 查询
Retry failed sqlalchemy queries
每次我重新启动 mysql 服务时，我的应用在任何查询中都会收到以下错误：
result = self._query(query)
File "/usr/local/lib/python3.6/site-packages/pymysql/cursors.py", line 328, in _query
conn.query(q)
File "/usr/local/lib/python3.6/site-packages/pymysql/connections.py", line 516, in query
self._affected_rows = self._read_query_result(unbuffered=unbuffered)
File "/usr/local/lib/python3.6/site-packages/pymysql/connections.py", line 727, in _read_query_result
result.read()
File "/usr/local/lib/python3.6/site-packages/pymysql/connections.py", line 1066, in read
first_packet = self.connection._read_packet()
File "/usr/local/lib/python3.6/site-packages/pymysql/connections.py", line 656, in _read_packet
packet_header = self._read_bytes(4)
File "/usr/local/lib/python3.6/site-packages/pymysql/connections.py", line 702, in _read_bytes
CR.CR_SERVER_LOST, "Lost connection to MySQL server during query")
sqlalchemy.exc.OperationalError: (pymysql.err.OperationalError) (2013, 'Lost connection to MySQL server during query') [SQL: ...] [parameters: {...}] (Background on this error at: http://sqlalche.me/e/e3q8)
之后的任何查询都会像往常一样成功。
例如，这只是一个常见的用例，通常我可能想根据错误重试任何查询。
有什么方法可以在某些低级别 sqlalchemy api 中捕获并重试查询？在我的代码中执行 try-except 或自定义 query 方法是不合理的，因为我使用它的次数太多而且无法维护。
显然 sqlalchemy 有一个很好的选项来自定义查询 class，这正是我所需要的。
class 实施：
import logging
from flask_sqlalchemy import BaseQuery
from sqlalchemy.exc import OperationalError
from time import sleep
class RetryingQuery(BaseQuery):
__retry_count__ = 3
__retry_sleep_interval_sec__ = 0.5
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
def __iter__(self):
attempts = 0
while True:
attempts += 1
try:
return super().__iter__()
except OperationalError as ex:
if "Lost connection to MySQL server during query" not in str(ex):
raise
if attempts < self.__retry_count__:
logging.debug(
"MySQL connection lost - sleeping for %.2f sec and will retry (attempt #%d)",
self.__retry_sleep_interval_sec__, attempts
)
sleep(self.__retry_sleep_interval_sec__)
continue
else:
raise
用法：
class BaseModel(Model):
...
query_class = RetryingQuery
...
db = SQLAlchemy(model_class=BaseModel, query_class=RetryingQuery)
我不得不稍微调整它以使其与 Postgres 一起工作，它有不同的错误消息。我知道这个问题被标记为 mysql，但通过搜索发现了这个问题（并且遇到了完全相同的问题）所以可能会对某人有所帮助。
我还必须赶上 StatementError: (sqlalchemy.exc.InvalidRequestError) Can't reconnect until invalid transaction is rolled back，它在重试发生之前炸毁了 Flask。
最后我做了指数退避，因为为什么不
import logging
from flask_sqlalchemy import BaseQuery
from sqlalchemy.exc import OperationalError, StatementError
from time import sleep
class RetryingQuery(BaseQuery):
__retry_count__ = 3
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
def __iter__(self):
attempts = 0
while True:
attempts += 1
try:
return super().__iter__()
except OperationalError as ex:
if "server closed the connection unexpectedly" not in str(ex):
raise
if attempts < self.__retry_count__:
sleep_for = 2 ** (attempts - 1)
logging.error(
"Database connection error: {} - sleeping for {}s"
" and will retry (attempt #{} of {})".format(
ex, sleep_for, attempts, self.__retry_count__
)
)
sleep(sleep_for)
continue
else:
raise
except StatementError as ex:
if "reconnect until invalid transaction is rolled back" not in str(ex):
raise
self.session.rollback()
非常感谢这个片段，我不得不对其进行一些调整以直接与 sqlalchemy.orm 一起使用：如果它对任何人都有用..
from sqlalchemy.exc import OperationalError, StatementError
from sqlalchemy.orm.query import Query as _Query
from time import sleep
class RetryingQuery(_Query):
__max_retry_count__ = 3
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
def __iter__(self):
attempts = 0
while True:
attempts += 1
try:
return super().__iter__()
except OperationalError as ex:
if "server closed the connection unexpectedly" not in str(ex):
raise
if attempts <= self.__max_retry_count__:
sleep_for = 2 ** (attempts - 1)
logging.error(
"/!\ Database connection error: retrying Strategy => sleeping for {}s"
" and will retry (attempt #{} of {}) \n Detailed query impacted: {}".format(
sleep_for, attempts, self.__max_retry_count__, ex)
)
sleep(sleep_for)
continue
else:
raise
except StatementError as ex:
if "reconnect until invalid transaction is rolled back" not in str(ex):
raise
self.session.rollback()
以及用法：将选项传递给 sessionmaker：
sqlalchemy.orm.sessionmaker(bind=engine, query_cls=RetryingQuery)
SQLAlchemy 还允许您监听在创建 connection 之前触发的 engine_connect 事件。这使得为悲观断开处理实现自定义逻辑成为可能
下面的代码片段实现了重试的指数退避。它取自 Apache Airflow 的 SQLAlchemy Utils：
http://airflow.apache.org/docs/1.10.3/_modules/airflow/utils/sqlalchemy.html
@event.listens_for(engine, "engine_connect")
def ping_connection(connection, branch):
"""
Pessimistic SQLAlchemy disconnect handling. Ensures that each
connection returned from the pool is properly connected to the database.
http://docs.sqlalchemy.org/en/rel_1_1/core/pooling.html#disconnect-handling-pessimistic
"""
if branch:
# "branch" refers to a sub-connection of a connection,
# we don't want to bother pinging on these.
return
start = time.time()
backoff = initial_backoff_seconds
# turn off "close with result". This flag is only used with
# "connectionless" execution, otherwise will be False in any case
save_should_close_with_result = connection.should_close_with_result
while True:
connection.should_close_with_result = False
try:
connection.scalar(select([1]))
# If we made it here then the connection appears to be healthy
break
except exc.DBAPIError as err:
if time.time() - start >= reconnect_timeout_seconds:
log.error(
"Failed to re-establish DB connection within %s secs: %s",
reconnect_timeout_seconds,
err)
raise
if err.connection_invalidated:
log.warning("DB connection invalidated. Reconnecting...")
# Use a truncated binary exponential backoff. Also includes
# a jitter to prevent the thundering herd problem of
# simultaneous client reconnects
backoff += backoff * random.random()
time.sleep(min(backoff, max_backoff_seconds))
# run the same SELECT again - the connection will re-validate
# itself and establish a new connection. The disconnect detection
# here also causes the whole connection pool to be invalidated
# so that all stale connections are discarded.
continue
else:
log.error(
"Unknown database connection error. Not retrying: %s",
err)
raise
finally:
# restore "close with result"
connection.should_close_with_result = save_should_close_with_result
每次我重新启动 mysql 服务时，我的应用在任何查询中都会收到以下错误：
result = self._query(query)
File "/usr/local/lib/python3.6/site-packages/pymysql/cursors.py", line 328, in _query
conn.query(q)
File "/usr/local/lib/python3.6/site-packages/pymysql/connections.py", line 516, in query
self._affected_rows = self._read_query_result(unbuffered=unbuffered)
File "/usr/local/lib/python3.6/site-packages/pymysql/connections.py", line 727, in _read_query_result
result.read()
File "/usr/local/lib/python3.6/site-packages/pymysql/connections.py", line 1066, in read
first_packet = self.connection._read_packet()
File "/usr/local/lib/python3.6/site-packages/pymysql/connections.py", line 656, in _read_packet
packet_header = self._read_bytes(4)
File "/usr/local/lib/python3.6/site-packages/pymysql/connections.py", line 702, in _read_bytes
CR.CR_SERVER_LOST, "Lost connection to MySQL server during query")
sqlalchemy.exc.OperationalError: (pymysql.err.OperationalError) (2013, 'Lost connection to MySQL server during query') [SQL: ...] [parameters: {...}] (Background on this error at: http://sqlalche.me/e/e3q8)
之后的任何查询都会像往常一样成功。
例如，这只是一个常见的用例，通常我可能想根据错误重试任何查询。
有什么方法可以在某些低级别 sqlalchemy api 中捕获并重试查询？在我的代码中执行 try-except 或自定义 query 方法是不合理的，因为我使用它的次数太多而且无法维护。
显然 sqlalchemy 有一个很好的选项来自定义查询 class，这正是我所需要的。
class 实施：
import logging
from flask_sqlalchemy import BaseQuery
from sqlalchemy.exc import OperationalError
from time import sleep
class RetryingQuery(BaseQuery):
__retry_count__ = 3
__retry_sleep_interval_sec__ = 0.5
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
def __iter__(self):
attempts = 0
while True:
attempts += 1
try:
return super().__iter__()
except OperationalError as ex:
if "Lost connection to MySQL server during query" not in str(ex):
raise
if attempts < self.__retry_count__:
logging.debug(
"MySQL connection lost - sleeping for %.2f sec and will retry (attempt #%d)",
self.__retry_sleep_interval_sec__, attempts
)
sleep(self.__retry_sleep_interval_sec__)
continue
else:
raise
用法：
class BaseModel(Model):
...
query_class = RetryingQuery
...
db = SQLAlchemy(model_class=BaseModel, query_class=RetryingQuery)
我不得不稍微调整它以使其与 Postgres 一起工作，它有不同的错误消息。我知道这个问题被标记为 mysql，但通过搜索发现了这个问题（并且遇到了完全相同的问题）所以可能会对某人有所帮助。
我还必须赶上 StatementError: (sqlalchemy.exc.InvalidRequestError) Can't reconnect until invalid transaction is rolled back，它在重试发生之前炸毁了 Flask。
最后我做了指数退避，因为为什么不
import logging
from flask_sqlalchemy import BaseQuery
from sqlalchemy.exc import OperationalError, StatementError
from time import sleep
class RetryingQuery(BaseQuery):
__retry_count__ = 3
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
def __iter__(self):
attempts = 0
while True:
attempts += 1
try:
return super().__iter__()
except OperationalError as ex:
if "server closed the connection unexpectedly" not in str(ex):
raise
if attempts < self.__retry_count__:
sleep_for = 2 ** (attempts - 1)
logging.error(
"Database connection error: {} - sleeping for {}s"
" and will retry (attempt #{} of {})".format(
ex, sleep_for, attempts, self.__retry_count__
)
)
sleep(sleep_for)
continue
else:
raise
except StatementError as ex:
if "reconnect until invalid transaction is rolled back" not in str(ex):
raise
self.session.rollback()
非常感谢这个片段，我不得不对其进行一些调整以直接与 sqlalchemy.orm 一起使用：如果它对任何人都有用..
from sqlalchemy.exc import OperationalError, StatementError
from sqlalchemy.orm.query import Query as _Query
from time import sleep
class RetryingQuery(_Query):
__max_retry_count__ = 3
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
def __iter__(self):
attempts = 0
while True:
attempts += 1
try:
return super().__iter__()
except OperationalError as ex:
if "server closed the connection unexpectedly" not in str(ex):
raise
if attempts <= self.__max_retry_count__:
sleep_for = 2 ** (attempts - 1)
logging.error(
"/!\ Database connection error: retrying Strategy => sleeping for {}s"
" and will retry (attempt #{} of {}) \n Detailed query impacted: {}".format(
sleep_for, attempts, self.__max_retry_count__, ex)
)
sleep(sleep_for)
continue
else:
raise
except StatementError as ex:
if "reconnect until invalid transaction is rolled back" not in str(ex):
raise
self.session.rollback()
以及用法：将选项传递给 sessionmaker：
sqlalchemy.orm.sessionmaker(bind=engine, query_cls=RetryingQuery)
SQLAlchemy 还允许您监听在创建 connection 之前触发的 engine_connect 事件。这使得为悲观断开处理实现自定义逻辑成为可能
下面的代码片段实现了重试的指数退避。它取自 Apache Airflow 的 SQLAlchemy Utils： http://airflow.apache.org/docs/1.10.3/_modules/airflow/utils/sqlalchemy.html
@event.listens_for(engine, "engine_connect")
def ping_connection(connection, branch):
"""
Pessimistic SQLAlchemy disconnect handling. Ensures that each
connection returned from the pool is properly connected to the database.
http://docs.sqlalchemy.org/en/rel_1_1/core/pooling.html#disconnect-handling-pessimistic
"""
if branch:
# "branch" refers to a sub-connection of a connection,
# we don't want to bother pinging on these.
return
start = time.time()
backoff = initial_backoff_seconds
# turn off "close with result". This flag is only used with
# "connectionless" execution, otherwise will be False in any case
save_should_close_with_result = connection.should_close_with_result
while True:
connection.should_close_with_result = False
try:
connection.scalar(select([1]))
# If we made it here then the connection appears to be healthy
break
except exc.DBAPIError as err:
if time.time() - start >= reconnect_timeout_seconds:
log.error(
"Failed to re-establish DB connection within %s secs: %s",
reconnect_timeout_seconds,
err)
raise
if err.connection_invalidated:
log.warning("DB connection invalidated. Reconnecting...")
# Use a truncated binary exponential backoff. Also includes
# a jitter to prevent the thundering herd problem of
# simultaneous client reconnects
backoff += backoff * random.random()
time.sleep(min(backoff, max_backoff_seconds))
# run the same SELECT again - the connection will re-validate
# itself and establish a new connection. The disconnect detection
# here also causes the whole connection pool to be invalidated
# so that all stale connections are discarded.
continue
else:
log.error(
"Unknown database connection error. Not retrying: %s",
err)
raise
finally:
# restore "close with result"
connection.should_close_with_result = save_should_close_with_result