ManyToMany class¶

@classmethod
def build_field(cls, **kwargs: Any) -> BaseFieldType:
    """
    Constructs and returns a new field instance based on the factory's configuration.

    This method orchestrates the creation of a `BaseFieldType` instance.
    It determines the column type, Pydantic type, and constraints from the
    provided `kwargs` and the factory's properties. It then instantiates
    the field and calls `overwrite_methods` to apply any factory-defined
    method overrides.

    Args:
        **kwargs (Any): Keyword arguments for configuring the field.

    Returns:
        BaseFieldType: The newly constructed and configured field instance.
    """
    column_type = cls.get_column_type(kwargs)
    pydantic_type = cls.get_pydantic_type(kwargs)
    constraints = cls.get_constraints(kwargs)

    # Get the dynamically generated field class.
    new_field = cls._get_field_cls(cls)

    # Instantiate the new field with all relevant properties.
    new_field_obj: BaseFieldType = new_field(  # type: ignore
        field_type=pydantic_type,
        annotation=pydantic_type,
        column_type=column_type,
        constraints=constraints,
        factory=cls,  # Store a reference to the factory itself.
        **kwargs,
    )
    # Apply any methods from the factory that should override the field's methods.
    cls.overwrite_methods(new_field_obj)
    return new_field_obj

@classmethod
def overwrite_methods(cls, field_obj: BaseFieldType) -> None:
    """
    Overwrites methods on the given `field_obj` with methods defined in the factory.

    This method iterates through the factory's own methods. If a method's name
    is present in `methods_overwritable_by_factory`, it replaces the corresponding
    method on the `field_obj`. It handles class methods and ensures proper
    `staticmethod` wrapping for consistent behavior across Python versions.

    Args:
        field_obj (BaseFieldType): The field instance whose methods are to be
                                   overwritten.
    """
    for key in dir(cls):
        # Check if the method is intended to be overwritten and exists in the factory.
        if key in cls.methods_overwritable_by_factory and hasattr(cls, key):
            fn = getattr(cls, key)
            original_fn = getattr(field_obj, key, None)
            # Unwrap the function if it's a method wrapper.
            if hasattr(fn, "__func__"):
                fn = fn.__func__

            # Set the method on the field_obj, binding it with partial to preserve
            # `cls` and `field_obj` context, and making it a static method.
            setattr(
                field_obj,
                key,
                # .__func__ is a workaround for python < 3.10, python >=3.10 works without
                staticmethod(partial(fn, cls, field_obj, original_fn=original_fn)).__func__,
            )

@classmethod
def repack(cls, field_obj: BaseFieldType) -> None:
    """
    Repacks methods on the given `field_obj` that were previously overwritten
    by the factory.

    This method is used to re-apply the `partial` and `staticmethod` wrappers
    to methods that were already overwritten. This can be necessary in scenarios
    where field objects are serialized/deserialized or otherwise lose their
    original method bindings. It ensures that the context (`cls`, `field_obj`)
    remains correctly bound.

    Args:
        field_obj (BaseFieldType): The field instance whose methods are to be repacked.
    """
    for key in dir(cls):
        # Check if the method is intended to be overwritten and exists in the factory.
        if key in cls.methods_overwritable_by_factory and hasattr(cls, key):
            packed_fn = getattr(field_obj, key, None)
            # If the method exists and has a `func` attribute (indicating a partial function).
            if packed_fn is not None and hasattr(packed_fn, "func"):
                # Reapply the staticmethod and partial binding.
                setattr(
                    field_obj,
                    key,
                    # .__func__ is a workaround for python < 3.10, python >=3.10 works without
                    staticmethod(
                        partial(packed_fn.func, cls, field_obj, **packed_fn.keywords)
                    ).__func__,
                )

@classmethod
def get_constraints(cls, kwargs: dict[str, Any]) -> Sequence[Any]:
    """
    Returns a sequence of constraints applicable to the column.

    This method can be overridden by subclasses to provide specific database
    constraints for the column associated with this field type.

    Args:
        kwargs (dict[str, Any]): Keyword arguments provided during field creation.

    Returns:
        Sequence[Any]: A sequence of constraint objects.
    """
    return []

@classmethod
def get_column_type(cls, kwargs: dict[str, Any]) -> Any:
    """
    Returns the SQL column type for the field.

    For regular fields, this will return the appropriate SQLAlchemy column type.
    For meta fields (fields that don't directly map to a database column),
    it should return `None`.

    Args:
        kwargs (dict[str, Any]): Keyword arguments provided during field creation.

    Returns:
        Any: The SQLAlchemy column type or `None` if it's a meta field.
    """
    return None

@classmethod
def get_pydantic_type(cls, kwargs: dict[str, Any]) -> Any:
    """
    Returns the Pydantic type for the field.

    This type is used by Pydantic for validation and serialization. By default,
    it returns the `field_type` attribute of the factory.

    Args:
        kwargs (dict[str, Any]): Keyword arguments provided during field creation.

    Returns:
        Any: The Pydantic type associated with the field.
    """
    return cls.field_type

@staticmethod
@lru_cache(None)
def _get_field_cls(cls: FieldFactory) -> BaseFieldType:
    """
    Internal static method to dynamically create and cache the actual field class.

    This method uses `lru_cache` to ensure that the field class is created
    only once for each `FieldFactory` type. It constructs a new type based
    on the `FieldFactory`'s `__name__` and `field_bases`.

    Args:
        cls (FieldFactory): The `FieldFactory` instance itself.

    Returns:
        BaseFieldType: The dynamically created and cached field class.
    """
    # Dynamically create a new type based on the factory's name and specified field bases.
    return cast(BaseFieldType, type(cls.__name__, cast(Any, cls.field_bases), {}))

def __new__(
    cls,
    to: BaseModelType | type[Model] | str,
    *,
    to_fields: Sequence[str] = (),
    to_foreign_key: str = "",
    from_fields: Sequence[str] = (),
    from_foreign_key: str = "",
    through: str | type[BaseModelType] | type[Model] = "",
    through_tablename: str | type[OLD_M2M_NAMING] | type[NEW_M2M_NAMING] = NEW_M2M_NAMING,
    embed_through: str | Literal[False] = False,
    **kwargs: Any,
) -> BaseFieldType:
    """
    Creates a new `ManyToManyField` instance.

    Args:
        to (Union[BaseModelType, type[Model], str]): The target model class or its string name.
        to_fields (Sequence[str]): Fields on the `target` model that the `through`
                                  model's foreign key to `target` will reference.
        to_foreign_key (str): The name of the foreign key field in the `through`
                             model that points to the `target` model.
        from_fields (Sequence[str]): Fields on the `owner` model that the `through`
                                    model's foreign key to `owner` will reference.
        from_foreign_key (str): The name of the foreign key field in the `through`
                               model that points to the `owner` model.
        through (str | type[BaseModelType] | type[Model]): The intermediate model.
        through_tablename (str | type[OLD_M2M_NAMING] | type[NEW_M2M_NAMING]):
            The name of the database table for the `through` model.
        embed_through (str | Literal[False]): If a string, embeds the `through` model.
        **kwargs (Any): Additional keyword arguments.

    Returns:
        BaseFieldType: The constructed `ManyToManyField` instance.
    """
    # Collect all relevant arguments into kwargs, excluding class-specific ones.
    kwargs = {
        **kwargs,
        **{key: value for key, value in locals().items() if key not in CLASS_DEFAULTS},
    }
    return super().__new__(
        cls,
        **kwargs,
    )

@classmethod
def validate(cls, kwargs: dict[str, Any]) -> None:
    """
    Validates the parameters for a `ManyToManyField` field.

    Enforces rules specific to Many-to-Many fields, such as disallowing
    server-side defaults/updates and ensuring `through_tablename` is set
    correctly. It also sets default values for `null`, `exclude`, `on_delete`,
    and `on_update`.

    Args:
        kwargs (dict[str, Any]): The dictionary of keyword arguments passed
                                 during field construction.

    Raises:
        FieldDefinitionError: If any validation rule is violated.
    """
    super().validate(kwargs)
    # Disallow auto_compute_server_default.
    if kwargs.get("auto_compute_server_default"):
        raise FieldDefinitionError(
            '"auto_compute_server_default" is not supported for ManyToMany.'
        ) from None
    kwargs["auto_compute_server_default"] = False
    # Disallow server_default.
    if kwargs.get("server_default"):
        raise FieldDefinitionError(
            '"server_default" is not supported for ManyToMany.'
        ) from None
    # Disallow server_onupdate.
    if kwargs.get("server_onupdate"):
        raise FieldDefinitionError(
            '"server_onupdate" is not supported for ManyToMany.'
        ) from None
    # Validate embed_through format.
    embed_through = kwargs.get("embed_through")
    if embed_through and "__" in embed_through:
        raise FieldDefinitionError('"embed_through" cannot contain "__".')

    # Set default values specific to Many-to-Many fields.
    kwargs["null"] = True  # M2M fields are conceptually null until related.
    kwargs["exclude"] = True  # M2M fields are typically excluded from direct model data.
    kwargs["on_delete"] = CASCADE  # Default cascade for M2M through table FKs.
    kwargs["on_update"] = CASCADE  # Default cascade for M2M through table FKs.

    # Validate through_tablename.
    through_tablename: str | type[OLD_M2M_NAMING] | type[NEW_M2M_NAMING] | None = kwargs.get(
        "through_tablename"
    )
    if not through_tablename or (
        not isinstance(through_tablename, str)
        and through_tablename is not OLD_M2M_NAMING
        and through_tablename is not NEW_M2M_NAMING
    ):
        raise FieldDefinitionError(
            'When specified, "through_tablename" must be set to either OLD_M2M_NAMING, NEW_M2M_NAMING or a non-empty string.'
        )