|
| 1 | +"""Strict ownership and structural verification for OpenStatSpec SQL catalogs.""" |
| 2 | + |
| 3 | +from __future__ import annotations |
| 4 | + |
| 5 | +import re |
| 6 | +from collections.abc import Iterable, Mapping |
| 7 | +from typing import Any |
| 8 | + |
| 9 | +from sqlalchemy import MetaData, Table, inspect, select |
| 10 | + |
| 11 | +from ..core import UnsupportedOperationError |
| 12 | +from .normative import CATALOG_CONTRACT_ID, CATALOG_SCHEMA_VERSION |
| 13 | + |
| 14 | + |
| 15 | +def _normalized_sql_type(inspector: Any, value: Any) -> str: |
| 16 | + compiled = " ".join( |
| 17 | + str(value.compile(dialect=inspector.bind.dialect)).strip().upper().split() |
| 18 | + ) |
| 19 | + if inspector.bind.dialect.name in {"mysql", "mariadb"}: |
| 20 | + if compiled in {"BOOL", "BOOLEAN", "TINYINT(1)"}: |
| 21 | + return "BOOLEAN/TINYINT(1)" |
| 22 | + compiled = re.sub( |
| 23 | + r"\b(TINYINT|SMALLINT|MEDIUMINT|INTEGER|INT|BIGINT)\(\d+\)", |
| 24 | + r"\1", |
| 25 | + compiled, |
| 26 | + ) |
| 27 | + return compiled |
| 28 | + |
| 29 | + |
| 30 | +def _normalized_default(value: Any) -> str | None: |
| 31 | + if value is None: |
| 32 | + return None |
| 33 | + result = " ".join(str(value).strip().split()) |
| 34 | + while result.startswith("(") and result.endswith(")"): |
| 35 | + result = result[1:-1].strip() |
| 36 | + return result |
| 37 | + |
| 38 | + |
| 39 | +def _expected_unique_constraints(table: Table) -> set[tuple[str, ...]]: |
| 40 | + return { |
| 41 | + tuple(column.name for column in constraint.columns) |
| 42 | + for constraint in table.constraints |
| 43 | + if getattr(constraint, "__visit_name__", "") == "unique_constraint" |
| 44 | + } |
| 45 | + |
| 46 | + |
| 47 | +def _actual_unique_constraints(inspector: Any, table_name: str) -> set[tuple[str, ...]]: |
| 48 | + constraints = { |
| 49 | + tuple(str(name) for name in item.get("column_names") or ()) |
| 50 | + for item in inspector.get_unique_constraints(table_name) |
| 51 | + } |
| 52 | + constraints.update( |
| 53 | + tuple(str(name) for name in item.get("column_names") or ()) |
| 54 | + for item in inspector.get_indexes(table_name) |
| 55 | + if item.get("unique") |
| 56 | + ) |
| 57 | + constraints.discard(()) |
| 58 | + return constraints |
| 59 | + |
| 60 | + |
| 61 | +def _expected_foreign_keys(table: Table) -> set[tuple[Any, ...]]: |
| 62 | + return { |
| 63 | + ( |
| 64 | + tuple(column.name for column in constraint.columns), |
| 65 | + next(iter(constraint.elements)).column.table.name, |
| 66 | + tuple(element.column.name for element in constraint.elements), |
| 67 | + ) |
| 68 | + for constraint in table.foreign_key_constraints |
| 69 | + } |
| 70 | + |
| 71 | + |
| 72 | +def _actual_foreign_keys(inspector: Any, table_name: str) -> set[tuple[Any, ...]]: |
| 73 | + return { |
| 74 | + ( |
| 75 | + tuple(str(name) for name in item.get("constrained_columns") or ()), |
| 76 | + str(item.get("referred_table") or ""), |
| 77 | + tuple(str(name) for name in item.get("referred_columns") or ()), |
| 78 | + ) |
| 79 | + for item in inspector.get_foreign_keys(table_name) |
| 80 | + } |
| 81 | + |
| 82 | + |
| 83 | +def _normalized_check_sql(value: Any) -> str: |
| 84 | + result = " ".join(str(value).strip().casefold().split()) |
| 85 | + result = result.replace("`", "").replace('"', "") |
| 86 | + result = re.sub(r"\[([^]]+)\]", r"\1", result) |
| 87 | + result = re.sub(r"(?<![a-z0-9])_[a-z0-9]+(?=')", "", result) |
| 88 | + result = re.sub( |
| 89 | + r"::\s*(?:character varying|varchar|text|integer|bigint|smallint|boolean)\b", |
| 90 | + "", |
| 91 | + result, |
| 92 | + ) |
| 93 | + result = re.sub(r"\(([a-z_][a-z0-9_]*)\)", r"\1", result) |
| 94 | + while result.startswith("(") and result.endswith(")"): |
| 95 | + result = result[1:-1].strip() |
| 96 | + return result |
| 97 | + |
| 98 | + |
| 99 | +def _expected_check_constraints(table: Table) -> set[str]: |
| 100 | + return { |
| 101 | + _normalized_check_sql(constraint.sqltext) |
| 102 | + for constraint in table.constraints |
| 103 | + if getattr(constraint, "__visit_name__", "") |
| 104 | + == "table_or_column_check_constraint" |
| 105 | + } |
| 106 | + |
| 107 | + |
| 108 | +def _actual_check_constraints(inspector: Any, table_name: str) -> set[str]: |
| 109 | + return { |
| 110 | + _normalized_check_sql(item.get("sqltext") or "") |
| 111 | + for item in inspector.get_check_constraints(table_name) |
| 112 | + } |
| 113 | + |
| 114 | + |
| 115 | +def _table_shape_valid( |
| 116 | + inspector: Any, |
| 117 | + table: Table, |
| 118 | + *, |
| 119 | + allowed_missing: Iterable[str] = (), |
| 120 | +) -> bool: |
| 121 | + actual = { |
| 122 | + str(column["name"]): column |
| 123 | + for column in inspector.get_columns(table.name) |
| 124 | + } |
| 125 | + expected = {column.name: column for column in table.columns} |
| 126 | + missing = set(expected) - set(actual) |
| 127 | + if set(actual) - set(expected) or missing - set(allowed_missing): |
| 128 | + return False |
| 129 | + for name in set(actual) & set(expected): |
| 130 | + expected_column = expected[name] |
| 131 | + actual_column = actual[name] |
| 132 | + if ( |
| 133 | + _normalized_sql_type(inspector, expected_column.type) |
| 134 | + != _normalized_sql_type(inspector, actual_column["type"]) |
| 135 | + or bool(actual_column.get("nullable")) != bool(expected_column.nullable) |
| 136 | + ): |
| 137 | + return False |
| 138 | + expected_default = _normalized_default( |
| 139 | + expected_column.server_default.arg |
| 140 | + if expected_column.server_default is not None else None |
| 141 | + ) |
| 142 | + if _normalized_default(actual_column.get("default")) != expected_default: |
| 143 | + return False |
| 144 | + if actual_column.get("identity") is not None or actual_column.get("computed") is not None: |
| 145 | + return False |
| 146 | + expected_pk = tuple(column.name for column in table.primary_key.columns) |
| 147 | + actual_pk = tuple( |
| 148 | + str(name) |
| 149 | + for name in inspector.get_pk_constraint(table.name).get("constrained_columns") or () |
| 150 | + ) |
| 151 | + return ( |
| 152 | + actual_pk == expected_pk |
| 153 | + and _actual_unique_constraints(inspector, table.name) |
| 154 | + == _expected_unique_constraints(table) |
| 155 | + and _actual_foreign_keys(inspector, table.name) |
| 156 | + == _expected_foreign_keys(table) |
| 157 | + and _actual_check_constraints(inspector, table.name) |
| 158 | + == _expected_check_constraints(table) |
| 159 | + ) |
| 160 | + |
| 161 | + |
| 162 | +def _reject(relations: Iterable[str] = ()) -> None: |
| 163 | + suffix = f": {', '.join(sorted(relations))}" if relations else "" |
| 164 | + raise UnsupportedOperationError( |
| 165 | + "The selected database catalog contains foreign, obsolete, or " |
| 166 | + f"structurally incompatible relations{suffix}. Remove them manually " |
| 167 | + "before continuing." |
| 168 | + ) |
| 169 | + |
| 170 | + |
| 171 | +def verify_catalog_relations( |
| 172 | + connection: Any, |
| 173 | + normative: Any, |
| 174 | + *, |
| 175 | + allowed_migrations: Mapping[str, set[str]] | None = None, |
| 176 | +) -> None: |
| 177 | + """Accept only exact normative and validated optional OpenStatSpec profiles.""" |
| 178 | + inspector = inspect(connection) |
| 179 | + existing_tables = set(inspector.get_table_names()) |
| 180 | + existing_views = set(inspector.get_view_names()) |
| 181 | + normative_tables = {table.name: table for table in normative.all()} |
| 182 | + if not set(normative_tables) <= existing_tables: |
| 183 | + _reject(set(normative_tables) - existing_tables) |
| 184 | + if any( |
| 185 | + not _table_shape_valid(inspector, table) |
| 186 | + for table in normative_tables.values() |
| 187 | + ): |
| 188 | + _reject(normative_tables) |
| 189 | + |
| 190 | + identities = connection.execute(select(normative.catalog_identity)).mappings().all() |
| 191 | + if len(identities) != 1 or ( |
| 192 | + identities[0]["catalog_identity_key"] != 1 |
| 193 | + or identities[0]["contract_id"] != CATALOG_CONTRACT_ID |
| 194 | + or identities[0]["schema_version"] != CATALOG_SCHEMA_VERSION |
| 195 | + ): |
| 196 | + _reject({normative.catalog_identity.name}) |
| 197 | + |
| 198 | + owned_tables = set(normative_tables) |
| 199 | + owned_views: set[str] = set() |
| 200 | + owned_tables.update( |
| 201 | + str(name) |
| 202 | + for name in connection.execute( |
| 203 | + select(normative.dataset.c.physical_table_name) |
| 204 | + ).scalars() |
| 205 | + if name |
| 206 | + ) |
| 207 | + |
| 208 | + from .workflow import PROFILE_ID, PROFILE_SCHEMA_VERSION, workflow_catalog |
| 209 | + |
| 210 | + workflow = workflow_catalog(MetaData()) |
| 211 | + workflow_tables = {table.name: table for table in workflow.all()} |
| 212 | + workflow_present = set(workflow_tables) & existing_tables |
| 213 | + if workflow_present: |
| 214 | + if set(workflow_tables) - existing_tables or any( |
| 215 | + not _table_shape_valid(inspector, table) |
| 216 | + for table in workflow_tables.values() |
| 217 | + ): |
| 218 | + _reject(workflow_present) |
| 219 | + identity_rows = connection.execute( |
| 220 | + select(workflow.transformation_profile_identity) |
| 221 | + ).mappings().all() |
| 222 | + if len(identity_rows) != 1 or ( |
| 223 | + identity_rows[0]["profile_identity_key"] != 1 |
| 224 | + or identity_rows[0]["contract_id"] != PROFILE_ID |
| 225 | + or identity_rows[0]["schema_version"] != PROFILE_SCHEMA_VERSION |
| 226 | + or identity_rows[0]["core_contract_id"] != CATALOG_CONTRACT_ID |
| 227 | + ): |
| 228 | + _reject({workflow.transformation_profile_identity.name}) |
| 229 | + owned_tables.update(workflow_tables) |
| 230 | + for row in connection.execute(select( |
| 231 | + workflow.derived_dataset.c.physical_relation_name, |
| 232 | + workflow.derived_dataset.c.output_mode, |
| 233 | + )).mappings(): |
| 234 | + name = str(row["physical_relation_name"]) |
| 235 | + if row["output_mode"] == "view": |
| 236 | + owned_views.add(name) |
| 237 | + else: |
| 238 | + owned_tables.add(name) |
| 239 | + |
| 240 | + from .inplace_transform import apply_audit_catalog |
| 241 | + |
| 242 | + audit = apply_audit_catalog(MetaData()) |
| 243 | + if audit.name in existing_tables: |
| 244 | + if not _table_shape_valid( |
| 245 | + inspector, |
| 246 | + audit, |
| 247 | + allowed_missing=(allowed_migrations or {}).get(audit.name, set()), |
| 248 | + ): |
| 249 | + _reject({audit.name}) |
| 250 | + owned_tables.add(audit.name) |
| 251 | + |
| 252 | + foreign = (existing_tables - owned_tables) | (existing_views - owned_views) |
| 253 | + missing = (owned_tables - existing_tables) | (owned_views - existing_views) |
| 254 | + collisions = (owned_tables & existing_views) | (owned_views & existing_tables) |
| 255 | + if foreign or missing or collisions: |
| 256 | + _reject(foreign | missing | collisions) |
0 commit comments