about summary refs log tree commit diff
path: root/qc_app/db_utils.py
diff options
context:
space:
mode:
authorFrederick Muriuki Muriithi2024-07-25 11:07:33 -0500
committerFrederick Muriuki Muriithi2024-07-25 14:34:09 -0500
commit754e8f214b940e05298cb360ed829f5c685d55a5 (patch)
tree62c2c5b601746621f0949b38937ad232f006dee2 /qc_app/db_utils.py
parentde9e1b9fe37928b864bea28b408de6c14d04526b (diff)
downloadgn-uploader-754e8f214b940e05298cb360ed829f5c685d55a5.tar.gz
Rename module: qc_app --> uploader
Diffstat (limited to 'qc_app/db_utils.py')
-rw-r--r--qc_app/db_utils.py46
1 files changed, 0 insertions, 46 deletions
diff --git a/qc_app/db_utils.py b/qc_app/db_utils.py
deleted file mode 100644
index ef26398..0000000
--- a/qc_app/db_utils.py
+++ /dev/null
@@ -1,46 +0,0 @@
-"""module contains all db related stuff"""
-import logging
-import traceback
-import contextlib
-from urllib.parse import urlparse
-from typing import Any, Tuple, Optional, Iterator, Callable
-
-import MySQLdb as mdb
-from redis import Redis
-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()
-
-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)