Skip to content

Model class

edgy.Model

Model(*args, **kwargs)

Bases: ModelRow, DeclarativeMixin

Representation of an Edgy Model.

This also means it can generate declarative SQLAlchemy models from anywhere by calling the Model.declarative() function.

Example

import edgy
from edgy import Database, Registry

database = Database("sqlite:///db.sqlite")
models = Registry(database=database)


class User(edgy.Model):
    '''
    The User model to be created in the database as a table
    If no name is provided the in Meta class, it will generate
    a "users" table for you.
    '''

    id: int = edgy.IntegerField(primary_key=True)
    is_active: bool = edgy.BooleanField(default=False)

    class Meta:
        registry = models
PARAMETER DESCRIPTION
*args

TYPE: Any DEFAULT: ()

**kwargs

TYPE: Any DEFAULT: {}

Source code in edgy/core/db/models/base.py
48
49
50
51
52
53
def __init__(self, *args: Any, **kwargs: Any) -> None:
    super().__init__(**kwargs)
    model_references = self.setup_model_references_from_kwargs(kwargs)
    values = self.setup_model_fields_from_kwargs(kwargs)
    self.__dict__ = values
    self.__model_references__ = model_references

parent class-attribute

parent

is_proxy_model class-attribute

is_proxy_model = False

query class-attribute

query = Manager()

meta class-attribute

meta = MetaInfo(None)

Meta class-attribute

Meta = DescriptiveMeta()

__proxy_model__ class-attribute

__proxy_model__ = None

__db_model__ class-attribute

__db_model__ = False

__raw_query__ class-attribute

__raw_query__ = None

__using_schema__ class-attribute

__using_schema__ = None

__model_references__ class-attribute instance-attribute

__model_references__ = model_references

pk property writable

pk

raw_query property writable

raw_query

proxy_model cached property

proxy_model

signals cached property

signals

table property writable

table

declarative classmethod

declarative()
Source code in edgy/core/db/models/mixins/generics.py
15
16
17
@classmethod
def declarative(cls) -> Any:
    return cls.generate_model_declarative()

generate_model_declarative classmethod

generate_model_declarative()

Transforms a core Edgy table into a Declarative model table.

Source code in edgy/core/db/models/mixins/generics.py
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
@classmethod
def generate_model_declarative(cls) -> Any:
    """
    Transforms a core Edgy table into a Declarative model table.
    """
    Base = cls.meta.registry.declarative_base

    # Build the original table
    fields = {"__table__": cls.table}

    # Generate base
    model_table = type(cls.__name__, (Base,), fields)

    # Make sure if there are foreignkeys, builds the relationships
    for column in cls.table.columns:
        if not column.foreign_keys:
            continue

        # Maps the relationships with the foreign keys and related names
        field = cls.fields.get(column.name)
        to = field.to.__name__ if inspect.isclass(field.to) else field.to
        mapped_model: Mapped[to] = relationship(to)  # type: ignore

        # Adds to the current model
        model_table.__mapper__.add_property(f"{column.name}_relation", mapped_model)

    return model_table

_extract_model_references

_extract_model_references(extracted_values, model_class)

Exracts any possible model references from the EdgyModel and returns a dictionary.

PARAMETER DESCRIPTION
extracted_values

TYPE: Any

model_class

TYPE: Optional[Type[Model]]

Source code in edgy/core/utils/models.py
62
63
64
65
66
67
68
69
70
71
72
73
def _extract_model_references(
    self, extracted_values: Any, model_class: Optional[Type["Model"]]
) -> Any:
    """
    Exracts any possible model references from the EdgyModel and returns a dictionary.
    """
    model_references = {
        name: extracted_values.get(name, None)
        for name in model_class.meta.model_references.keys()  # type: ignore
        if extracted_values.get(name)
    }
    return model_references

_extract_values_from_field

_extract_values_from_field(extracted_values, model_class=None, is_update=False)

Extracts all the deffault values from the given fields and returns the raw value corresponding to each field.

PARAMETER DESCRIPTION
extracted_values

TYPE: Any

model_class

TYPE: Optional[Type[Model]] DEFAULT: None

is_update

TYPE: bool DEFAULT: False

Source code in edgy/core/utils/models.py
 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
104
105
106
107
108
109
def _extract_values_from_field(
    self,
    extracted_values: Any,
    model_class: Optional[Type["Model"]] = None,
    is_update: bool = False,
) -> Any:
    """
    Extracts all the deffault values from the given fields and returns the raw
    value corresponding to each field.
    """
    model_cls = model_class or self
    validated = {}
    for name, field in model_cls.fields.items():  # type: ignore
        if field.read_only:
            if field.has_default():
                if not is_update:
                    validated[name] = field.get_default_value()
                else:
                    # For datetimes with `auto_now` and `auto_now_add`
                    if not _has_auto_now_add(field):
                        validated[name] = field.get_default_value()
            continue

        if name not in extracted_values:
            if field.has_default():
                validated[name] = field.get_default_value()
            continue

        item = extracted_values[name]
        value = field.check(item) if hasattr(field, "check") else None
        validated[name] = value

    # Update with any ModelRef
    validated.update(self._extract_model_references(extracted_values, model_cls))
    return validated

_extract_db_fields_from_model

_extract_db_fields_from_model(model_class)

Extacts all the db fields and excludes the related_names since those are simply relations.

PARAMETER DESCRIPTION
model_class

TYPE: Type[Model]

Source code in edgy/core/utils/models.py
111
112
113
114
115
116
117
def _extract_db_fields_from_model(self, model_class: Type["Model"]) -> Dict[Any, Any]:
    """
    Extacts all the db fields and excludes the related_names since those
    are simply relations.
    """
    related_names = model_class.meta.related_names
    return {k: v for k, v in model_class.model_fields.items() if k not in related_names}

_update_auto_now_fields

_update_auto_now_fields(values, fields)

Updates the auto_now fields

PARAMETER DESCRIPTION
values

TYPE: Any

fields

TYPE: Any

Source code in edgy/core/utils/models.py
41
42
43
44
45
46
47
48
def _update_auto_now_fields(self, values: Any, fields: Any) -> Any:
    """
    Updates the `auto_now` fields
    """
    for name, field in fields.items():
        if isinstance(field, Field) and _has_auto_now(field) and _is_datetime(field):
            values[name] = field.get_default_value()  # type: ignore
    return values

_resolve_value

_resolve_value(value)
PARAMETER DESCRIPTION
value

TYPE: Any

Source code in edgy/core/utils/models.py
50
51
52
53
54
55
56
57
58
def _resolve_value(self, value: typing.Any) -> typing.Any:
    if isinstance(value, dict):
        return dumps(
            value,
            option=OPT_SERIALIZE_NUMPY | OPT_OMIT_MICROSECONDS,
        ).decode("utf-8")
    elif isinstance(value, Enum):
        return value.name
    return value

setup_model_references_from_kwargs

setup_model_references_from_kwargs(kwargs)

Loops and setup the kwargs of the model

PARAMETER DESCRIPTION
kwargs

TYPE: Any

Source code in edgy/core/db/models/base.py
55
56
57
58
59
60
def setup_model_references_from_kwargs(self, kwargs: Any) -> Any:
    """
    Loops and setup the kwargs of the model
    """
    model_references = {k: v for k, v in kwargs.items() if k in self.meta.model_references}
    return model_references

setup_model_fields_from_kwargs

setup_model_fields_from_kwargs(kwargs)

Loops and setup the kwargs of the model

PARAMETER DESCRIPTION
kwargs

TYPE: Any

Source code in edgy/core/db/models/base.py
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def setup_model_fields_from_kwargs(self, kwargs: Any) -> Any:
    """
    Loops and setup the kwargs of the model
    """
    if "pk" in kwargs:
        kwargs[self.pkname] = kwargs.pop("pk")

    kwargs = {
        k: v
        for k, v in kwargs.items()
        if k in self.meta.fields_mapping and k not in self.meta.model_references
    }

    for key, value in kwargs.items():
        if key not in self.fields:
            if not hasattr(self, key):
                raise ValueError(f"Invalid keyword {key} for class {self.__class__.__name__}")

        # Set model field and add to the kwargs dict
        edgy_setattr(self, key, value)
        kwargs[key] = value
    return kwargs

generate_proxy_model classmethod

generate_proxy_model()

Generates a proxy model for each model. This proxy model is a simple shallow copy of the original model being generated.

Source code in edgy/core/db/models/base.py
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
@classmethod
def generate_proxy_model(cls) -> Type["Model"]:
    """
    Generates a proxy model for each model. This proxy model is a simple
    shallow copy of the original model being generated.
    """
    if cls.__proxy_model__:
        return cls.__proxy_model__

    fields = {key: copy.copy(field) for key, field in cls.fields.items()}
    proxy_model = ProxyModel(
        name=cls.__name__,
        module=cls.__module__,
        metadata=cls.meta,
        definitions=fields,
    )

    proxy_model.build()
    generify_model_fields(proxy_model.model)
    return proxy_model.model

model_dump

model_dump(show_pk=False, **kwargs)

An updated version of the model dump if the primary key is not provided.

PARAMETER DESCRIPTION
show_pk

TYPE: bool DEFAULT: False

**kwargs

TYPE: Any DEFAULT: {}

PARAMETER DESCRIPTION
show_pk

bool - Enforces showing the primary key in the model_dump.

TYPE: bool DEFAULT: False

Source code in edgy/core/db/models/base.py
146
147
148
149
150
151
152
153
154
155
156
def model_dump(self, show_pk: bool = False, **kwargs: Any) -> Dict[str, Any]:
    """
    An updated version of the model dump if the primary key is not provided.

    Args:
        show_pk: bool - Enforces showing the primary key in the model_dump.
    """
    model = super().model_dump(**kwargs)
    if self.pkname not in model and show_pk:
        model = {**{self.pkname: self.pk}, **model}
    return model

build classmethod

build(schema=None)

Builds the SQLAlchemy table representation from the loaded fields.

PARAMETER DESCRIPTION
schema

TYPE: Optional[str] DEFAULT: None

Source code in edgy/core/db/models/base.py
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
@classmethod
def build(cls, schema: Optional[str] = None) -> sqlalchemy.Table:
    """
    Builds the SQLAlchemy table representation from the loaded fields.
    """
    tablename: str = cls.meta.tablename  # type: ignore
    metadata: sqlalchemy.MetaData = cast("sqlalchemy.MetaData", cls.meta.registry._metadata)  # type: ignore
    metadata.schema = schema

    unique_together = cls.meta.unique_together
    index_constraints = cls.meta.indexes

    columns = []
    for name, field in cls.fields.items():
        columns.append(field.get_column(name))

    # Handle the uniqueness together
    uniques = []
    for field in unique_together or []:
        unique_constraint = cls._get_unique_constraints(field)
        uniques.append(unique_constraint)

    # Handle the indexes
    indexes = []
    for field in index_constraints or []:
        index = cls._get_indexes(field)
        indexes.append(index)

    return sqlalchemy.Table(
        tablename, metadata, *columns, *uniques, *indexes, extend_existing=True  # type: ignore
    )

_get_unique_constraints classmethod

_get_unique_constraints(columns)

Returns the unique constraints for the model.

The columns must be a a list, tuple of strings or a UniqueConstraint object.

:return: Model UniqueConstraint.

PARAMETER DESCRIPTION
columns

TYPE: Sequence

Source code in edgy/core/db/models/base.py
190
191
192
193
194
195
196
197
198
199
200
201
202
203
@classmethod
def _get_unique_constraints(cls, columns: Sequence) -> Optional[sqlalchemy.UniqueConstraint]:
    """
    Returns the unique constraints for the model.

    The columns must be a a list, tuple of strings or a UniqueConstraint object.

    :return: Model UniqueConstraint.
    """
    if isinstance(columns, str):
        return sqlalchemy.UniqueConstraint(columns)
    elif isinstance(columns, UniqueConstraint):
        return sqlalchemy.UniqueConstraint(*columns.fields, name=columns.name)
    return sqlalchemy.UniqueConstraint(*columns)

_get_indexes classmethod

_get_indexes(index)

Creates the index based on the Index fields

PARAMETER DESCRIPTION
index

TYPE: Index

Source code in edgy/core/db/models/base.py
205
206
207
208
209
210
@classmethod
def _get_indexes(cls, index: Index) -> Optional[sqlalchemy.Index]:
    """
    Creates the index based on the Index fields
    """
    return sqlalchemy.Index(index.name, *index.fields)  # type: ignore

update_from_dict

update_from_dict(dict_values)

Updates the current model object with the new fields and possible model_references

PARAMETER DESCRIPTION
dict_values

TYPE: Dict[str, Any]

Source code in edgy/core/db/models/base.py
212
213
214
215
216
def update_from_dict(self, dict_values: Dict[str, Any]) -> Self:
    """Updates the current model object with the new fields and possible model_references"""
    for key, value in dict_values.items():
        setattr(self, key, value)
    return self

extract_db_model_references

extract_db_model_references()

Extracts all the model references (ModelRef) from the model

Source code in edgy/core/db/models/base.py
218
219
220
221
222
223
def extract_db_model_references(self) -> Dict[str, Any]:
    """
    Extracts all the model references (ModelRef) from the model
    """
    related_names = self.meta.related_names
    return {k: v for k, v in self.__model_references__.items() if k not in related_names}

extract_db_fields

extract_db_fields()

Extacts all the db fields and excludes the related_names since those are simply relations.

Source code in edgy/core/db/models/base.py
225
226
227
228
229
230
231
232
233
234
235
def extract_db_fields(self) -> Dict[str, Any]:
    """
    Extacts all the db fields and excludes the related_names since those
    are simply relations.
    """
    related_names = self.meta.related_names
    return {
        k: v
        for k, v in self.__dict__.items()
        if k not in related_names and k not in EXCLUDED_LOOKUP
    }

get_instance_name

get_instance_name()

Returns the name of the class in lowercase.

Source code in edgy/core/db/models/base.py
237
238
239
240
241
def get_instance_name(self) -> str:
    """
    Returns the name of the class in lowercase.
    """
    return self.__class__.__name__.lower()

__setattr__

__setattr__(key, value)
PARAMETER DESCRIPTION
key

TYPE: Any

value

TYPE: Any

Source code in edgy/core/db/models/base.py
243
244
245
246
247
248
249
250
def __setattr__(self, key: Any, value: Any) -> Any:
    if key in self.fields:
        field = self.fields[key]
        if isinstance(field, BaseManyToManyForeignKeyField):
            value = getattr(self, settings.many_to_many_relation.format(key=key))
        else:
            value = self.fields[key].expand_relationship(value)
    edgy_setattr(self, key, value)

__get_instance_values

__get_instance_values(instance)
PARAMETER DESCRIPTION
instance

TYPE: Any

Source code in edgy/core/db/models/base.py
252
253
254
255
256
257
def __get_instance_values(self, instance: Any) -> Set[Any]:
    return {
        v
        for k, v in instance.__dict__.items()
        if k in instance.fields.keys() and v is not None
    }

__eq__

__eq__(other)
PARAMETER DESCRIPTION
other

TYPE: Any

Source code in edgy/core/db/models/base.py
259
260
261
262
263
264
265
266
267
def __eq__(self, other: Any) -> bool:
    if self.__class__ != other.__class__:
        return False

    original = self.__get_instance_values(instance=self)
    other_values = self.__get_instance_values(instance=other)
    if original != other_values:
        return False
    return True

from_sqla_row classmethod

from_sqla_row(row, select_related=None, prefetch_related=None, is_only_fields=False, only_fields=None, is_defer_fields=False, exclude_secrets=False, using_schema=None)

Class method to convert a SQLAlchemy Row result into a EdgyModel row type.

Looping through select_related fields if the query comes from a select_related operation. Validates if exists the select_related and related_field inside the models.

When select_related and related_field exist for the same field being validated, the related field is ignored as it won't override the value already collected from the select_related.

If there is no select_related, then goes through the related field where it should only return the instance of the the ForeignKey with the ID, making it lazy loaded.

:return: Model class.

PARAMETER DESCRIPTION
row

TYPE: Row

select_related

TYPE: Optional[Sequence[Any]] DEFAULT: None

prefetch_related

TYPE: Optional[Sequence[Prefetch]] DEFAULT: None

is_only_fields

TYPE: bool DEFAULT: False

only_fields

TYPE: Sequence[str] DEFAULT: None

is_defer_fields

TYPE: bool DEFAULT: False

exclude_secrets

TYPE: bool DEFAULT: False

using_schema

TYPE: Union[str, None] DEFAULT: None

Source code in edgy/core/db/models/row.py
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 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
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
@classmethod
def from_sqla_row(
    cls,
    row: Row,
    select_related: Optional[Sequence[Any]] = None,
    prefetch_related: Optional[Sequence["Prefetch"]] = None,
    is_only_fields: bool = False,
    only_fields: Sequence[str] = None,
    is_defer_fields: bool = False,
    exclude_secrets: bool = False,
    using_schema: Union[str, None] = None,
) -> Optional[Type["Model"]]:
    """
    Class method to convert a SQLAlchemy Row result into a EdgyModel row type.

    Looping through select_related fields if the query comes from a select_related operation.
    Validates if exists the select_related and related_field inside the models.

    When select_related and related_field exist for the same field being validated, the related
    field is ignored as it won't override the value already collected from the select_related.

    If there is no select_related, then goes through the related field where it **should**
    only return the instance of the the ForeignKey with the ID, making it lazy loaded.

    :return: Model class.
    """
    item: Dict[str, Any] = {}
    select_related = select_related or []
    prefetch_related = prefetch_related or []
    secret_fields = (
        [name for name, field in cls.fields.items() if field.secret] if exclude_secrets else []
    )

    for related in select_related:
        if "__" in related:
            first_part, remainder = related.split("__", 1)
            try:
                model_cls = cls.fields[first_part].target
            except KeyError:
                model_cls = getattr(cls, first_part).related_from
            item[first_part] = model_cls.from_sqla_row(
                row,
                select_related=[remainder],
                prefetch_related=prefetch_related,
                exclude_secrets=exclude_secrets,
                using_schema=using_schema,
            )
        else:
            try:
                model_cls = cls.fields[related].target
            except KeyError:
                model_cls = getattr(cls, related).related_from
            item[related] = model_cls.from_sqla_row(
                row, exclude_secrets=exclude_secrets, using_schema=using_schema
            )

    # Populate the related names
    # Making sure if the model being queried is not inside a select related
    # This way it is not overritten by any value
    for related, foreign_key in cls.meta.foreign_key_fields.items():
        ignore_related: bool = cls.__should_ignore_related_name(related, select_related)
        if ignore_related:
            continue

        model_related = foreign_key.target

        # Apply the schema to the model
        model_related = cls.__apply_schema(model_related, using_schema)

        child_item = {}
        for column in model_related.table.columns:
            if column.name in secret_fields or related in secret_fields:
                continue
            if column.name not in cls.fields.keys():
                continue
            elif related not in child_item:
                if row[related] is not None:
                    child_item[column.name] = row[related]

        # Make sure we generate a temporary reduced model
        # For the related fields. We simply chnage the structure of the model
        # and rebuild it with the new fields.
        if related not in secret_fields:
            item[related] = model_related.proxy_model(**child_item)

    # Check for the only_fields
    if is_only_fields or is_defer_fields:
        mapping_fields = (
            [str(field) for field in only_fields] if is_only_fields else list(row.keys())  # type: ignore
        )

        for column, value in row._mapping.items():
            if column in secret_fields:
                continue
            # Making sure when a table is reflected, maps the right fields of the ReflectModel
            if column not in mapping_fields:
                continue

            if column not in item:
                item[column] = value

        # We need to generify the model fields to make sure we can populate the
        # model without mandatory fields
        model = cast("Type[Model]", cls.proxy_model(**item))

        # Apply the schema to the model
        model = cls.__apply_schema(model, using_schema)

        model = cls.__handle_prefetch_related(
            row=row, model=model, prefetch_related=prefetch_related
        )
        return model
    else:
        # Pull out the regular column values.
        for column in cls.table.columns:
            # Making sure when a table is reflected, maps the right fields of the ReflectModel
            if column.name in secret_fields:
                continue
            if column.name not in cls.fields.keys():
                continue
            elif column.name not in item:
                item[column.name] = row[column]

    model = (
        cast("Type[Model]", cls(**item))
        if not exclude_secrets
        else cast("Type[Model]", cls.proxy_model(**item))
    )

    # Apply the schema to the model
    model = cls.__apply_schema(model, using_schema)

    # Handle prefetch related fields.
    model = cls.__handle_prefetch_related(
        row=row, model=model, prefetch_related=prefetch_related
    )
    return model

__apply_schema classmethod

__apply_schema(model, schema=None)
PARAMETER DESCRIPTION
model

TYPE: Type[Model]

schema

TYPE: Optional[str] DEFAULT: None

Source code in edgy/core/db/models/row.py
156
157
158
159
160
161
162
@classmethod
def __apply_schema(cls, model: Type["Model"], schema: Optional[str] = None) -> Type["Model"]:
    # Apply the schema to the model
    if schema is not None:
        model.table = model.build(schema)  # type: ignore
        model.proxy_model.table = model.proxy_model.build(schema)  # type: ignore
    return model
__should_ignore_related_name(related_name, select_related)

Validates if it should populate the related field if select related is not considered.

PARAMETER DESCRIPTION
related_name

TYPE: str

select_related

TYPE: Sequence[str]

Source code in edgy/core/db/models/row.py
164
165
166
167
168
169
170
171
172
173
174
175
@classmethod
def __should_ignore_related_name(
    cls, related_name: str, select_related: Sequence[str]
) -> bool:
    """
    Validates if it should populate the related field if select related is not considered.
    """
    for related_field in select_related:
        fields = related_field.split("__")
        if related_name in fields:
            return True
    return False
__handle_prefetch_related(row, model, prefetch_related, parent_cls=None, original_prefetch=None, is_nested=False)

Handles any prefetch related scenario from the model. Loads in advance all the models needed for a specific record

Recursively checks for the related field and validates if there is any conflicting attribute. If there is, a QuerySetError is raised.

PARAMETER DESCRIPTION
row

TYPE: Row

model

TYPE: Type[Model]

prefetch_related

TYPE: Sequence[Prefetch]

parent_cls

TYPE: Optional[Type[Model]] DEFAULT: None

original_prefetch

TYPE: Optional[Prefetch] DEFAULT: None

is_nested

TYPE: bool DEFAULT: False

Source code in edgy/core/db/models/row.py
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
@classmethod
def __handle_prefetch_related(
    cls,
    row: Row,
    model: Type["Model"],
    prefetch_related: Sequence["Prefetch"],
    parent_cls: Optional[Type["Model"]] = None,
    original_prefetch: Optional["Prefetch"] = None,
    is_nested: bool = False,
) -> Type["Model"]:
    """
    Handles any prefetch related scenario from the model.
    Loads in advance all the models needed for a specific record

    Recursively checks for the related field and validates if there is any conflicting
    attribute. If there is, a `QuerySetError` is raised.
    """
    if not parent_cls:
        parent_cls = model

    for related in prefetch_related:
        if not original_prefetch:
            original_prefetch = related

        if original_prefetch and not is_nested:
            original_prefetch = related

        # Check for conflicting names
        # If to_attr has the same name of any
        if hasattr(parent_cls, original_prefetch.to_attr):
            raise QuerySetError(
                f"Conflicting attribute to_attr='{original_prefetch.related_name}' with '{original_prefetch.to_attr}' in {parent_cls.__class__.__name__}"
            )

        if "__" in related.related_name:
            first_part, remainder = related.related_name.split("__", 1)
            model_cls = cls.meta.related_fields[first_part].related_to

            # Build the new nested Prefetch object
            remainder_prefetch = related.__class__(
                related_name=remainder, to_attr=related.to_attr, queryset=related.queryset
            )

            # Recursively continue the process of handling the
            # new prefetch
            model_cls.__handle_prefetch_related(
                row,
                model,
                prefetch_related=[remainder_prefetch],
                original_prefetch=original_prefetch,
                parent_cls=model,
                is_nested=True,
            )

        # Check for individual not nested querysets
        elif related.queryset is not None and not is_nested:
            filter_by_pk = row[cls.pkname]
            extra = {f"{related.related_name}__id": filter_by_pk}
            related.queryset.extra = extra

            # Execute the queryset
            records = asyncio.get_event_loop().run_until_complete(
                cls.run_query(queryset=related.queryset)
            )
            setattr(model, related.to_attr, records)
        else:
            model_cls = getattr(cls, related.related_name).related_from
            records = cls.process_nested_prefetch_related(
                row,
                prefetch_related=related,
                original_prefetch=original_prefetch,
                parent_cls=model,
                queryset=original_prefetch.queryset,
            )

            setattr(model, related.to_attr, records)
    return model
process_nested_prefetch_related(row, prefetch_related, parent_cls, original_prefetch, queryset)

Processes the nested prefetch related names.

PARAMETER DESCRIPTION
row

TYPE: Row

prefetch_related

TYPE: Prefetch

parent_cls

TYPE: Type[Model]

original_prefetch

TYPE: Prefetch

queryset

TYPE: QuerySet

Source code in edgy/core/db/models/row.py
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
@classmethod
def process_nested_prefetch_related(
    cls,
    row: Row,
    prefetch_related: "Prefetch",
    parent_cls: Type["Model"],
    original_prefetch: "Prefetch",
    queryset: "QuerySet",
) -> List[Type["Model"]]:
    """
    Processes the nested prefetch related names.
    """
    query_split = original_prefetch.related_name.split("__")
    query_split.reverse()
    query_split.pop(query_split.index(prefetch_related.related_name))

    # Get the model to query related
    model_class = getattr(cls, prefetch_related.related_name).related_from

    # Get the foreign key name from the model_class
    foreign_key_name = model_class.meta.related_names_mapping[prefetch_related.related_name]

    # Insert as the entry point of the query
    query_split.insert(0, foreign_key_name)

    # Build new filter
    query = "__".join(query_split)

    # Extact foreign key value
    filter_by_pk = row[parent_cls.pkname]

    extra = {f"{query}__id": filter_by_pk}

    records = asyncio.get_event_loop().run_until_complete(
        cls.run_query(model_class, extra, queryset)
    )
    return records

run_query async classmethod

run_query(model=None, extra=None, queryset=None)

Runs a specific query against a given model with filters.

PARAMETER DESCRIPTION
model

TYPE: Optional[Type[Model]] DEFAULT: None

extra

TYPE: Optional[Dict[str, Any]] DEFAULT: None

queryset

TYPE: Optional[QuerySet] DEFAULT: None

Source code in edgy/core/db/models/row.py
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
@classmethod
async def run_query(
    cls,
    model: Optional[Type["Model"]] = None,
    extra: Optional[Dict[str, Any]] = None,
    queryset: Optional["QuerySet"] = None,
) -> Union[List[Type["Model"]], Any]:
    """
    Runs a specific query against a given model with filters.
    """

    if not queryset:
        return await model.query.filter(**extra)  # type: ignore

    if extra:
        queryset.extra = extra

    return await queryset

update async

update(**kwargs)

Update operation of the database fields.

PARAMETER DESCRIPTION
**kwargs

TYPE: Any DEFAULT: {}

Source code in edgy/core/db/models/model.py
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
async def update(self, **kwargs: Any) -> Any:
    """
    Update operation of the database fields.
    """
    await self.signals.pre_update.send(sender=self.__class__, instance=self)

    kwargs = self._update_auto_now_fields(kwargs, self.fields)
    pk_column = getattr(self.table.c, self.pkname)
    expression = self.table.update().values(**kwargs).where(pk_column == self.pk)
    await self.database.execute(expression)
    await self.signals.post_update.send(sender=self.__class__, instance=self)

    # Update the model instance.
    for key, value in kwargs.items():
        setattr(self, key, value)

    return self

delete async

delete()

Delete operation from the database

Source code in edgy/core/db/models/model.py
67
68
69
70
71
72
73
74
75
async def delete(self) -> None:
    """Delete operation from the database"""
    await self.signals.pre_delete.send(sender=self.__class__, instance=self)

    pk_column = getattr(self.table.c, self.pkname)
    expression = self.table.delete().where(pk_column == self.pk)
    await self.database.execute(expression)

    await self.signals.post_delete.send(sender=self.__class__, instance=self)

load async

load()
Source code in edgy/core/db/models/model.py
77
78
79
80
81
82
83
84
85
86
87
88
async def load(self) -> None:
    # Build the select expression.

    pk_column = getattr(self.table.c, self.pkname)
    expression = self.table.select().where(pk_column == self.pk)

    # Perform the fetch.
    row = await self.database.fetch_one(expression)

    # Update the instance.
    for key, value in dict(row._mapping).items():
        setattr(self, key, value)

_save async

_save(**kwargs)

Performs the save instruction.

PARAMETER DESCRIPTION
**kwargs

TYPE: Any DEFAULT: {}

Source code in edgy/core/db/models/model.py
90
91
92
93
94
95
96
97
98
99
async def _save(self, **kwargs: Any) -> "Model":
    """
    Performs the save instruction.
    """
    expression = self.table.insert().values(**kwargs)
    awaitable = await self.database.execute(expression)
    if not awaitable:
        awaitable = kwargs.get(self.pkname)
    edgy_setattr(self, self.pkname, awaitable)
    return self

save_model_references async

save_model_references(model_references, model_ref=None)

If there is any ModelRef declared in the model, it will generate the subsquent model reference records for that same model created.

PARAMETER DESCRIPTION
model_references

TYPE: Any

model_ref

TYPE: Any DEFAULT: None

Source code in edgy/core/db/models/model.py
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
async def save_model_references(self, model_references: Any, model_ref: Any = None) -> None:
    """
    If there is any ModelRef declared in the model, it will generate the subsquent model
    reference records for that same model created.
    """

    for reference in model_references:
        if isinstance(reference, dict):
            model: Type["Model"] = self.meta.model_references[model_ref].__model__  # type: ignore
        else:
            model: Type["Model"] = reference.__model__  # type: ignore

        if isinstance(model, str):
            model = self.meta.registry.models[model]  # type: ignore

        # If the reference did come in a dict format
        # It is necessary to convert into the original ModelRef.
        if isinstance(reference, dict):
            reference = self.meta.model_references[model_ref](**reference)  # type: ignore

        foreign_key_target_field = None
        for name, foreign_key in model.meta.foreign_key_fields.items():
            if foreign_key.target == self.__class__:
                foreign_key_target_field = name

        if not foreign_key_target_field:
            raise RelationshipNotFound(
                f"There was no relationship found between '{model.__class__.__name__}' and {self.__class__.__name__}"
            )

        data = reference.model_dump(exclude={"__model__"})
        data[foreign_key_target_field] = self
        await model.query.create(**data)

update_model_references

update_model_references(**kwargs)
PARAMETER DESCRIPTION
**kwargs

TYPE: Any DEFAULT: {}

Source code in edgy/core/db/models/model.py
135
136
137
138
139
140
141
142
143
144
145
146
147
def update_model_references(self, **kwargs: Any) -> Any:
    model_refs_set: Set[str] = set()
    model_references: Dict[str, Any] = {}

    for name, value in kwargs.items():
        if name in self.meta.model_references:
            model_references[name] = value
            model_refs_set.add(name)

    for value in model_refs_set:
        kwargs.pop(value)

    return kwargs, model_references

save async

save(force_save=False, values=None, **kwargs)

Performs a save of a given model instance. When creating a user it will make sure it can update existing or create a new one.

PARAMETER DESCRIPTION
force_save

TYPE: bool DEFAULT: False

values

TYPE: Dict[str, Any] DEFAULT: None

**kwargs

TYPE: Any DEFAULT: {}

Source code in edgy/core/db/models/model.py
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
async def save(
    self,
    force_save: bool = False,
    values: Dict[str, Any] = None,
    **kwargs: Any,
) -> Union[Type["Model"], Any]:
    """
    Performs a save of a given model instance.
    When creating a user it will make sure it can update existing or
    create a new one.
    """
    await self.signals.pre_save.send(sender=self.__class__, instance=self)

    extracted_fields = self.extract_db_fields()
    extracted_model_references = self.extract_db_model_references()
    extracted_fields.update(extracted_model_references)

    if getattr(self, "pk", None) is None and self.fields[self.pkname].autoincrement:
        extracted_fields.pop(self.pkname, None)

    self.update_from_dict(dict_values=dict(extracted_fields.items()))

    # Performs the update or the create based on a possible existing primary key
    if getattr(self, "pk", None) is None or force_save:
        validated_values = values or self._extract_values_from_field(
            extracted_values=extracted_fields
        )
        kwargs = self._update_auto_now_fields(values=validated_values, fields=self.fields)
        kwargs, model_references = self.update_model_references(**kwargs)
        await self._save(**kwargs)
    else:
        # Broadcast the initial update details
        # Making sure it only updates the fields that should be updated
        # and excludes the fields aith `auto_now` as true
        validated_values = values or self._extract_values_from_field(
            extracted_values=extracted_fields, is_update=True
        )
        kwargs, model_references = self.update_model_references(**validated_values)
        update_model = {k: v for k, v in validated_values.items() if k in kwargs}

        await self.signals.pre_update.send(
            sender=self.__class__, instance=self, kwargs=update_model
        )
        await self.update(**update_model)

        # Broadcast the update complete
        await self.signals.post_update.send(sender=self.__class__, instance=self)

    # Save the model references
    if model_references:
        for model_ref, references in model_references.items():
            await self.save_model_references(references or [], model_ref=model_ref)

    # Refresh the results
    if any(
        field.server_default is not None
        for name, field in self.fields.items()
        if name not in extracted_fields
    ):
        await self.load()

    await self.signals.post_save.send(sender=self.__class__, instance=self)
    return self

__getattr__

__getattr__(name)

Run an one off query to populate any foreign key making sure it runs only once per foreign key avoiding multiple database calls.

PARAMETER DESCRIPTION
name

TYPE: str

Source code in edgy/core/db/models/model.py
213
214
215
216
217
218
219
220
221
def __getattr__(self, name: str) -> Any:
    """
    Run an one off query to populate any foreign key making sure
    it runs only once per foreign key avoiding multiple database calls.
    """
    if name not in self.__dict__ and name in self.fields and name != self.pkname:
        run_sync(self.load())
        return self.__dict__[name]
    return super().__getattr__(name)