luckywood.database.Database.open() - Code Metrics - gluecks-gmbh/luckywood - Measure and Improve Code Quality continuously with Scrutinizer

luckywood.database.Database.open() A
last analyzed 2023-03-29 20:26 UTC

↳ Parent: luckywood.database

Complexity

Conditions

Size

Total Lines	30
Code Lines	15

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	15
dl	0
loc	30
rs	9.65
c	0
b	0
f	0
cc	1
nop	0

"""
Helper class for mysql database usage
"""

__author__ = "Frederik Glücks"
__email__ = "[email protected]"
__copyright__ = "Frederik Glücks - Glücks GmbH"

# Built-in/Generic Imports
import os
import logging
import hashlib
from typing import List, Dict

# External libs/modules
import mysql.connector
import mysql.connector.cursor

# own libs
from .data_caching import DataCaching


class Database:
    """
    Helper class for mysql database usage
    """
    __cnx: mysql.connector = None
    __cursor: mysql.connector.cursor = None

    @staticmethod
    def open():
        """
        Creates a database connection by using environment vars.

        Environment vars:
        RDS_HOST = database host, default value "localhost"
        RDS_HOST_PORT = database port, default value "3306"
        RDS_DATABASE = database name, default value ""
        RDS_USERNAME = username, default value ""
        RDS_PASSWORD = password, default value ""

        :return: none
        """
        host: str = os.getenv('RDS_HOST', 'localhost')
        port: str = os.getenv('RDS_HOST_PORT', '3306')
        database: str = os.getenv('RDS_DATABASE', 'None')
        username: str = os.getenv('RDS_USERNAME', '')
        password: str = os.getenv('RDS_PASSWORD', '')

        Database.__cnx = mysql.connector.connect(username=username,
                                                 password=password,
                                                 host=host,
                                                 database=database,
                                                 port=port)

        Database.__cnx.autocommit = True
        Database.__cnx.get_warnings = True

        Database.__cursor = Database.__cnx.cursor(dictionary=True)

    @staticmethod
    def close():
        """
        Closes the database connection

        :return: none
        """
        if Database.__cnx:
            Database.__cursor.close()
            Database.__cnx.close()

    @staticmethod
    def fetchall_without_cache(query: str, param: tuple = ()) -> List[Dict[str, str]]:
        """
        Returns the result of the query as dict.

        The luckywood DataCaching class will NOT be used.

        :param query: str
        :param param: tuple
        :return: dict
        """
        if Database.__cursor is None:
            Database.open()

        if query == "":
            raise RuntimeError("Empty query")

        Database.__cursor.execute(query, param)

        return Database.__cursor.fetchall()

    @staticmethod
    def fetchall_with_cache(query: str, param: tuple = (),
                            force_update: bool = False) -> List[Dict[str, str]]:
        """
        Returns the result of the query as dict by using the luckywood DataCaching
        class.

        :param query: str
        :param param: tuple
        :param param: force_update
        :return: dict
        """
        tuple_str: str = ""

        for var in param:
            tuple_str += str(var)

        query_hash: str = hashlib.md5(
            query.encode() + tuple_str.encode()).hexdigest()
        cache_name = "database"

        if force_update is True:
            result_set: list = []
        else:
            result_set = DataCaching.get(cache_name, query_hash)

        if len(result_set) == 0:
            logging.info("Query load from database")
            result_set = Database.fetchall_without_cache(query, param)
            DataCaching.set(cache_name, query_hash, result_set)
        else:
            logging.info("Query load from cache")

        return result_set

    @staticmethod
    def query(query: str, param: tuple = ()) -> int:
        """
        Runs a query and returns the number of affected rows

        :param query: str
        :param param: tuple
        :return: int
        """
        if Database.__cursor is None:
            Database.open()

        Database.__cursor.execute(query, param)

        if Database.__cursor.fetchwarnings() is not None:
            logging.info(Database.__cursor.fetchwarnings())

        return Database.__cursor.rowcount

    @staticmethod
    def get_last_row_id() -> int:
        """
        This read-only property returns the value generated for an AUTO_INCREMENT column
        by the previous INSERT or UPDATE statement or None when there is no such value available.

        :return: int
        """
        if Database.__cursor is None:
            Database.open()

        return Database.__cursor.lastrowid

    @staticmethod
    def start_transaction() -> bool:
        """
        start a new MySQL Transaction

        :return: bool
        """
        if Database.__cursor is None:
            Database.open()

        Database.__cnx.start_transaction()

        return True

    @staticmethod
    def commit_transaction() -> bool:
        """
        commits the current transaction, making its changes permanent.

        :return: bool
        """
        if Database.__cursor is None:
            Database.open()

        Database.__cnx.commit()

        return True


1			"""
2			Helper class for mysql database usage
3			"""
4
5			__author__ = "Frederik Glücks"
6			__email__ = "[email protected]"
7			__copyright__ = "Frederik Glücks - Glücks GmbH"
8
9			# Built-in/Generic Imports
10			import os
11			import logging
12			import hashlib
13			from typing import List, Dict
14
15			# External libs/modules
16			import mysql.connector
17			import mysql.connector.cursor
18
19			# own libs
20			from .data_caching import DataCaching
21
22
23			class Database:
24			"""
25			Helper class for mysql database usage
26			"""
27			__cnx: mysql.connector = None
28			__cursor: mysql.connector.cursor = None
29
30			@staticmethod
31			def open():
32			"""
33			Creates a database connection by using environment vars.
34
35			Environment vars:
36			RDS_HOST = database host, default value "localhost"
37			RDS_HOST_PORT = database port, default value "3306"
38			RDS_DATABASE = database name, default value ""
39			RDS_USERNAME = username, default value ""
40			RDS_PASSWORD = password, default value ""
41
42			:return: none
43			"""
44			host: str = os.getenv('RDS_HOST', 'localhost')
45			port: str = os.getenv('RDS_HOST_PORT', '3306')
46			database: str = os.getenv('RDS_DATABASE', 'None')
47			username: str = os.getenv('RDS_USERNAME', '')
48			password: str = os.getenv('RDS_PASSWORD', '')
49
50			Database.__cnx = mysql.connector.connect(username=username,
51			password=password,
52			host=host,
53			database=database,
54			port=port)
55
56			Database.__cnx.autocommit = True
57			Database.__cnx.get_warnings = True
58
59			Database.__cursor = Database.__cnx.cursor(dictionary=True)
60
61			@staticmethod
62			def close():
63			"""
64			Closes the database connection
65
66			:return: none
67			"""
68			if Database.__cnx:
69			Database.__cursor.close()
70			Database.__cnx.close()
71
72			@staticmethod
73			def fetchall_without_cache(query: str, param: tuple = ()) -> List[Dict[str, str]]:
74			"""
75			Returns the result of the query as dict.
76
77			The luckywood DataCaching class will NOT be used.
78
79			:param query: str
80			:param param: tuple
81			:return: dict
82			"""
83			if Database.__cursor is None:
84			Database.open()
85
86			if query == "":
87			raise RuntimeError("Empty query")
88
89			Database.__cursor.execute(query, param)
90
91			return Database.__cursor.fetchall()
92
93			@staticmethod
94			def fetchall_with_cache(query: str, param: tuple = (),
95			force_update: bool = False) -> List[Dict[str, str]]:
96			"""
97			Returns the result of the query as dict by using the luckywood DataCaching
98			class.
99
100			:param query: str
101			:param param: tuple
102			:param param: force_update
103			:return: dict
104			"""
105			tuple_str: str = ""
106
107			for var in param:
108			tuple_str += str(var)
109
110			query_hash: str = hashlib.md5(
111			query.encode() + tuple_str.encode()).hexdigest()
112			cache_name = "database"
113
114			if force_update is True:
115			result_set: list = []
116			else:
117			result_set = DataCaching.get(cache_name, query_hash)
118
119			if len(result_set) == 0:
120			logging.info("Query load from database")
121			result_set = Database.fetchall_without_cache(query, param)
122			DataCaching.set(cache_name, query_hash, result_set)
123			else:
124			logging.info("Query load from cache")
125
126			return result_set
127
128			@staticmethod
129			def query(query: str, param: tuple = ()) -> int:
130			"""
131			Runs a query and returns the number of affected rows
132
133			:param query: str
134			:param param: tuple
135			:return: int
136			"""
137			if Database.__cursor is None:
138			Database.open()
139
140			Database.__cursor.execute(query, param)
141
142			if Database.__cursor.fetchwarnings() is not None:
143			logging.info(Database.__cursor.fetchwarnings())
144
145			return Database.__cursor.rowcount
146
147			@staticmethod
148			def get_last_row_id() -> int:
149			"""
150			This read-only property returns the value generated for an AUTO_INCREMENT column
151			by the previous INSERT or UPDATE statement or None when there is no such value available.
152
153			:return: int
154			"""
155			if Database.__cursor is None:
156			Database.open()
157
158			return Database.__cursor.lastrowid
159
160			@staticmethod
161			def start_transaction() -> bool:
162			"""
163			start a new MySQL Transaction
164
165			:return: bool
166			"""
167			if Database.__cursor is None:
168			Database.open()
169
170			Database.__cnx.start_transaction()
171
172			return True
173
174			@staticmethod
175			def commit_transaction() -> bool:
176			"""
177			commits the current transaction, making its changes permanent.
178
179			:return: bool
180			"""
181			if Database.__cursor is None:
182			Database.open()
183
184			Database.__cnx.commit()
185
186			return True
187

gluecks-gmbh / luckywood

luckywood.database.Database.open() A last analyzed 2023-03-29 20:26 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

luckywood.database.Database.open() A
last analyzed 2023-03-29 20:26 UTC