1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
|
"""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.debug(traceback.format_exc())
connection.rollback()
finally:
connection.close()
|
