retry_async_transaction

safir.database.retry_async_transaction(__func=None, /, *, delay=0.5, max_tries=3)

Retry if a transaction failed.

If the wrapped method fails with an sqlalchemy.exc.DBAPIError exception, it is retried up to max_tries times. Any method with this decorator must be idempotent, since it may be re-run multiple times.

One common use for this decorator is when the database engine has been configured with the REPEATABLE READ transaction isolation level and multiple writers may be updating the same object at the same time. The loser of the transaction race will raise one of the above exceptions, and can be retried with this decorator.

Parameters:

  • delay (float, default: 0.5) – How long, in seconds, to wait between tries.

  • max_tries (int, default: 3) – Number of times to retry the transaction. Practical experience with REPEATABLE READ isolation suggests a count of at least three.

Return type:

Callable[[ParamSpec(RetryP, bound= None)], Coroutine[None, None, TypeVar(RetryT)]] | Callable[[Callable[[ParamSpec(RetryP, bound= None)], Coroutine[None, None, TypeVar(RetryT)]]], Callable[[ParamSpec(RetryP, bound= None)], Coroutine[None, None, TypeVar(RetryT)]]]

Examples

This decorator can be used with any idempotent Python function or method that makes database calls and should be retried on the above-listed exceptions.

from safir.database import retry_async_transaction
from sqlalchemy.ext.asyncio import async_scoped_session

@retry_async_transaction(max_tries=5)
def make_some_database_call(session: async_scoped_session) -> None: ...

If the default value of max_tries is acceptable, this decorator can be used without arguments.

from safir.database import retry_async_transaction
from sqlalchemy.ext.asyncio import async_scoped_session


@retry_async_transaction
def make_some_database_call(session: async_scoped_session) -> None: ...
Parameters:

__func (Callable[[ParamSpec(RetryP, bound= None)], Coroutine[None, None, TypeVar(RetryT)]] | None, default: None)