Skip to content

sqlite_managers

Context manager utilities for interacting with SQLite databases, using the stdlib sqlite3 library.

The SQLiteConnManager class facilitates clean & safe transactions to the database by trying operations before committing.

SQLiteConnManager

Handle interactions with a SQLite database.

Uses built-in functions to query a database, execute SQL statements, and gracefully open/close the DB using context managers.

Parameters:

Name Type Description Default
path(str|Path)

A path to a SQLite database file to work on.

 required

Usage: Provide a path string to the SQLite database: 

1
sqlite_connection = SQLiteConnManager(path="/path/to/db.sqlite")

Call sqlite3 functions, i.e. .get_tables(): 

1
tables = sqlite_conn.get_tables()

Source code in src/red_utils/std/context_managers/database_managers/sqlite_managers.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
class SQLiteConnManager:
    """Handle interactions with a SQLite database.

    Uses built-in functions to query a database, execute SQL statements,
    and gracefully open/close the DB using context managers.

    Params:
        path(str|Path): A path to a SQLite database file to work on.

    Usage:
    Provide a path string to the SQLite database:
    ``` py linenums="1"
    sqlite_connection = SQLiteConnManager(path="/path/to/db.sqlite")
    ```

    Call sqlite3 functions, i.e. `.get_tables()`:
    ``` py linenums="1"
    tables = sqlite_conn.get_tables()
    ```
    """

    def __init__(self, path: Path):
        ## Initialize SQLite connection manager.
        if isinstance(path, str):
            path: Path = Path(path)

        self.path = path

    def __enter__(self):
        ## Executed automatically when class is used as a context handler, like `with SQLiteConnManager() as conn: ...`
        if not self.path.exists():
            print(FileNotFoundError(f"Database does not exist at path: {self.path}."))
            with sqlite3.connect(self.path):
                pass

        try:
            self.connection: sqlite3.Connection = sqlite3.connect(self.path)
            self.connection.row_factory = sqlite3.Row
            self.cursor: sqlite3.Cursor = self.connection.cursor()

            ## Return self, a configured SQLite client
            return self
        except FileNotFoundError as fnf:
            raise FileNotFoundError(
                f"Database not found at path: {self.path}. Details: {fnf}"
            )
        except PermissionError as perm:
            raise PermissionError(
                f"Unable to open database at path: {self.path}. Details: {perm}"
            )
        except sqlite3.Error as sqlite_exc:
            raise sqlite3.Error(f"SQLite3 error encountered. Details: {sqlite_exc}")
        except Exception as exc:
            raise Exception(
                f"Unhandled exception connecting to database. Details: ({exc.__class__}) {exc}"
            )

    def __exit__(self, exc_type, exc_val, exc_traceback):
        ## Executed automatically when `with SQLiteConnManager()` exits. Executes on success or failure.
        self.connection.close()

    def get_cols(self, table: str = None) -> list[str]:
        """Return list of column names from a given table.

        Params:
            table (str): Name of the table in the SQLite database

        Returns
        -------
            (list[str]): List of column names found in table
        """
        cols: list[str] = []
        stmt: str = f"SELECT * FROM {table}"

        try:
            with SQLiteConnManager(self.path) as conn:
                cursor = conn.cursor
                res = cursor.execute(stmt).fetchone()

                for col in res.keys():
                    cols.append(col)

                return cols

        except Exception as exc:
            raise Exception(f"Unhandled exception executing SQL. Details: {exc}")

    def get_tables(self) -> list[str]:
        """Get all table names from a SQLite databse.

        Returns
        -------
            (list[str]): List of table names found in SQLite database.
        """
        get_tbls_stmt: str = "SELECT name FROM sqlite_master WHERE type='table';"
        tables: list[str] = []

        try:
            with SQLiteConnManager(self.path) as conn:
                cursor = conn.cursor
                res = cursor.execute(get_tbls_stmt).fetchall()

                for row in res:
                    tables.append(row["name"])

                return tables

        except Exception as exc:
            raise Exception(f"Unhandled exception getting tables. Details: {exc}")

    def run_sqlite_stmt(self, stmt: str = None) -> list[sqlite3.Row]:
        """Execute a SQL statement.

        Params:
            stmt (str): The SQL statement to execute against a SQLite database

        Returns
        -------
            (list[sqlite3.Row]): The results from executing the query
        """
        assert stmt, "Must pass a SQL statement"
        assert isinstance(stmt, str), "Statement must be a Python str"

        try:
            with SQLiteConnManager(self.path) as conn:
                cursor = conn.cursor
                res = cursor.execute(stmt).fetchall()

            return res
        except Exception as exc:
            raise Exception(
                f"Unhandled exception running SQL statement. Details: {exc}"
            )

get_cols(table=None)

Return list of column names from a given table.

Parameters:

Name Type Description Default
table str

Name of the table in the SQLite database

 None
Returns
(list[str]): List of column names found in table
Source code in src/red_utils/std/context_managers/database_managers/sqlite_managers.py 
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
def get_cols(self, table: str = None) -> list[str]:
    """Return list of column names from a given table.

    Params:
        table (str): Name of the table in the SQLite database

    Returns
    -------
        (list[str]): List of column names found in table
    """
    cols: list[str] = []
    stmt: str = f"SELECT * FROM {table}"

    try:
        with SQLiteConnManager(self.path) as conn:
            cursor = conn.cursor
            res = cursor.execute(stmt).fetchone()

            for col in res.keys():
                cols.append(col)

            return cols

    except Exception as exc:
        raise Exception(f"Unhandled exception executing SQL. Details: {exc}")

get_tables()

Get all table names from a SQLite databse.

Returns
(list[str]): List of table names found in SQLite database.
Source code in src/red_utils/std/context_managers/database_managers/sqlite_managers.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
def get_tables(self) -> list[str]:
    """Get all table names from a SQLite databse.

    Returns
    -------
        (list[str]): List of table names found in SQLite database.
    """
    get_tbls_stmt: str = "SELECT name FROM sqlite_master WHERE type='table';"
    tables: list[str] = []

    try:
        with SQLiteConnManager(self.path) as conn:
            cursor = conn.cursor
            res = cursor.execute(get_tbls_stmt).fetchall()

            for row in res:
                tables.append(row["name"])

            return tables

    except Exception as exc:
        raise Exception(f"Unhandled exception getting tables. Details: {exc}")

run_sqlite_stmt(stmt=None)

Execute a SQL statement.

Parameters:

Name Type Description Default
stmt str

The SQL statement to execute against a SQLite database

 None
Returns
(list[sqlite3.Row]): The results from executing the query
Source code in src/red_utils/std/context_managers/database_managers/sqlite_managers.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
def run_sqlite_stmt(self, stmt: str = None) -> list[sqlite3.Row]:
    """Execute a SQL statement.

    Params:
        stmt (str): The SQL statement to execute against a SQLite database

    Returns
    -------
        (list[sqlite3.Row]): The results from executing the query
    """
    assert stmt, "Must pass a SQL statement"
    assert isinstance(stmt, str), "Statement must be a Python str"

    try:
        with SQLiteConnManager(self.path) as conn:
            cursor = conn.cursor
            res = cursor.execute(stmt).fetchall()

        return res
    except Exception as exc:
        raise Exception(
            f"Unhandled exception running SQL statement. Details: {exc}"
        )