Metadata Models¶

@model_validator(mode="after")
def validate_numeric_shape(self) -> TableColumn:
    if self.comment is not None:
        self.comment = self.comment.strip()
        if not self.comment:
            raise ValueError("column comment cannot be empty")
    precision_set = self.numeric_precision is not None
    scale_set = self.numeric_scale is not None
    if precision_set != scale_set:
        raise ValueError(
            "numeric_precision and numeric_scale must be provided together"
        )
    if self.autoincrement and self.has_identity:
        raise ValueError("autoincrement and identity cannot be enabled together")
    if (
        self.identity_always
        or self.identity_start is not None
        or self.identity_increment is not None
        or self.identity_min_value is not None
        or self.identity_max_value is not None
        or self.identity_no_min_value
        or self.identity_no_max_value
        or self.identity_cycle
        or self.identity_cache is not None
        or self.identity_order
        or self.identity_on_null
    ):
        self.identity = True
    if self.identity_increment == 0:
        raise ValueError("identity_increment cannot be zero")
    if self.identity_on_null and self.identity_always:
        raise ValueError("identity_on_null requires BY DEFAULT identity")
    if self.identity_no_min_value and self.identity_min_value is not None:
        raise ValueError(
            "identity_no_min_value cannot be combined with identity_min_value"
        )
    if self.identity_no_max_value and self.identity_max_value is not None:
        raise ValueError(
            "identity_no_max_value cannot be combined with identity_max_value"
        )
    if self.identity_cache is not None and self.identity_cache <= 0:
        raise ValueError("identity_cache must be positive")
    if (
        self.identity_min_value is not None
        and self.identity_max_value is not None
        and self.identity_min_value > self.identity_max_value
    ):
        raise ValueError("identity_min_value cannot exceed identity_max_value")
    if self.on_delete is not None:
        self.on_delete = normalized_foreign_key_action(self.on_delete)
    if self.on_update is not None:
        self.on_update = normalized_foreign_key_action(self.on_update)
    self.deferrable = normalized_constraint_timing(
        self.deferrable,
        initially_deferred=self.initially_deferred,
    )
    self.enum_type_name = normalized_enum_identifier(
        self.enum_type_name,
        option_name="enum_type_name",
    )
    self.enum_schema = normalized_enum_identifier(
        self.enum_schema,
        option_name="enum_schema",
    )
    if self.enum_type_comment is not None:
        self.enum_type_comment = self.enum_type_comment.strip()
        if not self.enum_type_comment:
            raise ValueError("enum_type_comment cannot be empty")
    self.sqlite_on_conflict_primary_key = normalized_sqlite_conflict(
        self.sqlite_on_conflict_primary_key,
        option_name="sqlite_on_conflict_primary_key",
    )
    self.sqlite_on_conflict_not_null = normalized_sqlite_conflict(
        self.sqlite_on_conflict_not_null,
        option_name="sqlite_on_conflict_not_null",
    )
    self.sqlite_on_conflict_unique = normalized_sqlite_conflict(
        self.sqlite_on_conflict_unique,
        option_name="sqlite_on_conflict_unique",
    )
    return self

@model_validator(mode="before")
@classmethod
def normalize_backend_options(cls, data: Any) -> Any:
    if not isinstance(data, Mapping):
        return data
    values = dict(data)
    index_name = values.get("name") or "<unknown>"
    values["postgres_with"] = normalized_postgres_storage_parameters(
        values.get("postgres_with"),
        object_name=f"index '{index_name}'",
    )
    values["postgres_ops"] = normalized_postgres_index_ops(
        values.get("postgres_ops"),
        index_name=str(index_name),
    )
    values["method"] = normalized_storage_token(
        values.get("method"),
        option_name=f"index '{index_name}' method",
    )
    values["postgres_tablespace"] = normalized_storage_identifier(
        values.get("postgres_tablespace"),
        option_name=f"PostgreSQL tablespace for index '{index_name}'",
    )
    values["mssql_filegroup"] = normalized_storage_identifier(
        values.get("mssql_filegroup"),
        option_name=f"SQL Server filegroup for index '{index_name}'",
    )
    values["oracle_tablespace"] = normalized_storage_identifier(
        values.get("oracle_tablespace"),
        option_name=f"Oracle tablespace for index '{index_name}'",
    )
    values["mysql_prefix"] = normalized_mysql_index_prefix(
        values.get("mysql_prefix"),
        index_name=str(index_name),
    )
    values["mysql_length"] = normalized_mysql_index_lengths(
        values.get("mysql_length"),
        index_name=str(index_name),
    )
    values["mysql_using"] = normalized_storage_token(
        values.get("mysql_using"),
        option_name=f"MySQL/MariaDB index USING method for index '{index_name}'",
    )
    return values

@model_validator(mode="after")
def validate_metadata(self) -> TableIndex:
    if not self.columns and not self.expressions:
        raise ValueError(
            "table index must reference at least one column or SQL expression"
        )
    if self.comment is not None:
        self.comment = self.comment.strip()
        if not self.comment:
            raise ValueError("index comment cannot be empty")
    if self.mysql_length:
        unknown_columns = sorted(set(self.mysql_length) - set(self.columns))
        if unknown_columns:
            unknown = ", ".join(unknown_columns)
            raise ValueError(
                f"MySQL/MariaDB index prefix lengths for index '{self.name}' "
                f"reference columns not present in the index: {unknown}"
            )
    if self.postgres_ops:
        unknown_items = sorted(
            set(self.postgres_ops) - set(self.columns) - set(self.expressions)
        )
        if unknown_items:
            unknown = ", ".join(unknown_items)
            raise ValueError(
                f"PostgreSQL index operator classes for index '{self.name}' "
                f"reference columns or expressions not present in the index: {unknown}"
            )
    if self.postgres_nulls_not_distinct and not self.unique:
        raise ValueError(
            "PostgreSQL index NULLS NOT DISTINCT requires a unique index "
            f"for index '{self.name}'"
        )
    if self.mysql_prefix is not None:
        if self.unique:
            raise ValueError(
                "MySQL/MariaDB index prefixes cannot be combined with "
                f"unique indexes for index '{self.name}'"
            )
        if self.expressions:
            raise ValueError(
                "MySQL/MariaDB index prefixes cannot be combined with "
                f"expression indexes for index '{self.name}'"
            )
        if self.mysql_using is not None:
            raise ValueError(
                "MySQL/MariaDB index prefixes cannot be combined with "
                f"USING methods for index '{self.name}'"
            )
    self.oracle_compress = normalized_oracle_index_compress(
        self.oracle_compress,
        index_name=self.name,
    )
    if self.oracle_bitmap and self.unique:
        raise ValueError(
            f"Oracle bitmap indexes cannot be unique for index '{self.name}'"
        )
    return self

@model_validator(mode="after")
def validate_comment(self) -> TableCheck:
    if self.comment is not None:
        self.comment = self.comment.strip()
        if not self.comment:
            raise ValueError("table check constraint comment cannot be empty")
    return self

@model_validator(mode="after")
def validate_columns(self) -> TableUnique:
    if not self.columns:
        raise ValueError(
            "table unique constraint must reference at least one column"
        )
    self.postgres_include = normalized_non_empty_string_list(
        self.postgres_include,
        option_name=f"PostgreSQL INCLUDE columns for unique constraint '{self.name}'",
    )
    self.deferrable = normalized_constraint_timing(
        self.deferrable,
        initially_deferred=self.initially_deferred,
    )
    self.sqlite_on_conflict = normalized_sqlite_conflict(
        self.sqlite_on_conflict,
        option_name="sqlite_on_conflict",
    )
    self.mssql_filegroup = normalized_storage_identifier(
        self.mssql_filegroup,
        option_name=f"SQL Server filegroup for unique constraint '{self.name}'",
    )
    self.oracle_tablespace = normalized_storage_identifier(
        self.oracle_tablespace,
        option_name=f"Oracle tablespace for unique constraint '{self.name}'",
    )
    self.oracle_compress = normalized_oracle_index_compress(
        self.oracle_compress,
        object_name=f"unique constraint '{self.name}'",
    )
    if self.comment is not None:
        self.comment = self.comment.strip()
        if not self.comment:
            raise ValueError("table unique constraint comment cannot be empty")
    return self

@model_validator(mode="after")
def validate_shape(self) -> TableForeignKey:
    if not self.columns:
        raise ValueError(
            "table foreign key constraint must reference at least one column"
        )
    if len(self.columns) != len(self.foreign_columns):
        raise ValueError(
            "table foreign key columns and foreign_columns must have the same length"
        )
    if self.on_delete is not None:
        self.on_delete = normalized_foreign_key_action(self.on_delete)
    if self.on_update is not None:
        self.on_update = normalized_foreign_key_action(self.on_update)
    if self.match is not None:
        self.match = normalized_foreign_key_match(self.match)
    self.deferrable = normalized_constraint_timing(
        self.deferrable,
        initially_deferred=self.initially_deferred,
    )
    if self.comment is not None:
        self.comment = self.comment.strip()
        if not self.comment:
            raise ValueError("table foreign key constraint comment cannot be empty")
    return self

@model_validator(mode="before")
@classmethod
def normalize_backend_options(cls, data: Any) -> Any:
    if not isinstance(data, Mapping):
        return data
    values = dict(data)
    constraint_name = values.get("name") or "<unknown>"
    values["ops"] = normalized_postgres_exclusion_ops(
        values.get("ops"),
        constraint_name=str(constraint_name),
    )
    return values

@model_validator(mode="after")
def validate_shape(self) -> TableExclusion:
    if not self.columns and not self.expressions:
        raise ValueError(
            "table exclusion constraint must reference at least one column or SQL expression"
        )
    self.columns = [
        self._validated_element(column, operator, "column")
        for column, operator in self.columns
    ]
    self.expressions = [
        self._validated_element(expression, operator, "expression")
        for expression, operator in self.expressions
    ]
    if self.ops:
        element_names = {value for value, _operator in self.columns} | {
            value for value, _operator in self.expressions
        }
        unknown_items = sorted(set(self.ops) - element_names)
        if unknown_items:
            unknown = ", ".join(unknown_items)
            raise ValueError(
                f"PostgreSQL exclusion operator classes for constraint "
                f"'{self.name}' reference columns or expressions not present "
                f"in the constraint: {unknown}"
            )
    self.using = self.using.strip()
    if not self.using:
        raise ValueError("table exclusion constraint using cannot be empty")
    if self.where is not None:
        self.where = self.where.strip()
        if not self.where:
            raise ValueError("table exclusion constraint where cannot be empty")
    self.deferrable = normalized_constraint_timing(
        self.deferrable,
        initially_deferred=self.initially_deferred,
    )
    if self.comment is not None:
        self.comment = self.comment.strip()
        if not self.comment:
            raise ValueError("table exclusion constraint comment cannot be empty")
    return self

@model_validator(mode="after")
def validate_options(self) -> DatabaseNamespace:
    self.name = (
        normalized_enum_identifier(self.name, option_name="namespace name") or ""
    )
    if self.comment is not None:
        self.comment = self.comment.strip()
        if not self.comment:
            raise ValueError("namespace comment cannot be empty")
    return self

def to_runtime(self) -> tuple[str, str | None]:
    """Return a compact runtime descriptor for migration snapshots."""
    return (self.name, self.comment)

@model_validator(mode="after")
def validate_options(self) -> DatabaseSequence:
    self.name = (
        normalized_enum_identifier(self.name, option_name="sequence name") or ""
    )
    self.schema_name = normalized_enum_identifier(
        self.schema_name,
        option_name="sequence schema",
    )
    self.data_type = normalized_sequence_data_type(
        self.data_type,
        sequence_name=self.name,
    )
    if self.increment == 0:
        raise ValueError("sequence increment cannot be zero")
    if self.no_min_value and self.min_value is not None:
        raise ValueError("no_min_value cannot be combined with min_value")
    if self.no_max_value and self.max_value is not None:
        raise ValueError("no_max_value cannot be combined with max_value")
    if self.cache is not None and self.cache <= 0:
        raise ValueError("sequence cache must be positive")
    if (
        self.min_value is not None
        and self.max_value is not None
        and self.min_value > self.max_value
    ):
        raise ValueError("sequence min_value cannot exceed max_value")
    if self.comment is not None:
        self.comment = self.comment.strip()
        if not self.comment:
            raise ValueError("sequence comment cannot be empty")
    return self

def to_runtime(
    self,
) -> tuple[
    str,
    str | None,
    int | None,
    int | None,
    int | None,
    int | None,
    bool,
    int | None,
    str | None,
    str | None,
    bool,
    bool,
    bool,
]:
    """Return a compact runtime descriptor for migration snapshots."""
    return (
        self.name,
        self.schema_name,
        self.start,
        self.increment,
        self.min_value,
        self.max_value,
        self.cycle,
        self.cache,
        self.comment,
        self.data_type,
        self.order,
        self.no_min_value,
        self.no_max_value,
    )

@model_validator(mode="after")
def validate_options(self) -> DatabaseView:
    self.name = normalized_enum_identifier(self.name, option_name="view name") or ""
    self.schema_name = normalized_enum_identifier(
        self.schema_name,
        option_name="view schema",
    )
    definition = self.definition.strip()
    if definition.endswith(";"):
        definition = definition[:-1].strip()
    if not definition:
        raise ValueError("view definition cannot be empty")
    self.definition = definition
    if self.comment is not None:
        self.comment = self.comment.strip()
        if not self.comment:
            raise ValueError("view comment cannot be empty")
    return self

def to_runtime(self) -> tuple[str, str | None, str, bool, str | None]:
    """Return a compact runtime descriptor for migration snapshots."""
    return (
        self.name,
        self.schema_name,
        self.definition,
        self.materialized,
        self.comment,
    )