Skip to content

RefForeignKey class

edgy.RefForeignKey

Bases: ForeignKeyFieldFactory, list

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 = (BaseField)

build_field classmethod

build_field(**kwargs)
PARAMETER DESCRIPTION
**kwargs

TYPE: Any DEFAULT: {}

Source code in edgy/core/db/fields/factories.py
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
@classmethod
def build_field(cls, **kwargs: Any) -> BaseFieldType:
    column_type = cls.get_column_type(**kwargs)
    pydantic_type = cls.get_pydantic_type(**kwargs)
    constraints = cls.get_constraints(**kwargs)

    new_field = cls._get_field_cls(cls)

    new_field_obj: BaseFieldType = new_field(  # type: ignore
        field_type=pydantic_type,
        annotation=pydantic_type,
        column_type=column_type,
        constraints=constraints,
        factory=cls,
        **kwargs,
    )
    cls.overwrite_methods(new_field_obj)
    return new_field_obj

overwrite_methods classmethod

overwrite_methods(field_obj)

Called in metaclasses

PARAMETER DESCRIPTION
field_obj

TYPE: BaseFieldType

Source code in edgy/core/db/fields/factories.py
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
@classmethod
def overwrite_methods(cls, field_obj: BaseFieldType) -> None:
    """Called in metaclasses"""
    for key in dir(cls):
        if key in cls.methods_overwritable_by_factory and hasattr(cls, key):
            fn = getattr(cls, key)
            original_fn = getattr(field_obj, key, None)
            # use original func, not the wrapper
            if hasattr(fn, "__func__"):
                fn = fn.__func__

            # fix classmethod, prevent injection of self or class
            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)
PARAMETER DESCRIPTION
field_obj

TYPE: BaseFieldType

Source code in edgy/core/db/fields/factories.py
106
107
108
109
110
111
112
113
114
115
116
117
118
119
@classmethod
def repack(cls, field_obj: BaseFieldType) -> None:
    for key in dir(cls):
        if key in cls.methods_overwritable_by_factory and hasattr(cls, key):
            packed_fn = getattr(field_obj, key, None)
            if packed_fn is not None and hasattr(packed_fn, "func"):
                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)

default validation useful for one_to_one and foreign_key

PARAMETER DESCRIPTION
kwargs

TYPE: dict[str, Any]

Source code in edgy/core/db/fields/factories.py
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
@classmethod
def validate(cls, kwargs: dict[str, Any]) -> None:
    """default validation useful for one_to_one and foreign_key"""
    on_update = kwargs.get("on_update", CASCADE)
    on_delete = kwargs.get("on_delete", RESTRICT)
    kwargs.setdefault("null", False)
    null = kwargs["null"]

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

    if on_delete == SET_NULL and not null:
        raise FieldDefinitionError("When SET_NULL is enabled, null must be True.")

    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", "")

    # tolerate None and False
    if related_name and not isinstance(related_name, str):
        raise FieldDefinitionError("related_name must be a string.")

    if related_name:
        kwargs["related_name"] = related_name.lower()

get_constraints classmethod

get_constraints(**kwargs)

Returns the propery column type for the field, None for Metafields

PARAMETER DESCRIPTION
**kwargs

TYPE: Any DEFAULT: {}

Source code in edgy/core/db/fields/factories.py
129
130
131
132
@classmethod
def get_constraints(cls, **kwargs: Any) -> Sequence[Any]:
    """Returns the propery column type for the field, None for Metafields"""
    return []

get_column_type classmethod

get_column_type(**kwargs)

Returns the propery column type for the field, None for Metafields

PARAMETER DESCRIPTION
**kwargs

TYPE: Any DEFAULT: {}

Source code in edgy/core/db/fields/factories.py
134
135
136
137
@classmethod
def get_column_type(cls, **kwargs: Any) -> Any:
    """Returns the propery column type for the field, None for Metafields"""
    return None

get_pydantic_type classmethod

get_pydantic_type(**kwargs)

Returns the type for pydantic

PARAMETER DESCRIPTION
**kwargs

TYPE: Any DEFAULT: {}

Source code in edgy/core/db/fields/factories.py
139
140
141
142
@classmethod
def get_pydantic_type(cls, **kwargs: Any) -> Any:
    """Returns the type for pydantic"""
    return cls.field_type

_get_field_cls cached staticmethod

_get_field_cls()
Source code in edgy/core/db/fields/factories.py
144
145
146
147
@staticmethod
@lru_cache(None)
def _get_field_cls(cls: "FieldFactory") -> BaseFieldType:
    return cast(BaseFieldType, type(cls.__name__, cast(Any, cls.field_bases), {}))

modify_input classmethod

modify_input(field_obj, name, kwargs, original_fn=None)
PARAMETER DESCRIPTION
field_obj

TYPE: BaseFieldType

name

TYPE: str

kwargs

TYPE: dict[str, Any]

original_fn

TYPE: Any DEFAULT: None

Source code in edgy/core/db/fields/ref_foreign_key.py
23
24
25
26
27
28
29
30
31
32
33
34
@classmethod
def modify_input(
    cls,
    field_obj: "BaseFieldType",
    name: str,
    kwargs: dict[str, Any],
    original_fn: Any = None,
) -> None:
    phase = CURRENT_PHASE.get()
    # we are empty
    if name not in kwargs and phase == "init_db":
        kwargs[name] = []

embed_field

embed_field(prefix, new_fieldname, owner=None, parent=None)
PARAMETER DESCRIPTION
prefix

TYPE: str

new_fieldname

TYPE: str

owner

TYPE: Optional[type[BaseModelType]] DEFAULT: None

parent

TYPE: Optional[BaseFieldType] DEFAULT: None

Source code in edgy/core/db/fields/ref_foreign_key.py
36
37
38
39
40
41
42
43
def embed_field(
    self,
    prefix: str,
    new_fieldname: str,
    owner: Optional[type["BaseModelType"]] = None,
    parent: Optional["BaseFieldType"] = None,
) -> Optional["BaseFieldType"]:
    return None

is_class_and_subclass classmethod

is_class_and_subclass(value, _type)
PARAMETER DESCRIPTION
value

TYPE: Any

_type

TYPE: Any

Source code in edgy/core/db/fields/ref_foreign_key.py
45
46
47
48
49
50
51
52
53
54
55
56
@classmethod
def is_class_and_subclass(cls, value: typing.Any, _type: typing.Any) -> bool:
    original = get_origin(value)
    if not original and not isclass(value):
        return False

    try:
        if original:
            return original and issubclass(original, _type)
        return issubclass(value, _type)
    except TypeError:
        return False

post_save_callback async classmethod

post_save_callback(obj, value, instance, force_insert, original_fn=None)
PARAMETER DESCRIPTION
obj

TYPE: BaseFieldType

value

TYPE: Optional[list]

instance

TYPE: BaseModelType

force_insert

TYPE: bool

original_fn

TYPE: Any DEFAULT: None

Source code in edgy/core/db/fields/ref_foreign_key.py
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
@classmethod
async def post_save_callback(
    cls,
    obj: "BaseFieldType",
    value: Optional[list],
    instance: "BaseModelType",
    force_insert: bool,
    original_fn: Any = None,
) -> None:
    if not value:
        return

    relation_field = instance.meta.fields[obj.to.__related_name__]
    extra_params = {}
    try:
        # m2m or foreign key
        target_model_class = relation_field.target
    except AttributeError:
        # reverse m2m or foreign key
        target_model_class = relation_field.related_from
    if not relation_field.is_m2m:
        # sometimes the foreign key is required, so set it already
        extra_params[relation_field.foreign_key.name] = instance
    relation = getattr(instance, obj.to.__related_name__)
    while value:
        instance_or_dict: Union[dict, ModelRef] = value.pop()
        if isinstance(instance_or_dict, dict):
            instance_or_dict = obj.to(**instance_or_dict)
        model = target_model_class(
            **cast("ModelRef", instance_or_dict).model_dump(exclude={"__related_name__"}),
            **extra_params,
        )
        # we are in a relationship field
        await relation.add(model)