Skip to content

statement

Module to define a SQL Statement

Statement

Bases: BaseModel

SQL Statment class

Source code in xerini/statement.py 
 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
class Statement(BaseModel):
    """SQL Statment class"""

    text: str
    _parsed_text: Optional[Parser] = None
    _query_type: Optional[QueryType] = None

    @model_validator(mode="before")
    @classmethod
    def only_one_statement(cls, data: dict):
        """Model validator to make sure that there is only a single sql statement in text"""
        if not 0 < meaningful_strings_count(data["text"]) < 2:
            raise ValueError(
                f"Expected a string containing a single SQL statement, but {data['text']=} !"
            )
        return data

    @property
    def formatted_text(self) -> str:
        """Returns sqlparse.formatted text"""
        return sqlparse.format(
            self.text,
            reindent=True,
            keyword_case="upper",
            strip_comments=True,
            comma_first=True,
            indent_tabs=True,
        )

    @model_validator(mode="after")
    def set_parsed_text(self) -> Self:
        """Model validator to set the parsed text"""
        self._parsed_text = Parser(self.formatted_text)
        return self

    @model_validator(mode="after")
    def set_query_type(self) -> Self:
        """Model validator to fix the augmented via the supported_query_types.py query type"""
        self._query_type = None
        try:
            self._query_type = self.parsed_text.query_type
        except ValueError as _ve:
            pass
        return self

    @property
    def query_type(self) -> QueryType | None:
        """The augmented via the supported_query_types.py query type"""
        return self._query_type

    @property
    def parsed_text(self) -> Parser:
        """Parsed_text property"""
        return self._parsed_text

    def __str__(self) -> str:
        """Returns a string for printing"""
        return self.text

    @property
    def statement_type(self) -> StatementType:
        """Whether the statement is a views, updates, or transfers data"""
        return StatementType.from_query_type(self.query_type)

    @property
    def write_type(self) -> TableWriteType:
        """Whether the statement builds, updates, or temporarily write"""
        return TableWriteType.from_statement_query(self.statement_type, self.query_type)

    @property
    def tables(self) -> Set[Optional[str]]:
        """Tables integral to the statement"""
        ans: Set = set()
        if self.query_type is not None:
            ans = set(self.parsed_text.tables)
        return ans

    @property
    def non_empty_tokens(self) -> list[sqlparse.sql.Token]:
        """From the sql_metadata Parser"""
        ans = []
        if self.query_type is None:
            pass
        else:
            ans = self.parsed_text.non_empty_tokens
        return ans

    @property
    def normalized_successive_token_pairs(self) -> list[tuple[str, str]]:
        """Needed to get to the affected tables"""
        return [
            (sp[0].normalized, sp[1].normalized)
            for sp in successive_pairs(self.parsed_text.tokens)
        ]

    @property
    def affected_table(self) -> str | None:
        """Where the data is going, according to the statement"""
        ans = None
        if self.query_type in {QueryType.TRUNCATE, QueryType.DROP}:
            ans = list(self.tables)[0]
        elif self.query_type == QueryType.UPDATE:
            ans = [
                tbl
                for tbl in self.tables
                if tbl
                and ("UPDATE", tbl.upper()) in self.normalized_successive_token_pairs
            ][0]
        elif ("DELETE", "FROM") in self.normalized_successive_token_pairs:
            ans = following_pairs_second_item(
                self.normalized_successive_token_pairs, ("DELETE", "FROM")
            )
        elif ("CREATE", "TABLE") in self.normalized_successive_token_pairs:
            ans = following_pairs_second_item(
                self.normalized_successive_token_pairs, ("CREATE", "TABLE")
            )
        elif ("TEMP", "TABLE") in self.normalized_successive_token_pairs:
            ans = following_pairs_second_item(
                self.normalized_successive_token_pairs, ("TEMP", "TABLE")
            )
        elif ("TEMPORARY", "TABLE") in self.normalized_successive_token_pairs:
            ans = following_pairs_second_item(
                self.normalized_successive_token_pairs, ("TEMPORARY", "TABLE")
            )
        elif ("INSERT", "INTO") in self.normalized_successive_token_pairs:
            ans = following_pairs_second_item(
                self.normalized_successive_token_pairs, ("INSERT", "INTO")
            )
        else:
            pass
        return ans.lower() if ans is not None else None

    def what_are_we_truncating_dropping(self) -> str | None:
        """We need to specify the destination table when we truncate"""
        if not self.query_type in {QueryType.TRUNCATE, QueryType.DROP}:
            raise TypeError(f"This method is unsupported for {self.query_type}!")
        return self.affected_table

    @property
    def source_tables(self) -> Set[Optional[str]]:
        """Tables from which the data is being sourced, according to the statement"""
        sources: Set = set()
        if self.statement_type in {StatementType.UPDATE, StatementType.VIEW}:
            if self.query_type in {QueryType.TRUNCATE, QueryType.DROP}:
                pass
            elif self.query_type == QueryType.DELETE:
                sources = self.tables - {
                    self.affected_table,
                }
            else:
                sources = self.tables
        else:
            sources = self.tables - {
                self.affected_table,
            }
        return sources

affected_table property

Where the data is going, according to the statement

formatted_text property

Returns sqlparse.formatted text

non_empty_tokens property

From the sql_metadata Parser

normalized_successive_token_pairs property

Needed to get to the affected tables

parsed_text property

Parsed_text property

query_type property

The augmented via the supported_query_types.py query type

source_tables property

Tables from which the data is being sourced, according to the statement

statement_type property

Whether the statement is a views, updates, or transfers data

tables property

Tables integral to the statement

write_type property

Whether the statement builds, updates, or temporarily write

__str__()

Returns a string for printing

Source code in xerini/statement.py 
77
78
79
def __str__(self) -> str:
    """Returns a string for printing"""
    return self.text

only_one_statement(data) classmethod

Model validator to make sure that there is only a single sql statement in text

Source code in xerini/statement.py 
29
30
31
32
33
34
35
36
37
@model_validator(mode="before")
@classmethod
def only_one_statement(cls, data: dict):
    """Model validator to make sure that there is only a single sql statement in text"""
    if not 0 < meaningful_strings_count(data["text"]) < 2:
        raise ValueError(
            f"Expected a string containing a single SQL statement, but {data['text']=} !"
        )
    return data

set_parsed_text()

Model validator to set the parsed text

Source code in xerini/statement.py 
51
52
53
54
55
@model_validator(mode="after")
def set_parsed_text(self) -> Self:
    """Model validator to set the parsed text"""
    self._parsed_text = Parser(self.formatted_text)
    return self

set_query_type()

Model validator to fix the augmented via the supported_query_types.py query type

Source code in xerini/statement.py 
57
58
59
60
61
62
63
64
65
@model_validator(mode="after")
def set_query_type(self) -> Self:
    """Model validator to fix the augmented via the supported_query_types.py query type"""
    self._query_type = None
    try:
        self._query_type = self.parsed_text.query_type
    except ValueError as _ve:
        pass
    return self

what_are_we_truncating_dropping()

We need to specify the destination table when we truncate

Source code in xerini/statement.py 
154
155
156
157
158
def what_are_we_truncating_dropping(self) -> str | None:
    """We need to specify the destination table when we truncate"""
    if not self.query_type in {QueryType.TRUNCATE, QueryType.DROP}:
        raise TypeError(f"This method is unsupported for {self.query_type}!")
    return self.affected_table