Add generic rule: models should implement uniqueness test for their PK

[matthieucan]

For models materialized as table or incremental:

• Loop over their columns to extract PK
• Loop over their tests to find a uniqueness test matching the PK columns

Consider both table- and column-level constraints and tests.

[druzhinin-kirill]

💪

[jochemvandooren]

Looks good! Some minor comments

[druzhinin-kirill]

Suggested change

	@rule(rule_filters={is_table()})
	def has_uniqueness_test(model: Model) -> RuleViolation | None:
	"""Model has uniqueness test for primary key."""
	# ruff: noqa: C901 [too-complex]
	# ruff: noqa: PLR0912 [too-many-branches]
	# Extract PK
	pk_columns = None
	# At column level?
	for column in model.columns:
	for column_constraint in column.constraints:
	if column_constraint.type == "primary_key":
	pk_columns = [column.name]
	break
	else:
	continue
	break
	# Or at table level?
	if pk_columns is None:
	for model_constraint in model.constraints:
	if model_constraint.type == "primary_key":
	pk_columns = model_constraint.columns
	break
	if pk_columns is None: # No PK, no need for uniqueness test
	return None
	# Look for matching uniqueness test
	if len(pk_columns) == 1:
	for column in model.columns:
	if column.name == pk_columns[0]:
	for data_test in column.tests:
	if data_test.type == "unique":
	return None
	for data_test in model.tests:
	if data_test.type == "unique_combination_of_columns":
	if set(data_test.kwargs.get("combination_of_columns")) == set(pk_columns): # type: ignore
	return None
	return RuleViolation("There is no uniqueness test defined and matching the PK.")
	pk_columns = next(
	(
	[column.name]
	for column in model.columns
	for constraint in column.constraints
	if constraint.type == "primary_key"
	),
	[],
	)
	if not pk_columns:
	pk_columns = next(
	(
	constraint.columns
	for constraint in model.constraints
	if constraint.columns and constraint.type == "primary_key"
	),
	[],
	)
	if len(pk_columns) == 1:
	pk_column = next(
	column for column in model.columns if column.name == pk_columns[0]
	)
	if not any(test.type == "unique" for test in pk_column.tests):
	return RuleViolation("There is no uniqueness test defined.")
	elif len(pk_columns) > 1:
	constraint_test = next(
	(
	test
	for test in model.tests
	if test.type == "unique_combination_of_columns"
	),
	None,
	)
	if not constraint_test:
	return RuleViolation("There is no uniqueness test defined.")
	if set(constraint_test.kwargs.get("combination_of_columns", [])) == set(
	pk_columns
	):
	return RuleViolation("Uniqueness test does not match the PK.")

I tried to get rid of lint exceptions and make it a bit more flat - feel free to use fully or partially

[matthieucan]

Thanks for challenging! I adopted some of it, making the function flatter.
I didn't take the list comprehensions however, as keeping breaks improves performances a bit. I could get rid of one ruff-ignore, not the other unfortunately.
Overall, while simplifying the logic, I was led to add 2 related rules, making it clearer when a model doesn't conform to expectations. PTAL :)

[jochemvandooren]

Nice, works well! 🚀

[npeshkov]

Proposing some changes to text and approving 👍

[npeshkov]

Suggested change

	def has_uniqueness_test(model: Model) -> RuleViolation | None:
	"""Model has uniqueness test for primary key."""
	def primary_key_uniqueness_is_tested(model: Model) -> RuleViolation | None:
	"""Model tests uniqueness of primary keys."""

[matthieucan]

I could go with has_uniqueness_test_for_pk, but find the suggestion too verbose

[npeshkov]

Suggested change

	f"No uniqueness test defined and matching PK {','.join(pk_columns)}."
	f"No uniqueness test defined for matching Primary Keys {','.join(pk_columns)}."

[matthieucan]

While acronyms are sometimes more confusing than anything else, in this context, I believe PK is well understood

[npeshkov]

Suggested change

	f"No unique constraint defined on PK column {column.name}."
	f"No unique constraint defined on Primary Key column {column.name}."