blob: a934fd9324f07d4bcef83cc35d2a476c2f46fdcf (
plain)
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
|
"""Connections to MariaDB"""
import traceback
import contextlib
from typing import Iterator
import MySQLdb as mdb
from .protocols import DbConnection
@contextlib.contextmanager
def database_connection(sql_uri) -> Iterator[DbConnection]:
"""Connect to MySQL database."""
host, user, passwd, db_name, port = parse_db_url(sql_uri)
connection = mdb.connect(db=db_name,
user=user,
passwd=passwd or '',
host=host,
port=port or 3306)
try:
yield connection
except Exception as _exc: # TODO: Make the Exception class less general
logging.debug(traceback.format_exc())
connection.rollback()
finally:
connection.commit()
connection.close()
|
