I've been implementing a lot of database interaction into Python recently and I've noticed there's no easy way to automate connection and session management resulting in boilerplate code into every function related to a transaction.
No matter what I try, I always end up with the following 2 scenarios:
1 - The cursor needs to be specified as an parameter in every function that needs it + a decorator.
2 - I need to explicitly declare the connection with or without context managers and then
explicitly close them at the end of each function with conn.close() and cursor.close(). It's
precisely these 2 lines I'm trying to avoid as they're always the same and always in the same
place at the end of the function.
So far the only 2 options I've really tried are a db function for creating the connection and the cursor, and a decorator for the whole thing.
The thing I actually use and works:
import mariadb as db
def db_create_connection():
return db.connect(
user='...',
password='...',
host='...',
database='...',
port=...
)
def db_some_fn():
with db_create_connection() as conn:
with conn.cursor() as cursor:
...
conn.commit()
cursor.close()
conn.close()
The second attempt that is a decorator but requires every function to have cursor as a parameter:
import mariadb as db
from functools import wraps
def with_db_cursor(func):
@wraps(func)
def wrapper(*args, **kwargs):
conn, cursor = None, None
try:
conn = db.connect(
user='...',
password='...',
host='...',
database='...',
port=...
)
cursor = conn.cursor()
result = func(cursor, *args, **kwargs)
conn.commit()
return result
except Exception as e:
if conn:
conn.rollback()
print(f"An error occurred: {e}")
raise
finally:
if cursor:
cursor.close()
if conn:
conn.close()
return wrapper
I've tried returning the connection and cursor both, closing them with the finally condition (hope dies last) and so forth, but I can't evade the boilerplate.
I know the decorator more or less solves the problem but I'm worried someone will end up messing with the cursor in one of the functions and then off to debugging land it is.
Is there a solution or am I just asking for too much?
Not very familiar with the mariadb connector, and I cannot find the documentation on the context managers, but do they not handle closing the connection / cursor automatically?
Anyway, you could create your own context manager like this:
from contextlib import contextmanager
@contextmanager
def db_cursor():
with db_create_connection() as conn:
with conn.cursor() as cursor:
try:
yield cursor
finally:
# if those actually need to be done
conn.commit()
if not cursor.closed: cursor.close()
else:
warnings.warn("content of the db_cursor context manager closed the cursor. Please don't do that.")
conn.close()
Now you can use this context manager instead of the ones you are currently using
def db_some_fn():
with db_cursor() as cursor:
...
#automatically commits / closes everything when exiting the context manager
Now if the new level of indentation is a problem to you, you could still wrap it in a decorator
from inspect import signature, Parameter
def needs_db_cursor(func):
func_params = signature(func).parameters
assert "cursor" in func_params, f"{func} was decorated with @needs_db_cursor, but does not take a ``cursor`` argument"
assert func_params["cursor"] in (Parameter.POSITIONAL_OR_KEYWORD, Parameter.KEYWORD_ONLY), f"{func} takes a ``cursor`` argument, but not as a keyword argument..."
@wraps(func)
def wrapper(*args, **kwargs):
with db_cursor() as cursor:
return func(*args, **kwargs, cursor=cursor)
@needs_db_cursor
def db_some_fn(cursor):
...
#automatically commits / closes everything when exiting the context manager