RefForeignKey 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 validate(cls, kwargs: dict[str, Any]) -> None:
    """
    Validates the parameters for a foreign key field.

    This method enforces rules specific to foreign keys, such as:
    - `on_delete` must not be null.
    - If `SET_NULL` is used for `on_delete` or `on_update`, the field must be nullable.
    - `related_name` must be a string if provided and not `False`.
    It also sets a default `null` value if not provided.

    Args:
        kwargs (dict[str, Any]): A dictionary of parameters for the foreign key field.

    Raises:
        FieldDefinitionError: If any foreign key validation rule is violated.
    """
    on_update = kwargs.get("on_update", CASCADE)
    on_delete = kwargs.get("on_delete", RESTRICT)
    # Set default null to False if not explicitly provided.
    kwargs.setdefault("null", False)
    null = kwargs["null"]

    if on_delete is None:
        raise FieldDefinitionError("on_delete must not be null.")

    # If SET_NULL is enabled for on_delete, the field must be nullable.
    if on_delete == SET_NULL and not null:
        raise FieldDefinitionError("When SET_NULL is enabled, null must be True.")

    # If SET_NULL is enabled for on_update, the field must be nullable.
    if on_update and (on_update == SET_NULL and not null):
        raise FieldDefinitionError("When SET_NULL is enabled, null must be True.")
    related_name = kwargs.get("related_name", "")

    # related_name can be None or False, which is tolerated.
    if related_name and not isinstance(related_name, str):
        raise FieldDefinitionError("related_name must be a string.")

    # Convert related_name to lowercase if it's a non-empty string.
    if related_name:
        kwargs["related_name"] = related_name.lower()

@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), {}))

@classmethod
def modify_input(
    cls,
    field_obj: BaseFieldType,
    name: str,
    kwargs: dict[str, Any],
    original_fn: Any = None,
) -> None:
    """
    Modifies the input `kwargs` during model initialization.

    If the field name is not present in `kwargs` and the phase is `init_db`,
    it initializes the field with an empty list. This ensures that the field
    is always a list, ready to store `ModelRef` instances.
    """
    phase = CURRENT_PHASE.get()
    # If the field is not in kwargs and we are in the 'init_db' phase,
    # initialize it as an empty list.
    if name not in kwargs and phase == "init_db":
        kwargs[name] = []

def embed_field(
    self,
    prefix: str,
    new_fieldname: str,
    owner: type[BaseModelType] | None = None,
    parent: BaseFieldType | None = None,
) -> BaseFieldType | None:
    """
    Placeholder for embedding logic. `RefForeignKey` typically does not
    embed directly into queries in the same way as traditional foreign keys,
    as its values are often handled through the associated `__related_name__`.
    """
    return None

@classmethod
async def post_save_callback(
    cls,
    field_obj: BaseFieldType,
    value: list | None,
    is_update: bool,
    original_fn: Any = None,
) -> None:
    """
    Asynchronous post-save callback for `RefForeignKey`.

    This callback is executed after a model instance containing this field is saved.
    It processes the list of `ModelRef` instances provided in `value` and
    establishes the actual relationships by adding them to the related field
    (determined by the `__related_name__` on the `ModelRef`).

    Args:
        field_obj (BaseFieldType): The field object itself.
        value (list | None): The list of `ModelRef` instances or dictionaries
                             representing the related models to be associated.
        is_update (bool): `True` if the operation is an update, `False` for an insert.
        original_fn (Any): The original callback function (if any).
    """
    instance = CURRENT_MODEL_INSTANCE.get()
    if not value:
        return

    model_ref = field_obj.to
    # Get the actual relationship field on the instance's meta.
    relation_field = instance.meta.fields[model_ref.__related_name__]
    extra_params = {}

    try:
        # Determine the target model class.
        # This handles both ManyToMany and ForeignKey relationships.
        target_model_class = relation_field.target
    except AttributeError:
        # For reverse relationships (e.g., reverse ManyToMany), get from related_from.
        target_model_class = relation_field.related_from

    # If it's not a Many-to-Many relationship (i.e., a ForeignKey),
    # prepare to set the foreign key on the target model.
    if not relation_field.is_m2m:
        extra_params[relation_field.foreign_key.name] = instance

    # Get the actual relation object from the instance (e.g., a ManyRelation or one-to-one).
    relation = getattr(instance, model_ref.__related_name__)

    # Process each item in the value list.
    while value:
        instance_or_dict: dict | ModelRef = value.pop()
        # If it's a dictionary, convert it to a ModelRef instance.
        if isinstance(instance_or_dict, dict):
            instance_or_dict = model_ref(**instance_or_dict)

        # Create an instance of the target model class from the ModelRef data.
        # Exclude '__related_name__' from the model_dump as it's an internal ModelRef attribute.
        model = target_model_class(
            **cast("ModelRef", instance_or_dict).model_dump(exclude={"__related_name__"}),
            **extra_params,  # Add foreign key if applicable.
        )
        # Add the created model to the relation. This persists the relationship.
        await relation.add(model)