Skip to content

Database Reader & Creation

Overview

The database_reader package handles connections with the databases file.

It contains two class:

  • SingleDatabase: Given a dictionary of pandas dataframe, it creates the database folder with the database sqlite file.
  • MultipleDatabases: Given the path where the databases are stored, it automatically creates the connection with the stored file.

If the data is already stored in database sqlite files following this pattern "db_save_path/db_id/db_id.sqlite" you can create a connection between the tool and the files with MultipleDatabase class:

from qatch.database_reader import MultipleDatabases

# The path to multiple databases
db_save_path = 'test_db'
databases = MultipleDatabases(db_save_path)

Instead, if you want to specify different data you can use the SingleDatabase class to create the sqlite databases in "db_save_path/db_id/db_id.sqlite"

Assume the PKs have all different names. No two tables with the same PK name.

import pandas as pd

from qatch.database_reader import SingleDatabase

# Create dummy table
data = {
    "year": [1896, 1900, 1904, 2004, 2008, 2012],
    "city": ["athens", "paris", "st. louis", "athens", "beijing", "london"]
}
table = pd.DataFrame.from_dict(data)

# define the tables in the database (<table_name> : <table>)
db_tables = {'olympic_games': table}

# define where to store the sqlite database
db_save_path = 'test_db'

# define the name of the database
db_id = 'olympic'

# define the PK
# Assume the PKs have all different names. Two tables cannot have same PK name.
table2primary_key = {'olympic_games': 'id'}

# create database connection
db = SingleDatabase(db_path=db_save_path, db_name=db_id, tables=db_tables, table2primary_key=table2primary_key)

MultipleDatabases

Manages multiple SQLite databases, allowing dynamic creation and access to individual databases.

Attributes:

Name Type Description
db_path str

The base path where the database files are stored.
db_ids2db dict

A dictionary where the key is the database name, and the value is the SingleDatabase object.
_max_db_in_memory int

The maximum number of databases to keep in memory. Default is 15.
Source code in qatch/database_reader/multiple_databases.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
 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
class MultipleDatabases:
    """Manages multiple SQLite databases, allowing dynamic creation and access to individual databases.

    Attributes:
        db_path (str): The base path where the database files are stored.
        db_ids2db (dict): A dictionary where the key is the database name, and the value is the SingleDatabase object.
        _max_db_in_memory (int): The maximum number of databases to keep in memory. Default is 15.
    """

    def __init__(self, db_path: str, _max_db_in_memory=15):
        self.db_path = db_path
        self.db_ids2db: dict[str, SingleDatabase] = dict()
        self._max_db_in_memory = _max_db_in_memory

    def get_db_names(self) -> list[str]:
        """Gets the name of the database file from the path.

        Returns:
            list[str]: A list of database file names.
        """
        # return x only if it is a directory
        return [x for x in os.listdir(self.db_path) if os.path.isdir(os.path.join(self.db_path, x))]

    def __contains__(self, other: str) -> bool:
        # Checks if a database with the given name exists in the managed databases.
        return any(other in db_id for db_id in self.db_ids2db)

    def __getitem__(self, key: str) -> SingleDatabase:
        # Allows accessing a specific database by its name
        # if the key does not exist, open the db
        if key not in self.db_ids2db:
            self.open_db(key)
        return self.db_ids2db[key]

    def open_db(self, db_id: str):
        """Opens a database with the given name and stores it in memory.

        Args:
            db_id (str): The name of the database to open.
        """
        if db_id not in self.db_ids2db:
            if len(self.db_ids2db) >= self._max_db_in_memory:
                # FIFO strategy if max reached
                self.db_ids2db.popitem()
            self.db_ids2db[db_id] = SingleDatabase(self.db_path, db_name=db_id)

    def get_table(self, db_id: str, tbl_name: str) -> pd.DataFrame:
        """Retrieves a specified table from the database as a Pandas DataFrame.

        Args:
            db_id (str): The name of the database.
            tbl_name (str): The name of the table to retrieve from the database.

        Returns:
            pd.DataFrame: A Pandas DataFrame representing the specified table from the database.
        """
        return self[db_id].get_table_given(tbl_name)

    def get_schema(self, db_id: str, tbl_name: str) -> pd.DataFrame:
        """Retrieves the schema of a specified table from the database.

        Args:
            db_id (str): The name of the database.
            tbl_name (str): The name of the table to retrieve the schema from.

        Returns:
            pd.DataFrame: A Pandas DataFrame representing the schema of the specified table.
        """
        return self[db_id].get_schema_given(tbl_name)

    def get_all_table_schema_given(self, db_id: str) -> dict[str, pd.DataFrame]:
        """Retrieves all the table schema from the database.

        Args:
            db_id (str): The name of the database.

        Returns:
            pd.DataFrame: A Pandas DataFrame representing the schema of the specified table.
        """

        return self[db_id].table_schemas

    def run_multiple_queries(self, db_id: str, queries: list) -> list[list]:
        """Executes multiple queries on the specified database and returns the results.

          Args:
              db_id (str): The name of the database to execute the query on.
              queries (list): The list of SQL queries to be executed on the database.

          Returns:
              list[list]: A list containing the query results.
        """
        db = self[db_id]
        queries_result = map(db.run_query, queries)
        queries_result = list(map(lambda x: [list(item) for item in x], queries_result))
        return queries_result

    def run_query(self, db_id: str, query: str) -> list | None:
        """Executes an SQL query on the specified database and returns the results.

        Args:
            db_id (str): The name of the database to execute the query on.
            query (str): The SQL query to be executed on the database.

        Returns:
            list | None: A list containing the query results.
        """
        ans_query = self[db_id].run_query(query)
        return [list(x) for x in ans_query]

    def close(self):
        """Closes all the opened databases and clears the memory."""
        [db.close_connection() for db in self.db_ids2db.values()]
        self.db_ids2db = dict()

close()

Closes all the opened databases and clears the memory.

Source code in qatch/database_reader/multiple_databases.py 
118
119
120
121
def close(self):
    """Closes all the opened databases and clears the memory."""
    [db.close_connection() for db in self.db_ids2db.values()]
    self.db_ids2db = dict()

get_all_table_schema_given(db_id)

Retrieves all the table schema from the database.

Parameters:

Name Type Description Default
db_id str

The name of the database.

 required

Returns:

Type Description
dict[str, DataFrame]

pd.DataFrame: A Pandas DataFrame representing the schema of the specified table.
Source code in qatch/database_reader/multiple_databases.py 
78
79
80
81
82
83
84
85
86
87
88
def get_all_table_schema_given(self, db_id: str) -> dict[str, pd.DataFrame]:
    """Retrieves all the table schema from the database.

    Args:
        db_id (str): The name of the database.

    Returns:
        pd.DataFrame: A Pandas DataFrame representing the schema of the specified table.
    """

    return self[db_id].table_schemas

get_db_names()

Gets the name of the database file from the path.

Returns:

Type Description
list[str]

list[str]: A list of database file names.
Source code in qatch/database_reader/multiple_databases.py 
22
23
24
25
26
27
28
29
def get_db_names(self) -> list[str]:
    """Gets the name of the database file from the path.

    Returns:
        list[str]: A list of database file names.
    """
    # return x only if it is a directory
    return [x for x in os.listdir(self.db_path) if os.path.isdir(os.path.join(self.db_path, x))]

get_schema(db_id, tbl_name)

Retrieves the schema of a specified table from the database.

Parameters:

Name Type Description Default
db_id str

The name of the database.

 required
tbl_name str

The name of the table to retrieve the schema from.

 required

Returns:

Type Description
DataFrame

pd.DataFrame: A Pandas DataFrame representing the schema of the specified table.
Source code in qatch/database_reader/multiple_databases.py 
66
67
68
69
70
71
72
73
74
75
76
def get_schema(self, db_id: str, tbl_name: str) -> pd.DataFrame:
    """Retrieves the schema of a specified table from the database.

    Args:
        db_id (str): The name of the database.
        tbl_name (str): The name of the table to retrieve the schema from.

    Returns:
        pd.DataFrame: A Pandas DataFrame representing the schema of the specified table.
    """
    return self[db_id].get_schema_given(tbl_name)

get_table(db_id, tbl_name)

Retrieves a specified table from the database as a Pandas DataFrame.

Parameters:

Name Type Description Default
db_id str

The name of the database.

 required
tbl_name str

The name of the table to retrieve from the database.

 required

Returns:

Type Description
DataFrame

pd.DataFrame: A Pandas DataFrame representing the specified table from the database.
Source code in qatch/database_reader/multiple_databases.py 
54
55
56
57
58
59
60
61
62
63
64
def get_table(self, db_id: str, tbl_name: str) -> pd.DataFrame:
    """Retrieves a specified table from the database as a Pandas DataFrame.

    Args:
        db_id (str): The name of the database.
        tbl_name (str): The name of the table to retrieve from the database.

    Returns:
        pd.DataFrame: A Pandas DataFrame representing the specified table from the database.
    """
    return self[db_id].get_table_given(tbl_name)

open_db(db_id)

Opens a database with the given name and stores it in memory.

Parameters:

Name Type Description Default
db_id str

The name of the database to open.

 required
Source code in qatch/database_reader/multiple_databases.py 
42
43
44
45
46
47
48
49
50
51
52
def open_db(self, db_id: str):
    """Opens a database with the given name and stores it in memory.

    Args:
        db_id (str): The name of the database to open.
    """
    if db_id not in self.db_ids2db:
        if len(self.db_ids2db) >= self._max_db_in_memory:
            # FIFO strategy if max reached
            self.db_ids2db.popitem()
        self.db_ids2db[db_id] = SingleDatabase(self.db_path, db_name=db_id)

run_multiple_queries(db_id, queries)

Executes multiple queries on the specified database and returns the results.

Parameters:

Name Type Description Default
db_id str

The name of the database to execute the query on.

 required
queries list

The list of SQL queries to be executed on the database.

 required

Returns:

Type Description
list[list]

list[list]: A list containing the query results.
Source code in qatch/database_reader/multiple_databases.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def run_multiple_queries(self, db_id: str, queries: list) -> list[list]:
    """Executes multiple queries on the specified database and returns the results.

      Args:
          db_id (str): The name of the database to execute the query on.
          queries (list): The list of SQL queries to be executed on the database.

      Returns:
          list[list]: A list containing the query results.
    """
    db = self[db_id]
    queries_result = map(db.run_query, queries)
    queries_result = list(map(lambda x: [list(item) for item in x], queries_result))
    return queries_result

run_query(db_id, query)

Executes an SQL query on the specified database and returns the results.

Parameters:

Name Type Description Default
db_id str

The name of the database to execute the query on.

 required
query str

The SQL query to be executed on the database.

 required

Returns:

Type Description
list | None

list | None: A list containing the query results.
Source code in qatch/database_reader/multiple_databases.py 
105
106
107
108
109
110
111
112
113
114
115
116
def run_query(self, db_id: str, query: str) -> list | None:
    """Executes an SQL query on the specified database and returns the results.

    Args:
        db_id (str): The name of the database to execute the query on.
        query (str): The SQL query to be executed on the database.

    Returns:
        list | None: A list containing the query results.
    """
    ans_query = self[db_id].run_query(query)
    return [list(x) for x in ans_query]

SingleDatabase

Provides a simplified interface for interacting with SQLite databases in Python.

Note: If the database already exist, do not replace the tables

Attributes:

Name Type Description
conn Connection

A connection object representing the SQLite database.
cursor Cursor

A cursor object used to execute SQL commands and retrieve results.
db_name str

The name of the database.
table_schemas Dict[str, DataFrame]

Table name as key, table schema as value. Table schema is a Pandas DataFrame with columns: cid, name, type, notnull, dflt_value, pk.
table_names str

The names of the tables in the database.
db_path str

Path to the folder which contains the SQLite file.
db_path_sqlite str

Path to the SQLite file.
Source code in qatch/database_reader/single_database.py 
  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
 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
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
class SingleDatabase:
    """Provides a simplified interface for interacting with SQLite databases in Python.

    Note: If the database already exist, do not replace the tables

    Attributes:
        conn (sqlite3.Connection): A connection object representing the SQLite database.
        cursor (sqlite3.Cursor): A cursor object used to execute SQL commands and retrieve results.
        db_name (str): The name of the database.
        table_schemas (Dict[str, pd.DataFrame]): Table name as key, table schema as value.
            Table schema is a Pandas DataFrame with columns: cid, name, type, notnull, dflt_value, pk.
        table_names (str): The names of the tables in the database.
        db_path (str): Path to the folder which contains the SQLite file.
        db_path_sqlite (str): Path to the SQLite file.
    """

    def __init__(self, db_path: str, db_name: str,
                 tables: dict[str, pd.DataFrame] | None = None,
                 table2primary_key: dict[str, str] | None = None):
        # Get the valid directory for the database
        self.db_path = self._ensure_valid_directory(db_path, db_name)
        # Store the name of the database
        self.db_name = db_name
        # Create the final path for the SQLite database file
        self.db_path_sqlite = os.path.join(self.db_path, f'{db_name}.sqlite')
        # Get the existing table names in the database
        self.table_names = self._get_existing_table_names()
        # If there are no existing tables
        if not self.table_names:
            # If no tables dictionary is provided
            if not tables:
                raise ValueError(f"No tables provided and no database found at "
                                 f"{self.db_path_sqlite}")
            # Otherwise, set the tables in the database
            self.set_tables_in_db(tables, table2primary_key)
            # Get the table names from the provided dictionary
            self.table_names = list(tables.keys())
            # Logs that the tables are stored in the database
            logging.info(f"Tables stored in {self.db_path_sqlite}")

        # Store the dictionary mapping table names to their primary keys
        self.table2primary_key = table2primary_key
        # Get the schemas of the tables
        self.table_schemas = self._get_table_schemas()

    @contextlib.contextmanager
    def connect_cursor(self):
        with sqlite3.connect(self.db_path_sqlite) as conn:
            conn.text_factory = lambda b: b.decode(errors='ignore')
            yield conn.cursor()

    @staticmethod
    def _ensure_valid_directory(db_path: str, db_name: str) -> str:
        """Ensures the db_path/db_name exist, create if necessary"""
        if db_path in ('', '.', './'):
            db_path = os.getcwd()

        full_path = os.path.join(db_path, db_name)
        os.makedirs(full_path, exist_ok=True)
        return full_path

    def _get_table_schemas(self) -> dict[str, pd.DataFrame]:
        """Retrieve schemas for all the tables in the SQLite database"""
        with self.connect_cursor() as cursor:
            table_names = self._get_existing_table_names()
            return {tbl_name: pd.read_sql_query(f'PRAGMA table_info("{tbl_name}")', cursor.connection)
                    for tbl_name in self.table_names}

    def _get_existing_table_names(self) -> list[str]:
        """Retrieve the table name in the SQLite database"""
        with self.connect_cursor() as cursor:
            cursor.execute('SELECT name from sqlite_master where type= "table"')
            return [tbl[0] for tbl in cursor.fetchall()]

    def set_tables_in_db(self, tables: dict[str, pd.DataFrame] | None,
                         table2primary_key: dict[str, str] | None):
        """
        Sets the tables in the SQLite database represented by the given connection object.

        This method takes a dictionary of tables in which keys are table names and values are Pandas DataFrames
        representing the tables, and sets these tables in the SQLite database represented by the `conn` object.

        The optional `table2primary_key` argument can be used to set primary keys for some or all tables.
        If not provided, all tables are created without primary keys.
        If the table contains an attribute with the same name of a primary key, a foreign key relationship is created.

        Note:
            - If a table is named as 'table', the method will replace its name with 'my_table'.
            - Assume the PKs have all different names. two tables must have different PK names.

        Args:
            tables (Optional[Dict[str, pd.DataFrame]]): A dictionary of tables to set in the SQLite database.
                Keys are table names and values are corresponding Pandas DataFrames.

            table2primary_key (Optional[Dict[str, str]]): A dictionary mapping table names to primary keys.
                For example, if you want to set the primary key of table `A` to be `Key_1`, you should pass
                `table2primary_key={'A': 'Key_1'}`. Default is None.
        """
        with self.connect_cursor() as cursor:
            conn = cursor.connection
            for name, table in tables.items():
                if name == 'table':
                    name = 'my_table'
                if not table2primary_key:
                    table.to_sql(name, conn, if_exists='replace', index=False)
                else:
                    create_table_string = SingleDatabase._create_table_in_db(name, table, table2primary_key)
                    conn.cursor().execute(create_table_string)
                    table.to_sql(name, conn, if_exists='append', index=False)

    @staticmethod
    def _create_table_in_db(name, table, table2primary_key):
        """
        Returns a SQLite CREATE TABLE command as a string, constructed based on the given table name, DataFrame,
        and primary_key2table dict.

        This method first converts pandas DataFrame dtypes to SQLite data types.
        Then, a SQLite CREATE TABLE command is built step by step.
        The command includes creating simple columns, adding primary keys, and creating foreign key relationships.
        Finally, the CREATE TABLE command is returned as a string.

        Args:
             name (str): The name of the table to be created in SQLite database.
             table (pd.DataFrame): A pandas DataFrame holding the data and structure of the SQL table.
                                  The dtype of each column translated to SQLite data types.
             table2primary_key (dict): A dictionary where the keys are the names of columns of the table,
                                        and the values are the names of the tables they are primary keys to.

        Example:
            >>> import pandas as pd
            >>> from qatch.database_reader import SingleDatabase

            >>> name = 'sample_table'
            >>> table = pd.DataFrame({
            >>>    'id': [1, 2, 3],
            >>>    'name': ['Alice', 'Bob', 'Charlie'],
            >>>})
            >>>primary_key2table = {
            >>>    'sample_table_id': 'sample_table'
            >>>}
            >>>print(SingleDatabase._create_table_in_db(name,table,primary_key2table))
            >>># Outputs: CREATE TABLE `sample_table`( "id" INTEGER, "name" TEXT, PRIMARY KEY ("id") );
        """

        def convert_pandas_dtype_to_sqlite_type(type_):
            if 'int' in type_:
                return 'INTEGER'
            if 'float' in type_:
                return 'REAL'
            if 'object' in type_ or 'date' in type_:
                return 'TEXT'

        primary_key2table = {tbl_PK: tbl_name for tbl_name, tbl_PK in
                             table2primary_key.items()} if table2primary_key else None
        column2type = {k: convert_pandas_dtype_to_sqlite_type(str(table.dtypes[k]))
                       for k in table.dtypes.index}
        create_table = [f'CREATE TABLE `{name}`(']
        # add simple col
        # "Round" real,
        [create_table.append(f'"{col}" {column2type[col]},') for col in table.columns]
        # Add primary key and foreign key
        for col in table.columns:
            if col in primary_key2table:
                if name == primary_key2table[col]:
                    # PRIMARY KEY ("Round"),
                    create_table.append(f'PRIMARY KEY ("{col}"),')
                else:
                    # FOREIGN KEY (`Winning_Aircraft`) REFERENCES `aircraft`(`Aircraft_ID`),
                    create_table.append(f'FOREIGN KEY (`{col}`) REFERENCES `{primary_key2table[col]}`(`{col}`),')
        # remove last comma
        create_table[-1] = create_table[-1][:-1]
        # add closing statement
        create_table.append(');')
        return " ".join(create_table)

    def get_table_given(self, table_name: str) -> pd.DataFrame:
        """
        Retrieves a specified table from the database as a Pandas DataFrame.

        Args:
            table_name (str): The name of the table to retrieve from the database.

        Returns:
            pd.DataFrame: A Pandas DataFrame representing the specified table from the database.
        """
        with self.connect_cursor() as cursor:
            conn = cursor.connection
            return pd.read_sql_query(f"SELECT * FROM {table_name}", conn)

    def get_schema_given(self, table_name: str) -> pd.DataFrame:
        """
        Given the table name, returns the schema of the table.
        Table schema is a Pandas DataFrame with columns: cid, name, type, notnull, dflt_value, pk.

        Args:
            table_name (str): The name of the table to retrieve the schema from the database.

        Returns:
            pd.DataFrame: A Pandas DataFrame representing the schema of the specified table from the database.
        """
        return self.table_schemas[table_name]

    def run_query(self, query: str) -> list[list]:
        """
        Run a query on the database and return the result.

        Args:
            query (str): The SQL query to be executed on the database.
        Returns:
            list[list]: A list of lists representing the result of the SQL query.
        """
        with self.connect_cursor() as cursor:
            try:
                cursor.execute(query)
                output = cursor.fetchall()
                if not output:
                    logging.warning(f'No query result for this query: {query}')
                return output
            except sqlite3.OperationalError as e:
                logging.error(f"Error while executing query: {query}")
                logging.error(e)
                raise

get_schema_given(table_name)

Given the table name, returns the schema of the table. Table schema is a Pandas DataFrame with columns: cid, name, type, notnull, dflt_value, pk.

Parameters:

Name Type Description Default
table_name str

The name of the table to retrieve the schema from the database.

 required

Returns:

Type Description
DataFrame

pd.DataFrame: A Pandas DataFrame representing the schema of the specified table from the database.
Source code in qatch/database_reader/single_database.py 
198
199
200
201
202
203
204
205
206
207
208
209
def get_schema_given(self, table_name: str) -> pd.DataFrame:
    """
    Given the table name, returns the schema of the table.
    Table schema is a Pandas DataFrame with columns: cid, name, type, notnull, dflt_value, pk.

    Args:
        table_name (str): The name of the table to retrieve the schema from the database.

    Returns:
        pd.DataFrame: A Pandas DataFrame representing the schema of the specified table from the database.
    """
    return self.table_schemas[table_name]

get_table_given(table_name)

Retrieves a specified table from the database as a Pandas DataFrame.

Parameters:

Name Type Description Default
table_name str

The name of the table to retrieve from the database.

 required

Returns:

Type Description
DataFrame

pd.DataFrame: A Pandas DataFrame representing the specified table from the database.
Source code in qatch/database_reader/single_database.py 
184
185
186
187
188
189
190
191
192
193
194
195
196
def get_table_given(self, table_name: str) -> pd.DataFrame:
    """
    Retrieves a specified table from the database as a Pandas DataFrame.

    Args:
        table_name (str): The name of the table to retrieve from the database.

    Returns:
        pd.DataFrame: A Pandas DataFrame representing the specified table from the database.
    """
    with self.connect_cursor() as cursor:
        conn = cursor.connection
        return pd.read_sql_query(f"SELECT * FROM {table_name}", conn)

run_query(query)

Run a query on the database and return the result.

Parameters:

Name Type Description Default
query str

The SQL query to be executed on the database.

 required

Returns: list[list]: A list of lists representing the result of the SQL query.

Source code in qatch/database_reader/single_database.py 
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
def run_query(self, query: str) -> list[list]:
    """
    Run a query on the database and return the result.

    Args:
        query (str): The SQL query to be executed on the database.
    Returns:
        list[list]: A list of lists representing the result of the SQL query.
    """
    with self.connect_cursor() as cursor:
        try:
            cursor.execute(query)
            output = cursor.fetchall()
            if not output:
                logging.warning(f'No query result for this query: {query}')
            return output
        except sqlite3.OperationalError as e:
            logging.error(f"Error while executing query: {query}")
            logging.error(e)
            raise

set_tables_in_db(tables, table2primary_key)

Sets the tables in the SQLite database represented by the given connection object.

This method takes a dictionary of tables in which keys are table names and values are Pandas DataFrames representing the tables, and sets these tables in the SQLite database represented by the conn object.

The optional table2primary_key argument can be used to set primary keys for some or all tables. If not provided, all tables are created without primary keys. If the table contains an attribute with the same name of a primary key, a foreign key relationship is created.

Note
  • If a table is named as 'table', the method will replace its name with 'my_table'.
  • Assume the PKs have all different names. two tables must have different PK names.

Parameters:

Name Type Description Default
tables Optional[Dict[str, DataFrame]]

A dictionary of tables to set in the SQLite database. Keys are table names and values are corresponding Pandas DataFrames.

 required
table2primary_key Optional[Dict[str, str]]

A dictionary mapping table names to primary keys. For example, if you want to set the primary key of table A to be Key_1, you should pass table2primary_key={'A': 'Key_1'}. Default is None.

 required
Source code in qatch/database_reader/single_database.py 
 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
def set_tables_in_db(self, tables: dict[str, pd.DataFrame] | None,
                     table2primary_key: dict[str, str] | None):
    """
    Sets the tables in the SQLite database represented by the given connection object.

    This method takes a dictionary of tables in which keys are table names and values are Pandas DataFrames
    representing the tables, and sets these tables in the SQLite database represented by the `conn` object.

    The optional `table2primary_key` argument can be used to set primary keys for some or all tables.
    If not provided, all tables are created without primary keys.
    If the table contains an attribute with the same name of a primary key, a foreign key relationship is created.

    Note:
        - If a table is named as 'table', the method will replace its name with 'my_table'.
        - Assume the PKs have all different names. two tables must have different PK names.

    Args:
        tables (Optional[Dict[str, pd.DataFrame]]): A dictionary of tables to set in the SQLite database.
            Keys are table names and values are corresponding Pandas DataFrames.

        table2primary_key (Optional[Dict[str, str]]): A dictionary mapping table names to primary keys.
            For example, if you want to set the primary key of table `A` to be `Key_1`, you should pass
            `table2primary_key={'A': 'Key_1'}`. Default is None.
    """
    with self.connect_cursor() as cursor:
        conn = cursor.connection
        for name, table in tables.items():
            if name == 'table':
                name = 'my_table'
            if not table2primary_key:
                table.to_sql(name, conn, if_exists='replace', index=False)
            else:
                create_table_string = SingleDatabase._create_table_in_db(name, table, table2primary_key)
                conn.cursor().execute(create_table_string)
                table.to_sql(name, conn, if_exists='append', index=False)