fix(migration): preserve quoted index semantics
This commit is contained in:
parent
485443a1e4
commit
1a97d7db44
1 changed files with 62 additions and 3 deletions
|
|
@ -54,20 +54,70 @@ def field(name, field_type="text", **options):
|
|||
|
||||
def index_name(statement):
|
||||
"""Extract an SQLite/PocketBase index name from a CREATE INDEX statement."""
|
||||
match = re.search(r"\bINDEX\s+`?([^`\s]+)`?", statement, re.IGNORECASE)
|
||||
match = re.search(r"\bINDEX\s+(?:`([^`]+)`|\"([^\"]+)\"|([^\s(]+))", statement, re.IGNORECASE)
|
||||
if not match:
|
||||
raise ValueError(f"not a CREATE INDEX statement: {statement!r}")
|
||||
return match.group(1).strip().lower()
|
||||
return next(value for value in match.groups() if value is not None).strip().lower()
|
||||
|
||||
|
||||
def normalise_index_sql(statement):
|
||||
"""Compare index SQL by meaning, not PocketBase's formatting choices."""
|
||||
value = statement.strip().rstrip(";")
|
||||
value = re.sub(r"`([^`]*)`", lambda match: f"`{match.group(1).lower()}`", value)
|
||||
chunks = []
|
||||
literals = []
|
||||
index = 0
|
||||
while index < len(value):
|
||||
character = value[index]
|
||||
if character == "'":
|
||||
start = index
|
||||
index += 1
|
||||
while index < len(value):
|
||||
if value[index] == "'":
|
||||
if index + 1 < len(value) and value[index + 1] == "'":
|
||||
index += 2
|
||||
continue
|
||||
index += 1
|
||||
break
|
||||
index += 1
|
||||
literals.append(value[start:index])
|
||||
chunks.append(f"\x00{len(literals) - 1}\x00")
|
||||
continue
|
||||
if character in '`"':
|
||||
quote = character
|
||||
index += 1
|
||||
identifier = []
|
||||
while index < len(value):
|
||||
if value[index] == quote:
|
||||
if index + 1 < len(value) and value[index + 1] == quote:
|
||||
identifier.append(quote)
|
||||
index += 2
|
||||
continue
|
||||
index += 1
|
||||
break
|
||||
identifier.append(value[index])
|
||||
index += 1
|
||||
chunks.append(f"`{''.join(identifier).lower()}`")
|
||||
continue
|
||||
if character.isalnum() or character in "_$":
|
||||
end = index + 1
|
||||
while end < len(value) and (value[end].isalnum() or value[end] in "_$"):
|
||||
end += 1
|
||||
chunks.append(value[index:end].lower())
|
||||
index = end
|
||||
continue
|
||||
chunks.append(character)
|
||||
index += 1
|
||||
|
||||
value = "".join(chunks)
|
||||
value = re.sub(r"\s+", " ", value).strip().lower()
|
||||
# SQLite/PocketBase may insert or remove spaces around these tokens while
|
||||
# preserving the same columns, uniqueness and WHERE predicate.
|
||||
value = re.sub(r"\s*([(),=<>!])\s*", r"\1", value)
|
||||
# Quoted identifiers and safe unquoted identifiers share one canonical
|
||||
# representation; single-quoted literals are placeholders until restore.
|
||||
value = value.replace("`", "")
|
||||
for index, literal in enumerate(literals):
|
||||
value = value.replace(f"\x00{index}\x00", literal)
|
||||
return value
|
||||
|
||||
|
||||
|
|
@ -208,6 +258,15 @@ def self_test():
|
|||
]
|
||||
for stored, near_miss in zip(pocketbase_stored, near_misses):
|
||||
assert merge_indexes([stored], [near_miss])[1] == [("replace", near_miss)]
|
||||
identifier_styles = (
|
||||
"CREATE UNIQUE INDEX `idx_style` ON `Delivery_Notes` (`Batch_Note`) WHERE `Batch_Note` != ''",
|
||||
'create unique index "IDX_STYLE" on "DELIVERY_NOTES" ("BATCH_NOTE") where "BATCH_NOTE" != \'\'',
|
||||
"CREATE UNIQUE INDEX idx_style ON delivery_notes (batch_note) WHERE batch_note != ''",
|
||||
)
|
||||
assert merge_indexes([identifier_styles[0]], [identifier_styles[1]])[1] == []
|
||||
assert merge_indexes([identifier_styles[0]], [identifier_styles[2]])[1] == []
|
||||
literal_near_miss = identifier_styles[2].replace("!= ''", "!= ' '")
|
||||
assert merge_indexes([identifier_styles[0]], [literal_near_miss])[1] == [("replace", literal_near_miss)]
|
||||
print("self-test: index replacement and idempotence passed")
|
||||
return 0
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue