Skip to content

Commit

Permalink
Add validator that checks if index names are unique within the schema
Browse files Browse the repository at this point in the history
  • Loading branch information
JeremyMcCormick committed Aug 26, 2024
1 parent c687d13 commit 33c3465
Show file tree
Hide file tree
Showing 2 changed files with 42 additions and 1 deletion.
32 changes: 31 additions & 1 deletion python/felis/datamodel.py
Original file line number Diff line number Diff line change
Expand Up @@ -848,7 +848,37 @@ def check_unique_constraint_names(self: Schema) -> Schema:
constraint_names.add(constraint_name)

if duplicate_names:
raise ValueError(f"Duplicate constraint names found in schema: {duplicate_names}")
raise ValueError(f"Duplicate index names found in schema: {duplicate_names}")

return self

@model_validator(mode="after")
def check_unique_index_names(self: Schema) -> Schema:
"""Check for duplicate index names in the schema.
Returns
-------
`Schema`
The schema being validated.
Raises
------
ValueError
Raised if duplicate index names are found in the schema.
"""
index_names = set()
duplicate_names = []

for table in self.tables:
for index in table.indexes:
index_name = index.name
if index_name in index_names:
duplicate_names.append(index_name)
else:
index_names.add(index_name)

if duplicate_names:
raise ValueError(f"Duplicate index names found in schema: {duplicate_names}")

return self

Expand Down
11 changes: 11 additions & 0 deletions tests/test_datamodel.py
Original file line number Diff line number Diff line change
Expand Up @@ -478,6 +478,17 @@ def test_check_unique_constraint_names(self) -> None:
with self.assertRaises(ValidationError):
Schema(name="testSchema", id="#test_id", tables=[test_tbl])

def test_check_unique_index_names(self) -> None:
"""Test that index names are unique."""
test_col = Column(name="test_column1", id="#test_table#test_column1", datatype="int")
test_col2 = Column(name="test_column2", id="##test_table#test_column2", datatype="string", length=256)
test_tbl = Table(name="test_table", id="#test_table", columns=[test_col, test_col2])
test_idx = Index(name="idx_test", id="#idx_test", columns=[test_col.id])
test_idx2 = Index(name="idx_test", id="#idx_test2", columns=[test_col2.id])
test_tbl.indexes = [test_idx, test_idx2]
with self.assertRaises(ValidationError):
Schema(name="test_schema", id="#test-schema", tables=[test_tbl])

def test_model_validate(self) -> None:
"""Load a YAML test file and validate the schema data model."""
with open(TEST_YAML) as test_yaml:
Expand Down

0 comments on commit 33c3465

Please sign in to comment.