Skip to content

Bot's Database

Design Pattern

The interaction with the database is done using a variation of the command design pattern, the implementation is inspired and derived from the Practices Of The Python Pro Book..

Click to view the UML diagram of the implementation

db_uml_diagram
Database UML diagram.

Pattern Interfaces

Commands Interface

The invokers call and execute commands which are concrete classes implementing the ICommand interface that has an execute() method.

Click to View the ICommand Interface

Bases: Protocol

This protocol abstracts the implementation of the predefined database commands classes

Source code in src/database/db_commands.py 
13
14
15
16
17
class ICommand(Protocol):
    """This protocol abstracts the implementation of the predefined database commands classes"""

    def execute(self):
        """Implement the command execution"""
execute()

Implement the command execution

Source code in src/database/db_commands.py 
16
17
def execute(self):
    """Implement the command execution"""

Persistence Interface

Each command calls a corresponding method in the Persistence class which is a concrete class implementing the IPersistenceLayer interface controlling the DatabaseManger class.

Click to View the IPersistenceLayer Interface

Bases: ABC

An abstracted interface for the database controller

Source code in src/database/persistence.py 
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
class IPersistenceLayer(ABC):
    """An abstracted interface for the database controller"""

    @abstractmethod
    def add_user(self, data) -> None:
        raise NotImplementedError("Database Controller must implement a create method")

    @abstractmethod
    def add_to_allow_list(self, data) -> None:
        raise NotImplementedError("Database Controller must implement a create method")

    @abstractmethod
    def get_user(self, order_by: str = None) -> list:
        raise NotImplementedError("Database Controller must implement a get method")

    @abstractmethod
    def get_group(self, order_by: str = None) -> list:
        raise NotImplementedError("Database Controller must implement a get method")

    @abstractmethod
    def edit(self, id: str) -> None:
        raise NotImplementedError("Database Controller must implement a edit method")

    @abstractmethod
    def delete(self, id: str) -> None:
        raise NotImplementedError("Database Controller must implement a delete method")

Pattern Implementation

Concrete Commands

After defining the the ICommand interface, we can easily create concrete command classes that implement it's functionality, the following is an example of the AddUserCommand() command.

Click to view the AddUserCommand() concrete class

Bases: ICommand

The command adds a new user to the database using the INSERT INTO SQL statement.

Source code in src/database/db_commands.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
class AddUserCommand(ICommand):
    """The command adds a new user to the database using the INSERT INTO SQL statement."""

    def __init__(self, *, user_id: str, block_type: str) -> None:
        """_summary_ : This method gets the data to initiate the command to add a user to the database.

        Parameters
        ----------
        user_id : str
            _description_ : User's user_id.
        block_type : str
            _description_ : User's block type (temp | perm).
        """
        # Storing the user's id.
        self.user_id = user_id
        # Lowering the block time string, and stripping it from any leading|tailing space.
        self.block_type = (block_type.lower()).strip()

    def execute(self) -> None:
        """This method executes the 'INSERT INTO' statement."""
        # Calling the add_user method with the user's_id and his/her block type.
        persistence.add_user(self.user_id, self.block_type)
__init__(*, user_id: str, block_type: str) -> None

summary : This method gets the data to initiate the command to add a user to the database.

Parameters:

Name Type Description Default
user_id str

description : User's user_id.

 required
block_type str

description : User's block type (temp | perm).

 required
Source code in src/database/db_commands.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
def __init__(self, *, user_id: str, block_type: str) -> None:
    """_summary_ : This method gets the data to initiate the command to add a user to the database.

    Parameters
    ----------
    user_id : str
        _description_ : User's user_id.
    block_type : str
        _description_ : User's block type (temp | perm).
    """
    # Storing the user's id.
    self.user_id = user_id
    # Lowering the block time string, and stripping it from any leading|tailing space.
    self.block_type = (block_type.lower()).strip()
execute() -> None

This method executes the 'INSERT INTO' statement.

Source code in src/database/db_commands.py 
38
39
40
41
def execute(self) -> None:
    """This method executes the 'INSERT INTO' statement."""
    # Calling the add_user method with the user's_id and his/her block type.
    persistence.add_user(self.user_id, self.block_type)

Concrete Persistence

The UsersDatabase concrete class is an implementation of the IPersistenceLayer interface which sits between the command classes and the DatabaseManger class that executes CURL SQL statements on the database.

The UsersDatabase class is a lower level of abstraction acting on top of the DatabaseManger class with a predefined SQL statements designed specifically for the Users and allow groups list tables in the database.

Click to view the UsersDatabase class

Bases: IPersistenceLayer

This class sits between the database commands and the database manger class

Source code in src/database/persistence.py 
 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
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
class UsersDatabase(IPersistenceLayer):
    """This class sits between the database commands and the database manger class"""

    def __init__(self) -> None:
        """_summary_ : This creates the 2 tables 'users' and 'allow_list'"""
        # Table name to be created if not existing
        self.table_name = "bot_users"
        # Initiating the data base users
        self.db = DatabaseManger("bot_db.sqlite")

        # Creating the table 'bot_users' in the database
        self.db.create_table(
            self.table_name,
            {
                # "id": "integer primary key autoincrement",
                "user_id": "text primary key not null",
                "block_type": "text not null",
                "date_added": "text not null",
            },
        )
        # Creating the allowlist table
        self.db.create_table(
            "allow_list",
            {
                "group_id": "text primary key not null",
            },
        )

    def add_user(self, user_id: str, block_type: str) -> None:
        """_summary_ : This methods adds users to the database.

        Parameters
        ----------
        user_id : str
            _description_ : user id.
        block_type : str
            _description_ : 'perm' for permanently block | 'temp' for temporary block
        """
        # Getting the current date to be added as an attribute to the user record
        date = datetime.now().strftime("%Y/%m/%d, %H:%M:%S")
        # Added the user to the database
        self.db.add(
            self.table_name,
            {"user_id": user_id, "block_type": block_type, "date_added": date},
        )

    def add_to_allow_list(self, group_id: str) -> None:
        """_summary_ : This methods adds users to the database.

        Parameters
        ----------
        group_id : str
            _description_ : group_id to allow
        """
        # Adds the user to the database
        self.db.add(
            "allow_list",
            {"group_id": group_id},
        )

    def get_user(
        self, user_id: str = None, block_type: str = None, order_by: str = None
    ) -> list:
        """_summary_ : This methods selects users from the database.

        Parameters
        ----------
        user_id : str, optional
            _description_, by default None : user_id to filter by.
        order_by : str, optional
            _description_, by default None : Column name to order the query result by, be default it will be order by the date added.

        Returns
        -------
        list
            _description_ : A list of tuples containing each records
        """
        # If a user_id and a block type were provided adds it to the select criteria.
        if block_type:
            select_criteria = {"block_type": block_type}
        else:
            # If no user_id was provided pass on an empty dict to the criteria parameter
            select_criteria = {"user_id": user_id} if user_id else None

        # Returning the list of records that fitted the query select criteria
        return self.db.select(
            self.table_name, criteria=select_criteria, order_by=order_by
        ).fetchall()

    def get_group(self, group_id: str = None, order_by: str = None) -> list:
        """_summary_ : This methods selects group from the database.

        Parameters
        ----------
        group_id : str, optional
            _description_, by default None : group_id to filter by.
        order_by : str, optional
            _description_, by default None : Column name to order the query result by, be default it will be order by the date added.

        Returns
        -------
        list
            _description_ : A list of tuples containing each records
        """
        # If no group_id was provided pass on an empty dict to the criteria parameter
        select_criteria = {"group_id": group_id} if group_id else None

        # Returning the list of records that fitted the query select criteria
        return self.db.select(
            "allow_list", criteria=select_criteria, order_by=order_by
        ).fetchall()

    def edit(self, user_id: str, block_type: str):
        """_summary_ : This method update the records in the database using the user user_id as a criteria.

        Parameters
        ----------
        user_id : str
            _description_ : user's user_id.
        block_type : str
            _description_ : The block type of the user perm for permanent | temp for temporary
        """
        # Sending the update statement to the database
        self.db.update(
            self.table_name, {"user_id": user_id}, {"block_type": block_type}
        )

    def delete(self, user_id: str):
        """_summary_ : This method deletes user from the database using his user_id as a criteria.

        Parameters
        ----------
        user_id : str
            _description_ : user's user_id.
        """
        # Deleting the user record using his user_id as a filter criteria
        self.db.delete(self.table_name, {"user_id": user_id})

    def delete_group(self, group_id: str):
        """_summary_ : This method deletes group from the database using his group_id as a criteria.

        Parameters
        ----------
        group_id : str
            _description_ : group's id.
        """
        # Deleting the user record using his user_id as a filter criteria
        self.db.delete("allow_list", {"group_id": group_id})
__init__() -> None

summary : This creates the 2 tables 'users' and 'allow_list'

Source code in src/database/persistence.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def __init__(self) -> None:
    """_summary_ : This creates the 2 tables 'users' and 'allow_list'"""
    # Table name to be created if not existing
    self.table_name = "bot_users"
    # Initiating the data base users
    self.db = DatabaseManger("bot_db.sqlite")

    # Creating the table 'bot_users' in the database
    self.db.create_table(
        self.table_name,
        {
            # "id": "integer primary key autoincrement",
            "user_id": "text primary key not null",
            "block_type": "text not null",
            "date_added": "text not null",
        },
    )
    # Creating the allowlist table
    self.db.create_table(
        "allow_list",
        {
            "group_id": "text primary key not null",
        },
    )
add_to_allow_list(group_id: str) -> None

summary : This methods adds users to the database.

Parameters:

Name Type Description Default
group_id str

description : group_id to allow

 required
Source code in src/database/persistence.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def add_to_allow_list(self, group_id: str) -> None:
    """_summary_ : This methods adds users to the database.

    Parameters
    ----------
    group_id : str
        _description_ : group_id to allow
    """
    # Adds the user to the database
    self.db.add(
        "allow_list",
        {"group_id": group_id},
    )
add_user(user_id: str, block_type: str) -> None

summary : This methods adds users to the database.

Parameters:

Name Type Description Default
user_id str

description : user id.

 required
block_type str

description : 'perm' for permanently block | 'temp' for temporary block

 required
Source code in src/database/persistence.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def add_user(self, user_id: str, block_type: str) -> None:
    """_summary_ : This methods adds users to the database.

    Parameters
    ----------
    user_id : str
        _description_ : user id.
    block_type : str
        _description_ : 'perm' for permanently block | 'temp' for temporary block
    """
    # Getting the current date to be added as an attribute to the user record
    date = datetime.now().strftime("%Y/%m/%d, %H:%M:%S")
    # Added the user to the database
    self.db.add(
        self.table_name,
        {"user_id": user_id, "block_type": block_type, "date_added": date},
    )
delete(user_id: str)

summary : This method deletes user from the database using his user_id as a criteria.

Parameters:

Name Type Description Default
user_id str

description : user's user_id.

 required
Source code in src/database/persistence.py 
170
171
172
173
174
175
176
177
178
179
def delete(self, user_id: str):
    """_summary_ : This method deletes user from the database using his user_id as a criteria.

    Parameters
    ----------
    user_id : str
        _description_ : user's user_id.
    """
    # Deleting the user record using his user_id as a filter criteria
    self.db.delete(self.table_name, {"user_id": user_id})
delete_group(group_id: str)

summary : This method deletes group from the database using his group_id as a criteria.

Parameters:

Name Type Description Default
group_id str

description : group's id.

 required
Source code in src/database/persistence.py 
181
182
183
184
185
186
187
188
189
190
def delete_group(self, group_id: str):
    """_summary_ : This method deletes group from the database using his group_id as a criteria.

    Parameters
    ----------
    group_id : str
        _description_ : group's id.
    """
    # Deleting the user record using his user_id as a filter criteria
    self.db.delete("allow_list", {"group_id": group_id})
edit(user_id: str, block_type: str)

summary : This method update the records in the database using the user user_id as a criteria.

Parameters:

Name Type Description Default
user_id str

description : user's user_id.

 required
block_type str

description : The block type of the user perm for permanent | temp for temporary

 required
Source code in src/database/persistence.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
def edit(self, user_id: str, block_type: str):
    """_summary_ : This method update the records in the database using the user user_id as a criteria.

    Parameters
    ----------
    user_id : str
        _description_ : user's user_id.
    block_type : str
        _description_ : The block type of the user perm for permanent | temp for temporary
    """
    # Sending the update statement to the database
    self.db.update(
        self.table_name, {"user_id": user_id}, {"block_type": block_type}
    )
get_group(group_id: str = None, order_by: str = None) -> list

summary : This methods selects group from the database.

Parameters:

Name Type Description Default
group_id str

description, by default None : group_id to filter by.

 None
order_by str

description, by default None : Column name to order the query result by, be default it will be order by the date added.

 None

Returns:

Type Description
list

description : A list of tuples containing each records
Source code in src/database/persistence.py 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def get_group(self, group_id: str = None, order_by: str = None) -> list:
    """_summary_ : This methods selects group from the database.

    Parameters
    ----------
    group_id : str, optional
        _description_, by default None : group_id to filter by.
    order_by : str, optional
        _description_, by default None : Column name to order the query result by, be default it will be order by the date added.

    Returns
    -------
    list
        _description_ : A list of tuples containing each records
    """
    # If no group_id was provided pass on an empty dict to the criteria parameter
    select_criteria = {"group_id": group_id} if group_id else None

    # Returning the list of records that fitted the query select criteria
    return self.db.select(
        "allow_list", criteria=select_criteria, order_by=order_by
    ).fetchall()
get_user(user_id: str = None, block_type: str = None, order_by: str = None) -> list

summary : This methods selects users from the database.

Parameters:

Name Type Description Default
user_id str

description, by default None : user_id to filter by.

 None
order_by str

description, by default None : Column name to order the query result by, be default it will be order by the date added.

 None

Returns:

Type Description
list

description : A list of tuples containing each records
Source code in src/database/persistence.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
def get_user(
    self, user_id: str = None, block_type: str = None, order_by: str = None
) -> list:
    """_summary_ : This methods selects users from the database.

    Parameters
    ----------
    user_id : str, optional
        _description_, by default None : user_id to filter by.
    order_by : str, optional
        _description_, by default None : Column name to order the query result by, be default it will be order by the date added.

    Returns
    -------
    list
        _description_ : A list of tuples containing each records
    """
    # If a user_id and a block type were provided adds it to the select criteria.
    if block_type:
        select_criteria = {"block_type": block_type}
    else:
        # If no user_id was provided pass on an empty dict to the criteria parameter
        select_criteria = {"user_id": user_id} if user_id else None

    # Returning the list of records that fitted the query select criteria
    return self.db.select(
        self.table_name, criteria=select_criteria, order_by=order_by
    ).fetchall()

The DatabaseManger

The DatabaseManger class initializes the connection with the database (sqlite3), and executes SQL statements on the database using CURL commands.

Click to View the DatabaseManger class

This class manges the connection to the sqlite database.

Source code in src/database/db_manger.py 
 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
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
class DatabaseManger:
    """This class manges the connection to the sqlite database."""

    def __init__(self, database_filename: Path) -> None:
        """_summary_ : Initializing a connection with the database.

        Parameters
        ----------
        database_filename : Path
            _description_ : A path to the database file to connect to, if doesn't exist, it will be created.
        """
        self.connection = sqlite3.connect(database_filename, check_same_thread=False)

    def __del__(self) -> None:
        """_summary_ : This method closes the connection with the database."""
        self.connection.close()

    def _execute(self, statement: str, values: tuple[str] = None) -> Cursor:
        """_summary_ : This method executes SQL statements and returns back a Cursor object containing the query result if any.

        Parameters
        ----------
        statement : str
            _description_ : The SQL statement to execute on the database.
        values : tuple[str], optional
            _description_, by default None : Tuples containing the values to replace the placeholders with to prevent sql injection.

        Returns
        -------
        Cursor
            _description_ : A Cursor object containing the result of the query.
        """
        # Opening a connection to the database using a context manger to automatically close when done
        with self.connection:
            # Create the cursor object
            cursor = self.connection.cursor()
            # Executing the received statement
            cursor.execute(statement, values or [])
            # Returning the cursor object to extract returned data if any
            return cursor

    def create_table(self, table_name: str, columns: dict[str, str]) -> None:
        """_summary_ : This method creates a table in the database if the table is note existing.

        Parameters
        ----------
        table_name : str
            _description_ : The name of the table to create.
        columns : dict[str, str]
            _description_ : A dict holding the columns names as a key and their types as the value

        """
        # Creating a list of column followed by their data types from the provided dict
        columns_with_types = [
            f"{column_name} {data_type}" for column_name, data_type in columns.items()
        ]

        # Executing the table creation SQL statement using the provided table name and columns names
        self._execute(
            f"""
                CREATE TABLE IF NOT EXISTS {table_name}
                ({", ".join(columns_with_types)});
            """
        )

    def add(self, table_name: str, data: dict[str, str]) -> None:
        """_summary_ : This method adds data into the database using the 'INSERT INTO' SQL statement.

        Parameters
        ----------
        table_name : str
            _description_ : Table name to perform the statement on.
        data : dict[str, str]
            _description_ : A dict of strings, specifying the columns names and the columns values to be added to the table.
        """
        # Creating placeholders for the provided data
        placeholders = ", ".join("?" * len(data))
        # Getting the columns names from the data: [dict] keys
        column_names = ", ".join(data.keys())
        # Getting the columns values from the data: [dict] values and storing them in a tuple to pass them to the _execute method
        column_values = tuple(data.values())

        # Executing the 'INSERT INTO' statement on the database, ignoring; if records already exists
        self._execute(
            f"""
            INSERT OR IGNORE INTO {table_name}
            ({column_names})
            VALUES ({placeholders})
            """,
            column_values,
        )

    def select(
        self, table_name: str, criteria: dict[str, str] = None, order_by: str = None
    ) -> Cursor:
        """_summary_: : This method selects data from the database using the 'SELECT' SQL statement and returns back  data.

        Parameters
        ----------
        table_name : str
            _description_ : Table name to perform the statement on.
        criteria : dict[str, str], optional
            _description_, by default None : The criteria to use as a filter on the SELECT statement, passed as a dict => {keys(criteria) : values(values)}
        order_by : str, optional
            _description_, by default None : A column name to sort the returned query by, will map to the ORDER BY {order_by}

        Returns
        -------
        Cursor
            _description_ : A Cursor object containing the data back from the database of any.
        """
        # Creating an empty dict if no criteria was provided
        criteria = criteria or {}

        # Creating the query 'SELECT' statement
        query = f"SELECT * FROM {table_name}"

        # If criteria was provided parse it and add placeholders for it
        if criteria:
            # Creating placeholders fro the provided criteria
            placeholders = [f"{column} = ?" for column in criteria.keys()]
            # Joining each statement with 'AND'
            select_criteria = " AND ".join(placeholders)
            # Adding the 'select_criteria' to the query after the 'WHERE' SQL statement
            query += f" WHERE {select_criteria}"

        # If order_by was provided adds in a 'ORDER BY' SQL statement to sort the data
        if order_by:
            # Adding the 'ORDER BY' statement to the query followed by the column name to sort by
            query += f" ORDER BY {order_by}"

        # Executing the query and passing the criteria values to the cursor and returning back the results
        return self._execute(query, tuple(criteria.values()))

    def update(
        self, table_name: str, criteria: dict[str, str], data: dict[str, str]
    ) -> None:
        """_summary_ : This method updates data in the database using the 'UPDATE' statement.

        Parameters
        ----------
        table_name : str
            _description_ : Table name to perform the statement on.
        criteria : dict[str, str]
            _description_ : The criteria to use as a filter on the UPDATE statement, passed as a dict => {keys(criteria) : values(values)}
        data : dict[str, str]
            _description_ : The actual data to update
        """
        # Creating placeholders for the update criteria
        placeholders = [f"{column} = ?" for column in criteria]
        # Joining the placeholders with the AND operator
        update_criteria = " AND ".join(placeholders)
        # Creating placeholders for the update data
        data_placeholders = ", ".join(f"{key} = ?" for key in data)
        # Creating two tuples holding the values to update and the new data to update them with
        values = tuple(data.values()) + tuple(criteria.values())
        # Executing the UPDATE statement
        self._execute(
            f"""
            UPDATE {table_name}
            SET {data_placeholders}
            WHERE {update_criteria};
            """,
            values,
        )

    def delete(self, table_name: str, criteria: dict[str, str]) -> None:
        """_summary_ : This method deletes data from the database using the 'DELETE' statement.

        Parameters
        ----------
        table_name : str
            _description_ : Table name to perform the statement on.
        criteria : dict[str, str]
            _description_ : The criteria to use as a filter on the DELETE statement, passed as a dict => {keys(criteria) : values(values)}
        """
        # Creating placeholders for the provided criteria
        placeholders = [
            f"{column} = ?" for column in criteria
        ]  #! add criteria.keys() if didn't work
        # Joining the created placeholders with AND operator
        delete_criteria = " AND ".join(placeholders)
        # Executing the DELETE statement
        self._execute(
            f"""
            DELETE FROM {table_name}
            WHERE {delete_criteria}
            """,
            tuple(criteria.values()),
        )
__del__() -> None

summary : This method closes the connection with the database.

Source code in src/database/db_manger.py 
26
27
28
def __del__(self) -> None:
    """_summary_ : This method closes the connection with the database."""
    self.connection.close()
__init__(database_filename: Path) -> None

summary : Initializing a connection with the database.

Parameters:

Name Type Description Default
database_filename Path

description : A path to the database file to connect to, if doesn't exist, it will be created.

 required
Source code in src/database/db_manger.py 
16
17
18
19
20
21
22
23
24
def __init__(self, database_filename: Path) -> None:
    """_summary_ : Initializing a connection with the database.

    Parameters
    ----------
    database_filename : Path
        _description_ : A path to the database file to connect to, if doesn't exist, it will be created.
    """
    self.connection = sqlite3.connect(database_filename, check_same_thread=False)
add(table_name: str, data: dict[str, str]) -> None

summary : This method adds data into the database using the 'INSERT INTO' SQL statement.

Parameters:

Name Type Description Default
table_name str

description : Table name to perform the statement on.

 required
data dict[str, str]

description : A dict of strings, specifying the columns names and the columns values to be added to the table.

 required
Source code in src/database/db_manger.py 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def add(self, table_name: str, data: dict[str, str]) -> None:
    """_summary_ : This method adds data into the database using the 'INSERT INTO' SQL statement.

    Parameters
    ----------
    table_name : str
        _description_ : Table name to perform the statement on.
    data : dict[str, str]
        _description_ : A dict of strings, specifying the columns names and the columns values to be added to the table.
    """
    # Creating placeholders for the provided data
    placeholders = ", ".join("?" * len(data))
    # Getting the columns names from the data: [dict] keys
    column_names = ", ".join(data.keys())
    # Getting the columns values from the data: [dict] values and storing them in a tuple to pass them to the _execute method
    column_values = tuple(data.values())

    # Executing the 'INSERT INTO' statement on the database, ignoring; if records already exists
    self._execute(
        f"""
        INSERT OR IGNORE INTO {table_name}
        ({column_names})
        VALUES ({placeholders})
        """,
        column_values,
    )
create_table(table_name: str, columns: dict[str, str]) -> None

summary : This method creates a table in the database if the table is note existing.

Parameters:

Name Type Description Default
table_name str

description : The name of the table to create.

 required
columns dict[str, str]

description : A dict holding the columns names as a key and their types as the value

 required
Source code in src/database/db_manger.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
def create_table(self, table_name: str, columns: dict[str, str]) -> None:
    """_summary_ : This method creates a table in the database if the table is note existing.

    Parameters
    ----------
    table_name : str
        _description_ : The name of the table to create.
    columns : dict[str, str]
        _description_ : A dict holding the columns names as a key and their types as the value

    """
    # Creating a list of column followed by their data types from the provided dict
    columns_with_types = [
        f"{column_name} {data_type}" for column_name, data_type in columns.items()
    ]

    # Executing the table creation SQL statement using the provided table name and columns names
    self._execute(
        f"""
            CREATE TABLE IF NOT EXISTS {table_name}
            ({", ".join(columns_with_types)});
        """
    )
delete(table_name: str, criteria: dict[str, str]) -> None

summary : This method deletes data from the database using the 'DELETE' statement.

Parameters:

Name Type Description Default
table_name str

description : Table name to perform the statement on.

 required
criteria dict[str, str]

description : The criteria to use as a filter on the DELETE statement, passed as a dict => {keys(criteria) : values(values)}

 required
Source code in src/database/db_manger.py 
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
def delete(self, table_name: str, criteria: dict[str, str]) -> None:
    """_summary_ : This method deletes data from the database using the 'DELETE' statement.

    Parameters
    ----------
    table_name : str
        _description_ : Table name to perform the statement on.
    criteria : dict[str, str]
        _description_ : The criteria to use as a filter on the DELETE statement, passed as a dict => {keys(criteria) : values(values)}
    """
    # Creating placeholders for the provided criteria
    placeholders = [
        f"{column} = ?" for column in criteria
    ]  #! add criteria.keys() if didn't work
    # Joining the created placeholders with AND operator
    delete_criteria = " AND ".join(placeholders)
    # Executing the DELETE statement
    self._execute(
        f"""
        DELETE FROM {table_name}
        WHERE {delete_criteria}
        """,
        tuple(criteria.values()),
    )
select(table_name: str, criteria: dict[str, str] = None, order_by: str = None) -> Cursor

summary: : This method selects data from the database using the 'SELECT' SQL statement and returns back data.

Parameters:

Name Type Description Default
table_name str

description : Table name to perform the statement on.

 required
criteria dict[str, str]

description, by default None : The criteria to use as a filter on the SELECT statement, passed as a dict => {keys(criteria) : values(values)}

 None
order_by str

description, by default None : A column name to sort the returned query by, will map to the ORDER BY {order_by}

 None

Returns:

Type Description
Cursor

description : A Cursor object containing the data back from the database of any.
Source code in src/database/db_manger.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
def select(
    self, table_name: str, criteria: dict[str, str] = None, order_by: str = None
) -> Cursor:
    """_summary_: : This method selects data from the database using the 'SELECT' SQL statement and returns back  data.

    Parameters
    ----------
    table_name : str
        _description_ : Table name to perform the statement on.
    criteria : dict[str, str], optional
        _description_, by default None : The criteria to use as a filter on the SELECT statement, passed as a dict => {keys(criteria) : values(values)}
    order_by : str, optional
        _description_, by default None : A column name to sort the returned query by, will map to the ORDER BY {order_by}

    Returns
    -------
    Cursor
        _description_ : A Cursor object containing the data back from the database of any.
    """
    # Creating an empty dict if no criteria was provided
    criteria = criteria or {}

    # Creating the query 'SELECT' statement
    query = f"SELECT * FROM {table_name}"

    # If criteria was provided parse it and add placeholders for it
    if criteria:
        # Creating placeholders fro the provided criteria
        placeholders = [f"{column} = ?" for column in criteria.keys()]
        # Joining each statement with 'AND'
        select_criteria = " AND ".join(placeholders)
        # Adding the 'select_criteria' to the query after the 'WHERE' SQL statement
        query += f" WHERE {select_criteria}"

    # If order_by was provided adds in a 'ORDER BY' SQL statement to sort the data
    if order_by:
        # Adding the 'ORDER BY' statement to the query followed by the column name to sort by
        query += f" ORDER BY {order_by}"

    # Executing the query and passing the criteria values to the cursor and returning back the results
    return self._execute(query, tuple(criteria.values()))
update(table_name: str, criteria: dict[str, str], data: dict[str, str]) -> None

summary : This method updates data in the database using the 'UPDATE' statement.

Parameters:

Name Type Description Default
table_name str

description : Table name to perform the statement on.

 required
criteria dict[str, str]

description : The criteria to use as a filter on the UPDATE statement, passed as a dict => {keys(criteria) : values(values)}

 required
data dict[str, str]

description : The actual data to update

 required
Source code in src/database/db_manger.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
def update(
    self, table_name: str, criteria: dict[str, str], data: dict[str, str]
) -> None:
    """_summary_ : This method updates data in the database using the 'UPDATE' statement.

    Parameters
    ----------
    table_name : str
        _description_ : Table name to perform the statement on.
    criteria : dict[str, str]
        _description_ : The criteria to use as a filter on the UPDATE statement, passed as a dict => {keys(criteria) : values(values)}
    data : dict[str, str]
        _description_ : The actual data to update
    """
    # Creating placeholders for the update criteria
    placeholders = [f"{column} = ?" for column in criteria]
    # Joining the placeholders with the AND operator
    update_criteria = " AND ".join(placeholders)
    # Creating placeholders for the update data
    data_placeholders = ", ".join(f"{key} = ?" for key in data)
    # Creating two tuples holding the values to update and the new data to update them with
    values = tuple(data.values()) + tuple(criteria.values())
    # Executing the UPDATE statement
    self._execute(
        f"""
        UPDATE {table_name}
        SET {data_placeholders}
        WHERE {update_criteria};
        """,
        values,
    )