Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
59 changes: 53 additions & 6 deletions sql_compare/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -185,23 +185,70 @@ class Statement(TokenList):

UNKNOWN_TYPE = "UNKNOWN"

# Verbs that act on an object, so the type is "<verb> <object>"
# (e.g. CREATE TABLE). Anything else stands alone (SELECT, INSERT, ...).
# Every one of these is always followed by the keyword naming the object
# it acts on, which is what makes taking the next keyword safe. A verb that
# acts directly on a name (TRUNCATE foo) must NOT be added: the next
# keyword would then be a trailing option, inventing types like
# "TRUNCATE CASCADE" and "TRUNCATE RESTART".
OBJECT_VERBS: typing.ClassVar[frozenset[str]] = frozenset(
{"CREATE", "CREATE OR REPLACE", "ALTER", "DROP", "REFRESH"},
)
# Keywords that sit between the verb and the object without changing what
# the object is: a CREATE TEMPORARY TABLE is still a table, and a
# CREATE UNIQUE INDEX is still an index.
# Only words that can appear *before* the object are listed: a keyword
# that follows it (CONCURRENTLY, CASCADE, ...) is never reached.
MODIFIER_KEYWORDS: typing.ClassVar[frozenset[str]] = frozenset(
{"TEMPORARY", "TEMP", "GLOBAL", "LOCAL", "UNIQUE", "RECURSIVE"},
)
# Keywords that qualify the object instead of being it. A MATERIALIZED VIEW
# is its own kind of object, so the word is kept and the name continues.
OBJECT_QUALIFIERS: typing.ClassVar[frozenset[str]] = frozenset({"MATERIALIZED"})
# Verbs sqlparse does not lex as keywords, so they are recovered from the
# statement's first token instead of the keyword list.
UNLEXED_VERBS: typing.ClassVar[frozenset[str]] = frozenset({"REFRESH"})

@property
def statement_type(self) -> str:
"""Return the type of SQL statement."""
keywords: list[str] = [
t.normalized for t in self.token_list.tokens if t.is_keyword
t.normalized
for t in self.token_list.tokens
if t.is_keyword and t.normalized not in self.MODIFIER_KEYWORDS
]

# No keywords found
if not keywords:
return self.UNKNOWN_TYPE

# Need 2 keywords to determine the statement type (e.g.: CREATE TABLE)
if keywords[0] in {"CREATE", "ALTER", "DROP"}:
return " ".join(keywords[:2])
words = self._leading_unlexed_verb() + keywords

# Only one word (e.g.: SELECT, INSERT, DELETE, etc.)
if words[0] not in self.OBJECT_VERBS:
return words[0]

# Verb plus the object it acts on, a qualifier extending the object
# name (e.g. CREATE MATERIALIZED VIEW).
parts: list[str] = [words[0]]
for word in words[1:]:
parts.append(word)
if word not in self.OBJECT_QUALIFIERS:
break

# Only one keyword (e.g.: SELECT, INSERT, DELETE, etc.)
return keywords[0]
return " ".join(parts)

def _leading_unlexed_verb(self) -> list[str]:
"""Return the leading verb when sqlparse did not lex it as a keyword."""
for token in self.token_list.tokens:
if token.is_whitespace:
continue
if token.is_keyword:
return []
verb = str(token.normalized).upper()
return [verb] if verb in self.UNLEXED_VERBS else []
return []

@property
def str_tokens(self) -> list[str]:
Expand Down
30 changes: 30 additions & 0 deletions tests/test_sql_compare.py
Original file line number Diff line number Diff line change
Expand Up @@ -196,6 +196,36 @@ def test_compare_neq(first_sql: str, second_sql: str) -> None:
),
("DROP TABLE foo", "DROP TABLE"),
("DROP INDEX foo_idx", "DROP INDEX"),
# A MATERIALIZED VIEW is its own kind of object, so the qualifier is
# part of the type. sqlparse does not lex REFRESH as a keyword, so it
# is recovered from the statement's first token.
(
"CREATE MATERIALIZED VIEW foo AS SELECT id FROM bar WITH NO DATA",
"CREATE MATERIALIZED VIEW",
),
("DROP MATERIALIZED VIEW foo", "DROP MATERIALIZED VIEW"),
("ALTER MATERIALIZED VIEW foo RENAME TO bar", "ALTER MATERIALIZED VIEW"),
("REFRESH MATERIALIZED VIEW foo", "REFRESH MATERIALIZED VIEW"),
("CREATE OR REPLACE VIEW foo AS SELECT id FROM bar", "CREATE OR REPLACE VIEW"),
# Modifier keywords do not change what the object is.
("CREATE TEMPORARY TABLE foo (id INT)", "CREATE TABLE"),
("CREATE TEMP TABLE foo (id INT)", "CREATE TABLE"),
("CREATE GLOBAL TEMPORARY TABLE foo (id INT)", "CREATE TABLE"),
("CREATE LOCAL TEMPORARY TABLE foo (id INT)", "CREATE TABLE"),
("CREATE RECURSIVE VIEW foo AS SELECT id FROM bar", "CREATE VIEW"),
("CREATE UNIQUE INDEX foo_idx ON foo (id)", "CREATE INDEX"),
# UNLOGGED is not a keyword to sqlparse, it comes through as an
# Identifier exactly like a table name would. It is skipped by the
# is_keyword filter rather than by MODIFIER_KEYWORDS, so this pins the
# filter, not the modifier list.
("CREATE UNLOGGED TABLE foo (id INT)", "CREATE TABLE"),
# TRUNCATE acts on a name, not on an object type, so it must stay out
# of OBJECT_VERBS: the next keyword is a trailing option, and pairing
# it with the verb would invent a type per spelling.
("TRUNCATE foo", "TRUNCATE"),
("TRUNCATE TABLE foo", "TRUNCATE"),
("TRUNCATE foo CASCADE", "TRUNCATE"),
("TRUNCATE foo RESTART IDENTITY", "TRUNCATE"),
],
)
def test_statement_type(sql: str, expected_type: str) -> None:
Expand Down