"""module contains all db related stuff""" import logging import traceback import contextlib from urllib.parse import urlparse from typing import Any, Tuple, Iterator, Callable import MySQLdb as mdb from redis import Redis from MySQLdb.cursors import Cursor from flask import current_app as app def parse_db_url(db_url) -> Tuple: """ Parse SQL_URI configuration variable. """ parsed_db = urlparse(db_url) return (parsed_db.hostname, parsed_db.username, parsed_db.password, parsed_db.path[1:], parsed_db.port) @contextlib.contextmanager def database_connection(db_url: str) -> Iterator[mdb.Connection]: """function to create db connector""" host, user, passwd, db_name, db_port = parse_db_url(db_url) connection = mdb.connect( host, user, passwd, db_name, port=(db_port or 3306)) try: yield connection connection.commit() except mdb.Error as _mdb_err: logging.error(traceback.format_exc()) connection.rollback() finally: connection.close() def with_db_connection(func: Callable[[mdb.Connection], Any]) -> Any: """Call `func` with a MySQDdb database connection.""" with database_connection(app.config["SQL_URI"]) as conn: return func(conn) def with_redis_connection(func: Callable[[Redis], Any]) -> Any: """Call `func` with a redis connection.""" redisuri = app.config["REDIS_URL"] with Redis.from_url(redisuri, decode_responses=True) as rconn: return func(rconn) def debug_query(cursor: Cursor): """Debug the actual query run with MySQLdb""" for attr in ("_executed", "statement", "_last_executed"): if hasattr(cursor, attr): logging.debug("MySQLdb QUERY: %s", getattr(cursor, attr)) break