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(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.null)

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)