Skip to content

Sqlite

blackline.adapters.sqlite.sqlite

SQLiteAdapter

Bases: SQLAdapter

Source code in BAR /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/blackline/adapters/sqlite/sqlite.py 
 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
class SQLiteAdapter(SQLAdapter):
    config_model = SQLiteDataStore
    dialect = "sqlite"

    def __init__(self, config: SQLiteDataStore.Config, *args, **kwargs):
        super().__init__(config=config, *args, **kwargs)

    def connection(self) -> Connection:
        return connect(**self.config.connection.dict())

    def execute(self, sql: str, values: Optional[dict[str, Any]] = None) -> Any:
        with self.connection() as conn:
            return conn.execute(sql, values or ())

    def fetchall(self, results: Cursor) -> list[tuple]:
        return results.fetchall()

    def update_template(self) -> str:
        return "UPDATE {{ table }}"

    def set_template(self) -> str:
        return "SET"

    def redact_template(self) -> str:
        return "{{ name }} = null"

    def replace_template(self) -> str:
        return "{{ name }} = :{{ value }}"

    def mask_template(self) -> str:
        """Mask template for SQLite.

        SQLite does not support regex, so we have to use a nested replace
        function to mask digits in a column. We are limited to digits because
        if we mask mask characters will will run into an OperationalError: parser
        stack overflow

        Returns:
            str: Mask template for SQLite.
        """

        replace_str = "{{ name }}"
        for c in string.digits:
            replace_str = f"REPLACE({replace_str}, '{c}', :{{{{ value }}}})"

        return f"{{{{ name }}}} = {replace_str}"

    def where_template(self) -> str:
        return "WHERE {{ datetime_column }} < :cutoff"

    def test_connection(self) -> bool:
        try:
            with self.connection() as conn:
                conn.execute("SELECT 1")
                return True
        except Exception:
            return False

    def table_exists(self, table: str) -> bool:
        """Check if a table exists.

        Args:
            table (str): Table name.

        Returns:
            bool: True if the table exists.
        """
        with self.connection() as conn:
            results = conn.execute(
                "SELECT name FROM sqlite_master WHERE type='table' AND name=:table",
                {"table": table},
            ).fetchall()
        return len(results) > 0

    def columns(self, table: str) -> list[Column]:
        """
        Return a list of columns for a given table.

        Args:
            table: Table name.

        Returns:
            A list of Column.
        """
        info = self.column_table_info(table=table)
        check_info = self.column_check_info(table=table)
        for column, check in check_info.items():
            info[column]["check"] = check
        return [Column.parse_obj(result) for result in info.values()]

    def column_table_info(
        self, table: str
    ) -> dict[str, dict[str, Union[str, bool, None]]]:
        """
        Return a dictionary of columns for a given table.

        Args:
            table: Table name.

        Returns:
            A dictionary of column info.
        """
        with self.connection() as con:
            info = con.execute(f"PRAGMA table_info({table})").fetchall()
        return {
            result[1]: {
                "name": result[1],
                "data_type": result[2],
                "nullable": not bool(result[3]),
                "default": result[4],
                "primary_key": bool(result[5]),
            }
            for result in info
        }

    def column_check_info(self, table: str) -> dict[str, str]:
        with self.connection() as con:
            schema = con.execute(
                f"select sql from sqlite_master where type='table' and name='{table}'"
            ).fetchone()
        if schema is None:
            return {}
        return {
            constraint.find(exp.Identifier).name: constraint.sql()
            for constraint in parse_one(schema[0]).find_all(exp.CheckColumnConstraint)
        }

config_model = SQLiteDataStore class-attribute instance-attribute

dialect = 'sqlite' class-attribute instance-attribute

__init__(config, *args, **kwargs)

Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/blackline/adapters/sqlite/sqlite.py 
15
16
def __init__(self, config: SQLiteDataStore.Config, *args, **kwargs):
    super().__init__(config=config, *args, **kwargs)

column_check_info(table)

Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/blackline/adapters/sqlite/sqlite.py 
126
127
128
129
130
131
132
133
134
135
136
def column_check_info(self, table: str) -> dict[str, str]:
    with self.connection() as con:
        schema = con.execute(
            f"select sql from sqlite_master where type='table' and name='{table}'"
        ).fetchone()
    if schema is None:
        return {}
    return {
        constraint.find(exp.Identifier).name: constraint.sql()
        for constraint in parse_one(schema[0]).find_all(exp.CheckColumnConstraint)
    }

column_table_info(table)

Return a dictionary of columns for a given table.

Parameters:

Name Type Description Default
table str

Table name.

 required

Returns:

Type Description
dict[str, dict[str, Union[str, bool, None]]]

A dictionary of column info.
Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/blackline/adapters/sqlite/sqlite.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
def column_table_info(
    self, table: str
) -> dict[str, dict[str, Union[str, bool, None]]]:
    """
    Return a dictionary of columns for a given table.

    Args:
        table: Table name.

    Returns:
        A dictionary of column info.
    """
    with self.connection() as con:
        info = con.execute(f"PRAGMA table_info({table})").fetchall()
    return {
        result[1]: {
            "name": result[1],
            "data_type": result[2],
            "nullable": not bool(result[3]),
            "default": result[4],
            "primary_key": bool(result[5]),
        }
        for result in info
    }

columns(table)

Return a list of columns for a given table.

Parameters:

Name Type Description Default
table str

Table name.

 required

Returns:

Type Description
list[Column]

A list of Column.
Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/blackline/adapters/sqlite/sqlite.py 
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def columns(self, table: str) -> list[Column]:
    """
    Return a list of columns for a given table.

    Args:
        table: Table name.

    Returns:
        A list of Column.
    """
    info = self.column_table_info(table=table)
    check_info = self.column_check_info(table=table)
    for column, check in check_info.items():
        info[column]["check"] = check
    return [Column.parse_obj(result) for result in info.values()]

connection()

Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/blackline/adapters/sqlite/sqlite.py 
18
19
def connection(self) -> Connection:
    return connect(**self.config.connection.dict())

execute(sql, values=None)

Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/blackline/adapters/sqlite/sqlite.py 
21
22
23
def execute(self, sql: str, values: Optional[dict[str, Any]] = None) -> Any:
    with self.connection() as conn:
        return conn.execute(sql, values or ())

fetchall(results)

Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/blackline/adapters/sqlite/sqlite.py 
25
26
def fetchall(self, results: Cursor) -> list[tuple]:
    return results.fetchall()

mask_template()

Mask template for SQLite.

SQLite does not support regex, so we have to use a nested replace function to mask digits in a column. We are limited to digits because if we mask mask characters will will run into an OperationalError: parser stack overflow

Returns:

Name Type Description
str str

Mask template for SQLite.
Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/blackline/adapters/sqlite/sqlite.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def mask_template(self) -> str:
    """Mask template for SQLite.

    SQLite does not support regex, so we have to use a nested replace
    function to mask digits in a column. We are limited to digits because
    if we mask mask characters will will run into an OperationalError: parser
    stack overflow

    Returns:
        str: Mask template for SQLite.
    """

    replace_str = "{{ name }}"
    for c in string.digits:
        replace_str = f"REPLACE({replace_str}, '{c}', :{{{{ value }}}})"

    return f"{{{{ name }}}} = {replace_str}"

redact_template()

Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/blackline/adapters/sqlite/sqlite.py 
34
35
def redact_template(self) -> str:
    return "{{ name }} = null"

replace_template()

Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/blackline/adapters/sqlite/sqlite.py 
37
38
def replace_template(self) -> str:
    return "{{ name }} = :{{ value }}"

set_template()

Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/blackline/adapters/sqlite/sqlite.py 
31
32
def set_template(self) -> str:
    return "SET"

table_exists(table)

Check if a table exists.

Parameters:

Name Type Description Default
table str

Table name.

 required

Returns:

Name Type Description
bool bool

True if the table exists.
Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/blackline/adapters/sqlite/sqlite.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def table_exists(self, table: str) -> bool:
    """Check if a table exists.

    Args:
        table (str): Table name.

    Returns:
        bool: True if the table exists.
    """
    with self.connection() as conn:
        results = conn.execute(
            "SELECT name FROM sqlite_master WHERE type='table' AND name=:table",
            {"table": table},
        ).fetchall()
    return len(results) > 0

test_connection()

Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/blackline/adapters/sqlite/sqlite.py 
61
62
63
64
65
66
67
def test_connection(self) -> bool:
    try:
        with self.connection() as conn:
            conn.execute("SELECT 1")
            return True
    except Exception:
        return False

update_template()

Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/blackline/adapters/sqlite/sqlite.py 
28
29
def update_template(self) -> str:
    return "UPDATE {{ table }}"

where_template()

Source code in /opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/blackline/adapters/sqlite/sqlite.py 
58
59
def where_template(self) -> str:
    return "WHERE {{ datetime_column }} < :cutoff"