I'd like to retry only for some cases of the exception.
Example, I have to retry when SQLAlchemy fails with lock timeout or statement timeout.
I can raise custom exceptions, but it is not useful for anything else.
@retry(...)
def foo():
try:
something
except sqlalchemy.exc.OperationalError as ex:
if "lock timeout" in ex.orig.diag.message_primary:
# RETRY
if "statement timeout" in ex.orig.diag.message_primary:
# RETRY
raise
I'd like to extract this logic to make code reusable:
def should_retry(exc):
return (
isinstance(exc, sqlalchemy.exc.OperationalError)
and (
"lock timeout" in exc.orig.diag.message_primary
or "statement timeout" in exc.orig.diag.message_primary
)
)
@retry(retry_on_exceptions=should_retry)
def foo():
something
Maybe I can make it per exception:
def should_retry(exc):
return (
"lock timeout" in exc.orig.diag.message_primary
or "statement timeout" in exc.orig.diag.message_primary
)
@retry(
retry_on_exceptions=[
(sqlalchemy.exc.OperationalError, should_retry),
],
)
def foo():
something
I'd like to retry only for some cases of the exception.
Example, I have to retry when SQLAlchemy fails with lock timeout or statement timeout.
I can raise custom exceptions, but it is not useful for anything else.
I'd like to extract this logic to make code reusable:
Maybe I can make it per exception: