ManyToManyField¶

def get_validator(self, **kwargs: typing.Any) -> SaffierField:
    return SaffierField(**kwargs)  # pragma: no cover

def get_column_type(self) -> sqlalchemy.types.TypeEngine:
    raise NotImplementedError()  # pragma: no cover

def get_constraints(self) -> typing.Any:
    return []

def get_columns_nullable(self) -> bool:
    """Resolve SQLAlchemy nullability for this field during metadata build.

    The model declaration remains the canonical field definition, but
    migration generation can temporarily ask Saffier to render selected
    columns as nullable. That temporary override makes it possible to add a
    required field to a table that already has rows, then backfill defaults
    in the generated online migration before the final model constraint is
    enforced by later migrations.

    Returns:
        bool: ``True`` when the field declaration or active migration
        override should make generated columns nullable.
    """
    if self.null:
        return True

    force_fields = FORCE_FIELDS_NULLABLE.get()
    owner_name = self.owner.__name__ if self.owner is not None else ""
    return (owner_name, self.name) in force_fields or ("", self.name) in force_fields

def get_columns(self, name: str) -> typing.Sequence[sqlalchemy.Column]:
    return [self.get_column(name)]

def get_global_constraints(
    self,
    name: str,
    columns: typing.Sequence[sqlalchemy.Column],
    *,
    schema: str | None = None,
) -> typing.Sequence[sqlalchemy.Constraint | sqlalchemy.Index]:
    del name, columns, schema
    return []

def has_column(self) -> bool:
    return not self.is_virtual

def get_embedded_fields(
    self,
    field_name: str,
    existing_fields: dict[str, "Field"],
) -> dict[str, "Field"]:
    del field_name, existing_fields
    return {}

def expand_relationship(self, value: typing.Any) -> typing.Any:
    return value

def clean(
    self, name: str, value: typing.Any, *, for_query: bool = False
) -> dict[str, typing.Any]:
    """Normalize one logical field value into column-value pairs.

    Args:
        name: Logical field name.
        value: User-facing field value.
        for_query: Reserved for field implementations that need distinct
            query-time normalization.

    Returns:
        dict[str, Any]: Database payload keyed by column name.
    """
    del for_query
    if not self.has_column():
        return {}
    return {name: value}

def modify_input(self, name: str, kwargs: dict[str, typing.Any]) -> None:
    del name, kwargs

def raise_for_non_default(self, default: typing.Any, server_default: typing.Any) -> typing.Any:
    del default, server_default

def get_default_value(self) -> typing.Any:
    return self.validator.get_default_value()

def has_default(self) -> bool:
    return self.validator.has_default()

def get_default_values(
    self,
    field_name: str,
    cleaned_data: dict[str, typing.Any],
) -> dict[str, typing.Any]:
    if field_name in cleaned_data or not self.has_column():
        return {}
    return {field_name: self.get_default_value()}

def is_required(self) -> bool:
    if self.primary_key and self.autoincrement:
        return False
    return not (self.null or self.server_default is not None or self.has_default())

def get_is_null_clause(self, column: typing.Any) -> typing.Any:
    return column == None  # noqa: E711

def get_is_empty_clause(self, column: typing.Any) -> typing.Any:
    return self.get_is_null_clause(column)

def operator_to_clause(
    self,
    field_name: str,
    operator: str,
    table: sqlalchemy.Table,
    value: typing.Any,
) -> typing.Any:
    """Translate one lookup operator into a SQLAlchemy clause.

    Args:
        field_name: Logical field name being filtered.
        operator: Saffier lookup suffix such as `exact` or `isnull`.
        table: SQLAlchemy table containing the target column.
        value: Lookup value supplied by the caller.

    Returns:
        Any: SQLAlchemy boolean clause implementing the lookup.
    """
    column = table.columns[field_name]
    mapped_operator = settings.filter_operators.get(operator, operator)

    if mapped_operator == "isnull":
        is_null = self.get_is_null_clause(column)
        return is_null if value else sqlalchemy.not_(is_null)

    if mapped_operator == "isempty":
        is_empty = self.get_is_empty_clause(column)
        return is_empty if value else sqlalchemy.not_(is_empty)

    return getattr(column, mapped_operator)(value)

def get_column(self, name: str) -> sqlalchemy.Column:
    target = self.target
    to_field = target.fields[target.pkname]

    column_type = to_field.get_column_type()
    constraints = [
        sqlalchemy.schema.ForeignKey(
            f"{target.meta.tablename}.{target.pkname}",
            ondelete=saffier.CASCADE,
            onupdate=saffier.CASCADE,
            name=self.get_fk_name(name=name),
        )
    ]
    return sqlalchemy.Column(
        name,
        column_type,
        *constraints,
        nullable=self.get_columns_nullable(),
    )

def add_model_to_register(self, model: type["Model"]) -> None:
    """Register an auto-generated through model on the owning registry.

    Args:
        model: Through model created for the relation.
    """
    self.registry.models[model.__name__] = model

def get_fk_name(self, name: str) -> str:
    """Build a stable FK constraint name for the through table.

    Args:
        name: Local field or column name participating in the constraint.

    Returns:
        str: Constraint name truncated to common identifier limits.
    """
    fk_name = f"fk_{self.owner.meta.tablename}_{self.target.meta.tablename}_{self.target.pkname}_{name}"
    if not len(fk_name) > CHAR_LIMIT:
        return fk_name
    return fk_name[:CHAR_LIMIT]

def create_through_model(self) -> typing.Any:
    """Create or normalize the through model used to persist the relation.

    Returns:
        typing.Any: Concrete through model class used by the relation.

    Raises:
        ImproperlyConfigured: If an explicit through model does not expose a
            valid integer `id` primary key.
    """
    self.to = typing.cast(type["Model"], self.target)

    if self.through:
        if isinstance(self.through, str):
            registry = self.owner.meta.registry
            self.through = (
                registry.models.get(self.through) or registry.reflected[self.through]
            )

        if not self.from_foreign_key:
            candidates = [
                field_name
                for field_name, field in self.through.fields.items()
                if isinstance(field, ForeignKey) and field.target is self.owner
            ]
            if len(candidates) == 1:
                self.from_foreign_key = candidates[0]
        if not self.to_foreign_key:
            candidates = [
                field_name
                for field_name, field in self.through.fields.items()
                if isinstance(field, ForeignKey) and field.target is self.to
            ]
            if len(candidates) == 1:
                self.to_foreign_key = candidates[0]

        # M2M through models in Saffier are always required to expose an
        # auto-incrementing integer "id" primary key.
        id_field = self.through.fields.get("id")
        if id_field is None:
            has_non_id_pk = any(
                field.primary_key
                for field_name, field in self.through.fields.items()
                if field_name != "id"
            )
            if has_non_id_pk:
                raise ImproperlyConfigured(
                    "ManyToMany through models must use an auto-incrementing 'id' primary key."
                )
            id_field = saffier.IntegerField(primary_key=True, autoincrement=True)
            id_field.owner = self.through
            id_field.registry = self.through.meta.registry
            id_field.name = "id"
            self.through.fields["id"] = id_field
            self.through.meta.fields["id"] = id_field
            self.through.meta.fields_mapping["id"] = id_field
            self.through.meta.pk = id_field
            self.through.meta.pk_attribute = "id"
            self.through.pkname = "id"
            self.through._table = None
            self.through.__proxy_model__ = None
        elif not id_field.primary_key:
            raise ImproperlyConfigured(
                "ManyToMany through models must define 'id' as the primary key."
            )
        elif not isinstance(id_field, (IntegerField, SmallIntegerField, BigIntegerField)):
            raise ImproperlyConfigured(
                "ManyToMany through model 'id' primary key must be an integer type."
            )
        else:
            id_field.autoincrement = True

        self.through.meta.is_multi = True
        if not self.from_foreign_key:
            self.from_foreign_key = self.owner.__name__.lower()
        if not self.to_foreign_key:
            self.to_foreign_key = self.to.__name__.lower()
        self.through.meta.multi_related = [self.to_foreign_key]
        return self.through

    owner_name = self.owner.__name__
    to_name = self.to.__name__
    if not self.from_foreign_key:
        self.from_foreign_key = owner_name.lower()
    if not self.to_foreign_key:
        self.to_foreign_key = to_name.lower()
    class_name = f"{owner_name}{self.name.capitalize()}Through"
    if self.through_tablename is None or self.through_tablename is NEW_M2M_NAMING:
        tablename = class_name.lower()
    else:
        tablename = self.through_tablename.format(field=self).lower()
    if self.owner.meta.table_prefix:
        tablename = f"{self.owner.meta.table_prefix}_{tablename}"

    new_meta_namespace = {
        "tablename": tablename,
        "registry": self.owner.meta.registry,
        "is_multi": True,
        "multi_related": [self.to_foreign_key],
        "unique_together": [(self.from_foreign_key, self.to_foreign_key)],
    }

    new_meta = type("MetaInfo", (), new_meta_namespace)

    to_related_name = (
        f"{self.related_name}"
        if self.related_name
        else (
            f"{to_name.lower()}_{owner_name.lower()}{to_name.lower()}"
            if self.unique
            else f"{to_name.lower()}_{owner_name.lower()}{to_name.lower()}s_set"
        )
    )
    self.reverse_name = to_related_name if to_related_name is not False else ""

    through_model = type(
        class_name,
        (saffier.Model,),
        {
            "Meta": new_meta,
            "id": saffier.IntegerField(primary_key=True, autoincrement=True),
            f"{self.from_foreign_key}": ForeignKey(
                self.owner,
                on_delete=saffier.CASCADE,
                index=self.index,
                null=True,
                related_name=False,
            ),
            f"{self.to_foreign_key}": ForeignKey(
                self.to,
                on_delete=saffier.CASCADE,
                unique=self.unique,
                index=self.index,
                null=True,
                embed_parent=(
                    (self.from_foreign_key, self.embed_through or "")
                    if self.embed_through is not False
                    else None
                ),
                related_name=(False if self.related_name is False else to_related_name),
            ),
        },
    )
    self.through = typing.cast(type["Model"], through_model)

    self.add_model_to_register(self.through)
    tenant_models = getattr(self.registry, "tenant_models", None)
    if tenant_models is not None and (
        getattr(self.owner.meta, "is_tenant", False)
        or getattr(self.to.meta, "is_tenant", False)
    ):
        tenant_models[self.through.__name__] = self.through
        if getattr(self.owner.meta, "register_default", None) is False:
            self.registry.models.pop(self.through.__name__, None)