One case use context manager to handle exceptions during execution of the with statement as can be seen in the snippet below. This is useful for example for rolling back database transactions in case of an exception, where the database connections can be retrieved from and returned to a connection pool.
from contextlib import contextmanager
@contextmanager
def managed_resource(param):
try:
print(f"db connection acquired {param}")
yield "this is the connection"
except Exception as e:
print(f"rolling back due to ex {e}")
raise
finally:
print("returning db connection to a pool")
with managed_resource("this is param value") as connection:
print(connection)
raise ValueError("hello")
Execution of above prints below.
db connection acquired this is param value
this is the connection
rolling back due to ex hello
returning db connection to a pool
Traceback (most recent call last):
File ".../test/test.py", line 36, in <module>
raise ValueError("hello")
ValueError: hello
Without The With
What happens when we try to use managed_resource without with?
Nothing. Returned object is only holds a reference to a generator and has __enter__() and __exit__() methods. The context object will only return a value from the generator upon __enter__() call, and will run the rest of the code after yield during __exit__() call.
Alternative Error Handling In Exit Method
Instead of the contextmanager wrapper you can implement __enter__ and __exit__ methods on our custom object like below:
class ManagedResource:
def __init__(self, param):
self.param = param
def __enter__(self):
print(f"db connection acquired {self.param}")
return self
def __exit__(self, exc_type, exc_val, exc_tb):
if exc_type is not None:
print(f"rolling back due to ex {exc_val}")
print("returning db connection to a pool")
with ManagedResource("this is param value") as connection:
print(connection)
raise ValueError("hello")
Execution of above prints below.
db connection acquired this is param value
<__main__.ManagedResource object at 0x7f0dc9b1af70>
rolling back due to ex hello
returning db connection to a pool
Traceback (most recent call last):
File "/home/vackosar/src/vackosar.github.io/test/test.py", line 42, in <module>
raise ValueError("hello")
ValueError: hello
Retries Using Context Manager Are Not Possible - Here Is An Alternative
Do you need to retry when for example storage is momentarily not available?
However, You cannot implement retries of the code wrapped in with.
You simply cannot execute yield multiple times.
Instead, you can pass a callback to retry method containing a exponential backoff for loop below.
def retry_on_exception(
fun: Callable,
args=(),
kwargs: Optional[dict] = None,
retry_exceptions: Tuple = (Exception,),
max_retries: int = 3,
base_sleep_secs: float = 3.0,
):
if kwargs is None:
kwargs = dict()
ex = None
for retry in range(max_retries + 1):
try:
val = fun(*args, **kwargs)
return val
except retry_exceptions as e:
ex = e
sleep_secs = base_sleep_secs * 2 ** retry
sleep(sleep_secs)
raise RuntimeError(f"Too many retries ({max_retries}) of {fun.__name__}") from ex
Other Useful Posts
BCE architecture is the simplest way to structure your source code files, read all about it in my post on BCE.
Did you know that you can implement functional ForEach in Bash?
