Skip to content

Database etiquette interface/base class

ISQL_Etiquette

Parent class for sql-facing classes

Source code in bayesian_ab_testing\utils.py 
 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
class ISQL_Etiquette:
    """Parent class for sql-facing classes"""
    def __init__(self):
        pass


    def chk_conn(conn) -> bool:
        """Check if connection is still alive

        Args:
            conn (sqlite3.connection): connection

        Returns:
            bool: True if is alive, false otherwise
        """        

        try:
            conn.cursor()
            return True
        except Exception as ex:
            return False


    def refresh_conn(self):
        """Make sure the connection is still alive"""
        if not ISQL_Etiquette.chk_conn(self.cnxn):
            self.cnxn = sqlite3.connect(db_path)


    def exec(self, query: str, *args, **kwargs) -> sqlite3.Cursor:
        """Executres a given query and commits immediately after

        Args:
            query (str): query string

        Returns:
            sqlite3.Cursor: cursor
        """       
        self.refresh_conn()

        cur = self.cnxn.execute(query, *args, **kwargs)
        self.cnxn.commit() 

        return cur


    def exec_many(self, query: str, *args, **kwargs) -> sqlite3.Cursor:
        """Executres many queries and commits immediately after

        Args:
            query (str): query string

        Returns:
            sqlite3.Cursor: cursor
        """        

        self.refresh_conn()

        cur = self.cnxn.executemany(query, *args, **kwargs)
        self.cnxn.commit()

        return cur

chk_conn(conn)

Check if connection is still alive

Parameters:

Name Type Description Default
conn connection

connection

 required

Returns:

Name Type Description
bool bool

True if is alive, false otherwise
Source code in bayesian_ab_testing\utils.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
def chk_conn(conn) -> bool:
    """Check if connection is still alive

    Args:
        conn (sqlite3.connection): connection

    Returns:
        bool: True if is alive, false otherwise
    """        

    try:
        conn.cursor()
        return True
    except Exception as ex:
        return False

exec(query, *args, **kwargs)

Executres a given query and commits immediately after

Parameters:

Name Type Description Default
query str

query string

 required

Returns:

Type Description
Cursor

sqlite3.Cursor: cursor
Source code in bayesian_ab_testing\utils.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def exec(self, query: str, *args, **kwargs) -> sqlite3.Cursor:
    """Executres a given query and commits immediately after

    Args:
        query (str): query string

    Returns:
        sqlite3.Cursor: cursor
    """       
    self.refresh_conn()

    cur = self.cnxn.execute(query, *args, **kwargs)
    self.cnxn.commit() 

    return cur

exec_many(query, *args, **kwargs)

Executres many queries and commits immediately after

Parameters:

Name Type Description Default
query str

query string

 required

Returns:

Type Description
Cursor

sqlite3.Cursor: cursor
Source code in bayesian_ab_testing\utils.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def exec_many(self, query: str, *args, **kwargs) -> sqlite3.Cursor:
    """Executres many queries and commits immediately after

    Args:
        query (str): query string

    Returns:
        sqlite3.Cursor: cursor
    """        

    self.refresh_conn()

    cur = self.cnxn.executemany(query, *args, **kwargs)
    self.cnxn.commit()

    return cur

refresh_conn()

Make sure the connection is still alive

Source code in bayesian_ab_testing\utils.py 
31
32
33
34
def refresh_conn(self):
    """Make sure the connection is still alive"""
    if not ISQL_Etiquette.chk_conn(self.cnxn):
        self.cnxn = sqlite3.connect(db_path)




Logger

CustomFormatter

Bases: Formatter

Custom formatter for Informative Logging

Source code in bayesian_ab_testing\logger\logger.py 
 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
class CustomFormatter(logging.Formatter):

    """ 
        Custom formatter for Informative Logging
    """
    grey = "\x1b[38;20m"
    violet="\x1b[38;5;183m"
    yellow = "\x1b[33;20m"
    red = "\x1b[31;20m"
    bold_red = "\x1b[31;1m"
    reset = "\x1b[0m"
    format = "%(asctime)s - %(name)s - %(funcName)s - %(levelname)s - (%(message)s) - line: %(lineno)d"
    FORMATS = {
        logging.DEBUG: grey + format + reset,
        logging.INFO: violet + format + reset,
        logging.WARNING: yellow + format + reset,
        logging.ERROR: red + format + reset,
        logging.CRITICAL: bold_red + format + reset
    }
    def format(self, record):
        """

        Parameters
        ----------
        record : takes the record


        Returns
        -------
        returns formated(colored) output
        """
        log_fmt = self.FORMATS.get(record.levelno)
        formatter = logging.Formatter(log_fmt)
        return formatter.format(record)

format(record)

Parameters

record : takes the record

Returns

returns formated(colored) output

Source code in bayesian_ab_testing\logger\logger.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
def format(self, record):
    """

    Parameters
    ----------
    record : takes the record


    Returns
    -------
    returns formated(colored) output
    """
    log_fmt = self.FORMATS.get(record.levelno)
    formatter = logging.Formatter(log_fmt)
    return formatter.format(record)