RefForeignKey class

edgy.fields.RefForeignKey

Bases: ForeignKeyFieldFactory, list

A specialized foreign key field designed to work with ModelRef classes.

This field allows a model to establish a relationship with a ModelRef, which represents a reference to another model without necessarily defining a direct foreign key column in the database. It is particularly useful for handling generic relationships or when the related model's details are managed through a separate "through" model or a ModelRef.

It enforces that the to argument is an instance of ModelRef and that the ModelRef subclass has a __related_name__ defined.

ATTRIBUTE	DESCRIPTION
field_type	The Python type representing the field (always list), as it can hold a list of ModelRef instances. TYPE: list
field_bases	The base classes for this field, including BaseRefForeignKey. TYPE: tuple

methods_overwritable_by_factory class-attribute instance-attribute

methods_overwritable_by_factory = frozenset(default_methods_overwritable_by_factory)

field_type class-attribute instance-attribute

field_type = list

field_bases class-attribute instance-attribute

field_bases = (BaseRefForeignKey,)

build_field classmethod

build_field(**kwargs)

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.

PARAMETER	DESCRIPTION
**kwargs	Keyword arguments for configuring the field. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
BaseFieldType	The newly constructed and configured field instance. TYPE: BaseFieldType

Source code in edgy/core/db/fields/factories.py

@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

overwrite_methods classmethod

overwrite_methods(field_obj)

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.

PARAMETER	DESCRIPTION
field_obj	The field instance whose methods are to be overwritten. TYPE: BaseFieldType

Source code in edgy/core/db/fields/factories.py

@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__,
            )

repack classmethod

repack(field_obj)

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.

PARAMETER	DESCRIPTION
field_obj	The field instance whose methods are to be repacked. TYPE: BaseFieldType

Source code in edgy/core/db/fields/factories.py

@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__,
                )

validate classmethod

validate(kwargs)

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.

PARAMETER	DESCRIPTION
kwargs	A dictionary of parameters for the foreign key field. TYPE: dict[str, Any]

RAISES	DESCRIPTION
FieldDefinitionError	If any foreign key validation rule is violated.

Source code in edgy/core/db/fields/factories.py

@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()

get_constraints classmethod

get_constraints(kwargs)

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.

PARAMETER	DESCRIPTION
kwargs	Keyword arguments provided during field creation. TYPE: dict[str, Any]

RETURNS	DESCRIPTION
Sequence[Any]	Sequence[Any]: A sequence of constraint objects.

Source code in edgy/core/db/fields/factories.py

@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 []

get_column_type classmethod

get_column_type(kwargs)

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.

PARAMETER	DESCRIPTION
kwargs	Keyword arguments provided during field creation. TYPE: dict[str, Any]

RETURNS	DESCRIPTION
Any	The SQLAlchemy column type or None if it's a meta field. TYPE: Any

Source code in edgy/core/db/fields/factories.py

@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

get_pydantic_type classmethod

get_pydantic_type(kwargs)

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.

PARAMETER	DESCRIPTION
kwargs	Keyword arguments provided during field creation. TYPE: dict[str, Any]

RETURNS	DESCRIPTION
Any	The Pydantic type associated with the field. TYPE: Any

Source code in edgy/core/db/fields/factories.py

@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

_get_field_cls cached staticmethod

_get_field_cls()

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.

PARAMETER	DESCRIPTION
cls	The FieldFactory instance itself. TYPE: FieldFactory

RETURNS	DESCRIPTION
BaseFieldType	The dynamically created and cached field class. TYPE: BaseFieldType

Source code in edgy/core/db/fields/factories.py

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

modify_input classmethod

modify_input(field_obj, name, kwargs, original_fn=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.

Source code in edgy/core/db/fields/ref_foreign_key.py

@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] = []

embed_field

embed_field(prefix, new_fieldname, owner=None, parent=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__.

Source code in edgy/core/db/fields/ref_foreign_key.py

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

post_save_callback async classmethod

post_save_callback(field_obj, value, is_update, original_fn=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).

PARAMETER	DESCRIPTION
field_obj	The field object itself. TYPE: BaseFieldType
value	The list of ModelRef instances or dictionaries representing the related models to be associated. TYPE: list | None
is_update	True if the operation is an update, False for an insert. TYPE: bool
original_fn	The original callback function (if any). TYPE: Any DEFAULT: None

Source code in edgy/core/db/fields/ref_foreign_key.py

@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)