"""module contains all db related stuff""" import logging import traceback import contextlib from typing import Tuple, Optional, Iterator from urllib.parse import urlparse import MySQLdb as mdb 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: Optional[str] = None) -> Iterator[mdb.Connection]: """function to create db connector""" host, user, passwd, db_name, db_port = parse_db_url( db_url or app.config["SQL_URI"]) 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()