MarcSimplePythonDatabaseModule/database_records.py at master · marcanthonyconcepcion/MarcSimplePythonDatabaseModule · GitHub

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
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
__author__ = "Marc Concepcion"
__copyright__ = "Copyright 2021, Marc Concepcion"
__credits__ = ["Marc Concepcion"]
__maintainer__ = "Marc Concepcion"
__email__ = "marcanthonyconcepcion@gmail.com"
__status__ = "Demo"

from typing import Final
import yaml
import mysql.connector

with open(r'resources\database_demo.yaml') as file:
    configuration = yaml.full_load(file)

HOST: Final = configuration['database']['host']
DBNAME: Final = configuration['database']['dbname']
USER: Final = configuration['database']['user']
PASSWORD: Final = configuration['database']['password']


class DatabaseRecords:

    def __init__(self):
        try:
            self.database = mysql.connector.connect(host=HOST, database=DBNAME,
                                                    user=USER, password=PASSWORD)
            self.cursor = self.database.cursor()
        except mysql.connector.errors.Error as database_error:
            raise DatabaseError('Database error. {details}'.format(details=database_error))

    def disconnect(self):
        self.database.disconnect()

    def test_connection(self):
        return self.database.is_connected()

    def reconnect(self):
        try:
            self.database.reconnect()
        except mysql.connector.errors.Error as database_error:
            raise DatabaseError('Database error. {details}'.format(details=database_error))

    def fetch(self, query, parameters=None):
        try:
            self.cursor.execute(operation=query, params=parameters)
            for result in self.cursor:
                yield result
        except mysql.connector.errors.Error as database_error:
            raise DatabaseError('Database error. {details}'.format(details=database_error))
        finally:
            self.cursor.reset()

    def edit(self, query, parameters=None):
        try:
            self.cursor.execute(operation=query, params=parameters)
            self.database.commit()
        except mysql.connector.errors.Error as database_error:
            self.database.rollback()
            raise DatabaseError('Database error. {details}'.format(details=database_error))

    def bulk_edit(self, query, parameter_sets):
        try:
            self.cursor.executemany(operation=query, seq_params=parameter_sets)
            self.database.commit()
        except mysql.connector.errors.Error as database_error:
            self.database.rollback()
            raise DatabaseError('Database error. {details}'.format(details=database_error))

    def transact_edit(self, queries):
        try:
            for query in queries:
                self.cursor.execute(operation=query['statement'], params=query['parameters'])
            self.database.commit()
        except mysql.connector.errors.Error as database_error:
            self.database.rollback()
            raise DatabaseError('Database error. {details}'.format(details=database_error))

    __instance__ = None

    @staticmethod
    def get():
        if DatabaseRecords.__instance__ is None:
            DatabaseRecords.__instance__ = DatabaseRecords()
        return DatabaseRecords.__instance__


class DatabaseError(Exception):
    pass