Model class

edgy.Model

Model(*args, __show_pk__=False, __phase__='init', **kwargs)

Bases: ModelRowMixin, DeclarativeMixin, DatabaseMixin, EdgyBaseModel

Representation of an Edgy Model.

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

Example

import edgyBaseFieldType
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, autoincrement=True)
    is_active: bool = edgy.BooleanField(default=False)

    class Meta:
        registry = models

PARAMETER	DESCRIPTION
*args	TYPE: Any DEFAULT: ()
__show_pk__	TYPE: bool DEFAULT: False
__phase__	TYPE: str DEFAULT: 'init'
**kwargs	TYPE: Any DEFAULT: {}

Source code in edgy/core/db/models/base.py

def __init__(
    self, *args: Any, __show_pk__: bool = False, __phase__: str = "init", **kwargs: Any
) -> None:
    self.__show_pk__ = __show_pk__
    # always set them in __dict__ to prevent __getattr__ loop
    self._loaded_or_deleted = False
    # inject in relation fields anonymous ModelRef (without a Field)
    for arg in args:
        if isinstance(arg, ModelRef):
            relation_field = self.meta.fields[arg.__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] = self
            model = target_model_class(
                **arg.model_dump(exclude={"__related_name__"}),
                **extra_params,
            )
            existing: Any = kwargs.get(arg.__related_name__)
            if isinstance(existing, Sequence):
                existing = [*existing, model]
            elif existing is None:
                existing = [model]
            else:
                existing = [existing, model]
            kwargs[arg.__related_name__] = existing

    kwargs = self.transform_input(kwargs, phase=__phase__, instance=self)
    super().__init__(**kwargs)
    # move to dict (e.g. reflected or subclasses which allow extra attributes)
    if self.__pydantic_extra__ is not None:
        # default was triggered
        self.__dict__.update(self.__pydantic_extra__)
        self.__pydantic_extra__ = None

    # cleanup fields
    for field_name in self.meta.fields:
        if field_name not in kwargs:
            self.__dict__.pop(field_name, None)

columns class-attribute

columns

database class-attribute

database = None

table property writable

table

pkcolumns property

pkcolumns

pknames property

pknames

__parent__ class-attribute

__parent__ = None

__is_proxy_model__ class-attribute

__is_proxy_model__ = False

__require_model_based_deletion__ class-attribute

__require_model_based_deletion__ = False

__reflected__ class-attribute

__reflected__ = False

_db_schemas class-attribute

_db_schemas

proxy_model property

proxy_model

identifying_db_fields cached property

identifying_db_fields

The columns used for loading, can be set per instance defaults to pknames

can_load property

can_load

__proxy_model__ class-attribute

__proxy_model__ = None

__show_pk__ class-attribute instance-attribute

__show_pk__ = __show_pk__

__using_schema__ class-attribute instance-attribute

__using_schema__ = Undefined

_loaded_or_deleted class-attribute instance-attribute

_loaded_or_deleted = False

__pydantic_extra__ instance-attribute

__pydantic_extra__ = None

signals property

signals

fields property

fields

_removed_copy_keys class-attribute

_removed_copy_keys = _removed_copy_keys

query class-attribute

query = Manager()

query_related class-attribute

query_related = RedirectManager(redirect_name='query')

meta class-attribute

meta = MetaInfo(None, abstract=True, registry=False)

Meta

abstract class-attribute instance-attribute

abstract = True

transaction

transaction(*, force_rollback=False, **kwargs)

Return database transaction for the assigned database

PARAMETER	DESCRIPTION
force_rollback	TYPE: bool DEFAULT: False
**kwargs	TYPE: Any DEFAULT: {}

Source code in edgy/core/db/models/mixins/db.py

def transaction(self, *, force_rollback: bool = False, **kwargs: Any) -> "Transaction":
    """Return database transaction for the assigned database"""
    return cast(
        "Transaction", self.database.transaction(force_rollback=force_rollback, **kwargs)
    )

get_columns_for_name

get_columns_for_name(name)

PARAMETER	DESCRIPTION
name	TYPE: str

Source code in edgy/core/db/models/mixins/db.py

def get_columns_for_name(self: "Model", name: str) -> Sequence["sqlalchemy.Column"]:
    table = self.table
    meta = self.meta
    if name in meta.field_to_columns:
        return meta.field_to_columns[name]
    elif name in table.columns:
        return (table.columns[name],)
    else:
        return cast(Sequence["sqlalchemy.Column"], _empty)

identifying_clauses

identifying_clauses(prefix='')

PARAMETER	DESCRIPTION
prefix	TYPE: str DEFAULT: ''

Source code in edgy/core/db/models/mixins/db.py

def identifying_clauses(self, prefix: str = "") -> list[Any]:
    # works only if the class of the model is the main class of the queryset
    # TODO: implement prefix handling and return generic column without table attached
    if prefix:
        raise NotImplementedError()
    clauses: list[Any] = []
    for field_name in self.identifying_db_fields:
        field = self.meta.fields.get(field_name)
        if field is not None:
            for column_name, value in field.clean(
                field_name, self.__dict__[field_name]
            ).items():
                clauses.append(getattr(self.table.columns, column_name) == value)
        else:
            clauses.append(
                getattr(self.table.columns, field_name) == self.__dict__[field_name]
            )
    return clauses

load async

load(only_needed=False)

PARAMETER	DESCRIPTION
only_needed	TYPE: bool DEFAULT: False

Source code in edgy/core/db/models/mixins/db.py

async def load(self, only_needed: bool = False) -> None:
    if only_needed and self._loaded_or_deleted:
        return
    row = None
    clauses = self.identifying_clauses()
    if clauses:
        # Build the select expression.
        expression = self.table.select().where(*clauses)

        # Perform the fetch.
        check_db_connection(self.database)
        async with self.database as database:
            row = await database.fetch_one(expression)
    # check if is in system
    if row is None:
        raise ObjectNotFound("row does not exist anymore")
    # Update the instance.
    self.__dict__.update(self.transform_input(dict(row._mapping), phase="load", instance=self))
    self._loaded_or_deleted = True

update async

update(**kwargs)

PARAMETER	DESCRIPTION
**kwargs	TYPE: Any DEFAULT: {}

Source code in edgy/core/db/models/mixins/db.py

async def update(self: "Model", **kwargs: Any) -> "Model":
    token = EXPLICIT_SPECIFIED_VALUES.set(set(kwargs.keys()))
    try:
        await self._update(**kwargs)
    finally:
        EXPLICIT_SPECIFIED_VALUES.reset(token)
    return self

save async

save(force_insert=False, values=None, force_save=None)

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_insert	TYPE: bool DEFAULT: False
values	TYPE: Union[dict[str, Any], set[str], None] DEFAULT: None
force_save	TYPE: Optional[bool] DEFAULT: None

Source code in edgy/core/db/models/mixins/db.py

async def save(
    self: "Model",
    force_insert: bool = False,
    values: Union[dict[str, Any], set[str], None] = None,
    force_save: Optional[bool] = None,
) -> "Model":
    """
    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.
    """
    if force_save is not None:
        warnings.warn(
            "'force_save' is deprecated in favor of 'force_insert'",
            DeprecationWarning,
            stacklevel=2,
        )
        force_insert = force_save

    await self.meta.signals.pre_save.send_async(self.__class__, instance=self)

    extracted_fields = self.extract_db_fields()
    if values is None:
        explicit_values: set[str] = set()
    elif isinstance(values, set):
        # special mode for marking values as explicit values
        explicit_values = set(values)
        values = None
    else:
        explicit_values = set(values.keys())

    token = MODEL_GETATTR_BEHAVIOR.set("coro")
    try:
        for pkcolumn in self.__class__.pkcolumns:
            # should trigger load in case of identifying_db_fields
            value = getattr(self, pkcolumn, None)
            if inspect.isawaitable(value):
                value = await value
            if value is None and self.table.columns[pkcolumn].autoincrement:
                extracted_fields.pop(pkcolumn, None)
                force_insert = True
            field = self.meta.fields.get(pkcolumn)
            # this is an IntegerField with primary_key set
            if field is not None and getattr(field, "increment_on_save", 0) != 0:
                # we create a new revision.
                force_insert = True
                # Note: we definitely want this because it is easy for forget a force_insert
    finally:
        MODEL_GETATTR_BEHAVIOR.reset(token)

    token2 = EXPLICIT_SPECIFIED_VALUES.set(explicit_values)
    try:
        if force_insert:
            if values:
                extracted_fields.update(values)
            # force save must ensure a complete mapping
            await self._insert(**extracted_fields)
        else:
            await self._update(**(extracted_fields if values is None else values))
    finally:
        EXPLICIT_SPECIFIED_VALUES.reset(token2)
    await self.meta.signals.post_save.send_async(self.__class__, instance=self)
    return self

delete async

delete(skip_post_delete_hooks=False, remove_referenced_call=False)

Delete operation from the database

PARAMETER	DESCRIPTION
skip_post_delete_hooks	TYPE: bool DEFAULT: False
remove_referenced_call	TYPE: bool DEFAULT: False

Source code in edgy/core/db/models/mixins/db.py

async def delete(
    self, skip_post_delete_hooks: bool = False, remove_referenced_call: bool = False
) -> None:
    """Delete operation from the database"""
    await self.meta.signals.pre_delete.send_async(self.__class__, instance=self)
    # get values before deleting
    field_values: dict[str, Any] = {}
    if not skip_post_delete_hooks and self.meta.post_delete_fields:
        token = MODEL_GETATTR_BEHAVIOR.set("coro")
        try:
            for field_name in self.meta.post_delete_fields:
                try:
                    field_value = getattr(self, field_name)
                except AttributeError:
                    continue
                if inspect.isawaitable(field_value):
                    try:
                        field_value = await field_value
                    except AttributeError:
                        continue
                field_values[field_name] = field_value
        finally:
            MODEL_GETATTR_BEHAVIOR.reset(token)
    clauses = self.identifying_clauses()
    if clauses:
        expression = self.table.delete().where(*clauses)
        check_db_connection(self.database)
        async with self.database as database:
            await database.execute(expression)
    # we cannot load anymore
    self._loaded_or_deleted = True
    # now cleanup with the saved values
    for field_name, value in field_values.items():
        field = self.meta.fields[field_name]
        await field.post_delete_callback(value, instance=self)

    await self.meta.signals.post_delete.send_async(self.__class__, instance=self)

load_recursive async

load_recursive(only_needed=False, only_needed_nest=False, _seen=None)

PARAMETER	DESCRIPTION
only_needed	TYPE: bool DEFAULT: False
only_needed_nest	TYPE: bool DEFAULT: False
_seen	TYPE: Optional[set[Any]] DEFAULT: None

Source code in edgy/core/db/models/base.py

async def load_recursive(
    self,
    only_needed: bool = False,
    only_needed_nest: bool = False,
    _seen: Optional[set[Any]] = None,
) -> None:
    if _seen is None:
        _seen = {self.create_model_key()}
    else:
        model_key = self.create_model_key()
        if model_key in _seen:
            return
        else:
            _seen.add(model_key)
    _loaded_or_deleted = self._loaded_or_deleted
    if self.can_load:
        await self.load(only_needed)
    if only_needed_nest and _loaded_or_deleted:
        return
    for field_name in self.meta.foreign_key_fields:
        value = getattr(self, field_name, None)
        if value is not None:
            # if a subinstance is fully loaded stop
            await value.load_recursive(
                only_needed=only_needed, only_needed_nest=True, _seen=_seen
            )

model_dump

model_dump(show_pk=None, **kwargs)

An updated version of the model dump. It can show the pk always and handles the exclude attribute on fields correctly and contains the custom logic for fields with getters

PARAMETER	DESCRIPTION
show_pk	TYPE: Union[bool, None] DEFAULT: None
**kwargs	TYPE: Any DEFAULT: {}

Extra Args

show_pk: bool - Enforces showing the primary key in the model_dump.

Source code in edgy/core/db/models/base.py

def model_dump(self, show_pk: Union[bool, None] = None, **kwargs: Any) -> dict[str, Any]:
    """
    An updated version of the model dump.
    It can show the pk always and handles the exclude attribute on fields correctly and
    contains the custom logic for fields with getters

    Extra Args:
        show_pk: bool - Enforces showing the primary key in the model_dump.
    """
    # we want a copy
    exclude: Union[set[str], dict[str, Any], None] = kwargs.pop("exclude", None)
    if exclude is None:
        initial_full_field_exclude = _empty
        # must be writable
        exclude = set()
    elif isinstance(exclude, dict):
        initial_full_field_exclude = {k for k, v in exclude.items() if v is True}
        exclude = copy.copy(exclude)
    else:
        initial_full_field_exclude = set(exclude)
        exclude = copy.copy(initial_full_field_exclude)

    if isinstance(exclude, dict):
        # exclude __show_pk__ attribute from showing up
        exclude["__show_pk__"] = True
        for field_name in self.meta.excluded_fields:
            exclude[field_name] = True
    else:
        exclude.update(self.meta.special_getter_fields)
        exclude.update(self.meta.excluded_fields)
        # exclude __show_pk__ attribute from showing up
        exclude.add("__show_pk__")
    include: Union[set[str], dict[str, Any], None] = kwargs.pop("include", None)
    mode: Union[Literal["json", "python"], str] = kwargs.pop("mode", "python")

    should_show_pk = self.__show_pk__ if show_pk is None else show_pk
    model = super().model_dump(exclude=exclude, include=include, mode=mode, **kwargs)
    # Workaround for metafields, computed field logic introduces many problems
    # so reimplement the logic here
    for field_name in self.meta.special_getter_fields:
        if field_name == "pk":
            continue
        if not should_show_pk or field_name not in self.pknames:
            if field_name in initial_full_field_exclude:
                continue
            if include is not None and field_name not in include:
                continue
            if getattr(field_name, "exclude", False):
                continue
        field: BaseFieldType = self.meta.fields[field_name]
        try:
            retval = field.__get__(self, self.__class__)
        except AttributeError:
            continue
        sub_include = None
        if isinstance(include, dict):
            sub_include = include.get(field_name, None)
            if sub_include is True:
                sub_include = None
        sub_exclude = None
        if isinstance(exclude, dict):
            sub_exclude = exclude.get(field_name, None)
            if sub_exclude is True:
                sub_exclude = None
        if isinstance(retval, BaseModel):
            retval = retval.model_dump(
                include=sub_include, exclude=sub_exclude, mode=mode, **kwargs
            )
        else:
            assert (
                sub_include is None
            ), "sub include filters for CompositeField specified, but no Pydantic model is set"
            assert (
                sub_exclude is None
            ), "sub exclude filters for CompositeField specified, but no Pydantic model is set"
            if mode == "json" and not getattr(field, "unsafe_json_serialization", False):
                # skip field if it isn't a BaseModel and the mode is json and unsafe_json_serialization is not set
                # currently unsafe_json_serialization exists only on CompositeFields
                continue
        alias: str = field_name
        if getattr(field, "serialization_alias", None):
            alias = cast(str, field.serialization_alias)
        elif getattr(field, "alias", None):
            alias = field.alias
        model[alias] = retval
    # proxyModel? cause excluded fields to reappear
    # TODO: find a better bugfix
    for excluded_field in self.meta.excluded_fields:
        model.pop(excluded_field, None)
    return model

build classmethod

build(schema=None, metadata=None)

Builds the SQLAlchemy table representation from the loaded fields.

PARAMETER	DESCRIPTION
schema	TYPE: Optional[str] DEFAULT: None
metadata	TYPE: Optional[MetaData] DEFAULT: None

Source code in edgy/core/db/models/mixins/db.py

@classmethod
def build(
    cls, schema: Optional[str] = None, metadata: Optional[sqlalchemy.MetaData] = None
) -> sqlalchemy.Table:
    """
    Builds the SQLAlchemy table representation from the loaded fields.
    """
    tablename: str = cls.meta.tablename
    registry = cls.meta.registry
    assert registry, "registry is not set"
    if metadata is None:
        metadata = registry.metadata_by_url[str(cls.database.url)]
    schemes: list[str] = []
    if schema:
        schemes.append(schema)
    if cls.__using_schema__ is not Undefined:
        schemes.append(cls.__using_schema__)
    db_schema = cls.get_db_schema() or ""
    schemes.append(db_schema)

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

    columns: list[sqlalchemy.Column] = []
    global_constraints: list[sqlalchemy.Constraint] = [
        copy.copy(constraint) for constraint in cls.meta.constraints
    ]
    for name, field in cls.meta.fields.items():
        current_columns = field.get_columns(name)
        columns.extend(current_columns)
        global_constraints.extend(field.get_global_constraints(name, current_columns, schemes))

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

    # Handle the indexes
    indexes = []
    for index_c in index_constraints:
        index = cls._get_indexes(index_c)
        indexes.append(index)
    return sqlalchemy.Table(
        tablename,
        metadata,
        *columns,
        *uniques,
        *indexes,
        *global_constraints,
        extend_existing=True,
        schema=schema
        if schema
        else cls.get_active_class_schema(check_schema=False, check_tenant=False),
    )

execute_post_save_hooks async

execute_post_save_hooks(fields, force_insert)

PARAMETER	DESCRIPTION
fields	TYPE: Sequence[str]
force_insert	TYPE: bool

Source code in edgy/core/db/models/base.py

async def execute_post_save_hooks(self, fields: Sequence[str], force_insert: bool) -> None:
    affected_fields = self.meta.post_save_fields.intersection(fields)
    if affected_fields:
        # don't trigger loads, AttributeErrors are used for skipping fields
        token = MODEL_GETATTR_BEHAVIOR.set("passdown")
        token2 = CURRENT_MODEL_INSTANCE.set(self)
        try:
            for field_name in affected_fields:
                field = self.meta.fields[field_name]
                try:
                    value = getattr(self, field_name)
                except AttributeError:
                    continue
                await field.post_save_callback(value, instance=self, force_insert=force_insert)
        finally:
            MODEL_GETATTR_BEHAVIOR.reset(token)
            CURRENT_MODEL_INSTANCE.reset(token2)

execute_pre_save_hooks async

execute_pre_save_hooks(column_values, original, force_insert)

PARAMETER	DESCRIPTION
column_values	TYPE: dict[str, Any]
original	TYPE: dict[str, Any]
force_insert	TYPE: bool

Source code in edgy/core/db/models/base.py

async def execute_pre_save_hooks(
    self, column_values: dict[str, Any], original: dict[str, Any], force_insert: bool
) -> dict[str, Any]:
    # also handle defaults
    keys = {*column_values.keys(), *original.keys()}
    affected_fields = self.meta.pre_save_fields.intersection(keys)
    retdict: dict[str, Any] = {}
    if affected_fields:
        # don't trigger loads
        token = MODEL_GETATTR_BEHAVIOR.set("passdown")
        token2 = CURRENT_MODEL_INSTANCE.set(self)
        try:
            for field_name in affected_fields:
                if field_name not in column_values and field_name not in original:
                    continue
                field = self.meta.fields[field_name]
                retdict.update(
                    await field.pre_save_callback(
                        column_values.get(field_name),
                        original.get(field_name),
                        force_insert=force_insert,
                        instance=self,
                    )
                )
        finally:
            MODEL_GETATTR_BEHAVIOR.reset(token)
            CURRENT_MODEL_INSTANCE.reset(token2)
    return retdict

extract_column_values classmethod

extract_column_values(extracted_values, is_update=False, is_partial=False, phase='', instance=None, model_instance=None)

PARAMETER	DESCRIPTION
extracted_values	TYPE: dict[str, Any]
is_update	TYPE: bool DEFAULT: False
is_partial	TYPE: bool DEFAULT: False
phase	TYPE: str DEFAULT: ''
instance	TYPE: Optional[Union[BaseModelType, QuerySet]] DEFAULT: None
model_instance	TYPE: Optional[BaseModelType] DEFAULT: None

Source code in edgy/core/db/models/base.py

@classmethod
def extract_column_values(
    cls,
    extracted_values: dict[str, Any],
    is_update: bool = False,
    is_partial: bool = False,
    phase: str = "",
    instance: Optional[Union[BaseModelType, QuerySet]] = None,
    model_instance: Optional[BaseModelType] = None,
) -> dict[str, Any]:
    validated: dict[str, Any] = {}
    token = CURRENT_PHASE.set(phase)
    token2 = CURRENT_INSTANCE.set(instance)
    token3 = CURRENT_MODEL_INSTANCE.set(model_instance)
    try:
        # phase 1: transform when required
        if cls.meta.input_modifying_fields:
            extracted_values = {**extracted_values}
            for field_name in cls.meta.input_modifying_fields:
                cls.meta.fields[field_name].modify_input(field_name, extracted_values)
        # phase 2: validate fields and set defaults for readonly
        need_second_pass: list[BaseFieldType] = []
        for field_name, field in cls.meta.fields.items():
            if field.read_only:
                # if read_only, updates are not possible anymore
                if (
                    not is_partial or (field.inject_default_on_partial_update and is_update)
                ) and field.has_default():
                    validated.update(field.get_default_values(field_name, validated))
                continue
            if field_name in extracted_values:
                item = extracted_values[field_name]
                assert field.owner
                for sub_name, value in field.clean(field_name, item).items():
                    if sub_name in validated:
                        raise ValueError(f"value set twice for key: {sub_name}")
                    validated[sub_name] = value
            elif (
                not is_partial or (field.inject_default_on_partial_update and is_update)
            ) and field.has_default():
                # add field without a value to the second pass (in case no value appears)
                # only include fields which have inject_default_on_partial_update set or if not is_partial
                need_second_pass.append(field)

        # phase 3: set defaults for the rest if not partial or inject_default_on_partial_update
        if need_second_pass:
            for field in need_second_pass:
                # check if field appeared e.g. by composite
                # Note: default values are directly passed without validation
                if field.name not in validated:
                    validated.update(field.get_default_values(field.name, validated))
    finally:
        CURRENT_MODEL_INSTANCE.reset(token3)
        CURRENT_INSTANCE.reset(token2)
        CURRENT_PHASE.reset(token)
    return validated

extract_db_fields

extract_db_fields(only=None)

Extracts all the db fields, model references and fields. Related fields are not included because they are disjoint.

PARAMETER	DESCRIPTION
only	TYPE: Optional[Sequence[str]] DEFAULT: None

Source code in edgy/core/db/models/types.py

def extract_db_fields(self, only: Optional[Sequence[str]] = None) -> dict[str, Any]:
    """
    Extracts all the db fields, model references and fields.
    Related fields are not included because they are disjoint.
    """
    fields = self.meta.fields
    columns = self.table.columns

    if only is not None:
        return {k: v for k, v in self.__dict__.items() if k in only}

    return {k: v for k, v in self.__dict__.items() if k in fields or hasattr(columns, k)}

get_instance_name

get_instance_name()

Returns the name of the class in lowercase.

Source code in edgy/core/db/models/types.py

def get_instance_name(self) -> str:
    """
    Returns the name of the class in lowercase.
    """
    return self.__class__.__name__.lower()

create_model_key

create_model_key()

Build a cache key for the model.

Source code in edgy/core/db/models/types.py

def create_model_key(self) -> tuple:
    """
    Build a cache key for the model.
    """
    pk_key_list: list[Any] = [type(self).__name__]
    # there are no columns, only column results
    for attr in self.pkcolumns:
        pk_key_list.append(str(getattr(self, attr)))
    return tuple(pk_key_list)

transform_input classmethod

transform_input(kwargs, phase='', instance=None)

Expand to_models and apply input modifications.

PARAMETER	DESCRIPTION
kwargs	TYPE: Any
phase	TYPE: str DEFAULT: ''
instance	TYPE: Optional[BaseModelType] DEFAULT: None

Source code in edgy/core/db/models/base.py

@classmethod
def transform_input(
    cls,
    kwargs: Any,
    phase: str = "",
    instance: Optional[BaseModelType] = None,
) -> Any:
    """
    Expand to_models and apply input modifications.
    """

    kwargs = {**kwargs}
    new_kwargs: dict[str, Any] = {}

    fields = cls.meta.fields
    token = CURRENT_INSTANCE.set(instance)
    token2 = CURRENT_MODEL_INSTANCE.set(instance)
    token3 = CURRENT_PHASE.set(phase)
    try:
        # phase 1: transform
        # Note: this is order dependend. There should be no overlap.
        for field_name in cls.meta.input_modifying_fields:
            fields[field_name].modify_input(field_name, kwargs)
        # phase 2: apply to_model
        for key, value in kwargs.items():
            field = fields.get(key, None)
            if field is not None:
                new_kwargs.update(**field.to_model(key, value))
            else:
                new_kwargs[key] = value
    finally:
        CURRENT_PHASE.reset(token3)
        CURRENT_MODEL_INSTANCE.reset(token2)
        CURRENT_INSTANCE.reset(token)
    return new_kwargs

model_dump_json

model_dump_json(**kwargs)

PARAMETER	DESCRIPTION
**kwargs	TYPE: Any DEFAULT: {}

Source code in edgy/core/db/models/base.py

def model_dump_json(self, **kwargs: Any) -> str:
    return orjson.dumps(self.model_dump(mode="json", **kwargs)).decode()

__setattr__

__setattr__(key, value)

PARAMETER	DESCRIPTION
key	TYPE: str
value	TYPE: Any

Source code in edgy/core/db/models/base.py

def __setattr__(self, key: str, value: Any) -> None:
    fields = self.meta.fields
    field = fields.get(key, None)
    token = CURRENT_INSTANCE.set(self)
    token2 = CURRENT_PHASE.set("set")
    try:
        if field is not None:
            if hasattr(field, "__set__"):
                # not recommended, better to use to_model instead except for kept objects
                # used in related_fields to mask and not to implement to_model
                field.__set__(self, value)
            else:
                for k, v in field.to_model(key, value).items():
                    if k in self.model_fields:
                        # __dict__ is updated and validator is executed
                        super().__setattr__(k, v)
                    else:
                        # bypass __setattr__ method
                        # ensures, __dict__ is updated
                        object.__setattr__(self, k, v)
        elif key in self.model_fields:
            # __dict__ is updated and validator is executed
            super().__setattr__(key, value)
        else:
            # bypass __setattr__ method
            # ensures, __dict__ is updated
            object.__setattr__(self, key, value)
    finally:
        CURRENT_INSTANCE.reset(token)
        CURRENT_PHASE.reset(token2)

_agetattr_helper async

_agetattr_helper(name, getter)

PARAMETER	DESCRIPTION
name	TYPE: str
getter	TYPE: Any

Source code in edgy/core/db/models/base.py

async def _agetattr_helper(self, name: str, getter: Any) -> Any:
    await self.load()
    if getter is not None:
        token = MODEL_GETATTR_BEHAVIOR.set("coro")
        try:
            result = getter(self, self.__class__)
            if inspect.isawaitable(result):
                result = await result
            return result
        finally:
            MODEL_GETATTR_BEHAVIOR.reset(token)
    try:
        return self.__dict__[name]
    except KeyError:
        raise AttributeError(f"Attribute: {name} not found") from None

__getattr__

__getattr__(name)

Does following things 1. Initialize managers on access 2. Redirects get accesses to getter fields 3. 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/base.py

def __getattr__(self, name: str) -> Any:
    """
    Does following things
    1. Initialize managers on access
    2. Redirects get accesses to getter fields
    3. Run an one off query to populate any foreign key making sure
       it runs only once per foreign key avoiding multiple database calls.
    """
    behavior = MODEL_GETATTR_BEHAVIOR.get()
    manager = self.meta.managers.get(name)
    if manager is not None:
        if name not in self.__dict__:
            manager = copy.copy(manager)
            manager.instance = self
            self.__dict__[name] = manager
        return self.__dict__[name]

    field = self.meta.fields.get(name)
    getter: Any = None
    if field is not None and hasattr(field, "__get__"):
        getter = field.__get__
        if behavior == "coro" or behavior == "passdown":
            return field.__get__(self, self.__class__)
        else:
            token = MODEL_GETATTR_BEHAVIOR.set("passdown")
            # no need to set an descriptor object
            try:
                return field.__get__(self, self.__class__)
            except AttributeError:
                # forward to load routine
                pass
            finally:
                MODEL_GETATTR_BEHAVIOR.reset(token)
    if (
        name not in self.__dict__
        and behavior != "passdown"
        and not self.__dict__.get("_loaded_or_deleted", False)
        and (field is not None or self.__reflected__)
        and name not in self.identifying_db_fields
        and self.can_load
    ):
        coro = self._agetattr_helper(name, getter)
        if behavior == "coro":
            return coro
        return run_sync(coro)
    return super().__getattr__(name)

__eq__

__eq__(other)

PARAMETER	DESCRIPTION
other	TYPE: Any

Source code in edgy/core/db/models/base.py

def __eq__(self, other: Any) -> bool:
    # if self.__class__ != other.__class__:
    #     return False
    # somehow meta gets regenerated, so just compare tablename and registry.
    if self.meta.registry is not other.meta.registry:
        return False
    if self.meta.tablename != other.meta.tablename:
        return False
    self_dict = self.extract_column_values(
        self.extract_db_fields(self.pkcolumns),
        is_partial=True,
        phase="compare",
        instance=self,
        model_instance=self,
    )
    other_dict = other.extract_column_values(
        other.extract_db_fields(self.pkcolumns),
        is_partial=True,
        phase="compare",
        instance=other,
        model_instance=other,
    )
    key_set = {*self_dict.keys(), *other_dict.keys()}
    for field_name in key_set:
        if self_dict.get(field_name) != other_dict.get(field_name):
            return False
    return True

add_to_registry classmethod

add_to_registry(registry, name='', database='keep', replace_related_field=False)

PARAMETER	DESCRIPTION
registry	TYPE: Registry
name	TYPE: str DEFAULT: ''
database	TYPE: Union[bool, Database, Literal['keep']] DEFAULT: 'keep'
replace_related_field	TYPE: Union[bool, type[BaseModelType]] DEFAULT: False

Source code in edgy/core/db/models/mixins/db.py

@classmethod
def add_to_registry(
    cls: type["BaseModelType"],
    registry: "Registry",
    name: str = "",
    database: Union[bool, "Database", Literal["keep"]] = "keep",
    replace_related_field: Union[bool, type["BaseModelType"]] = False,
) -> None:
    # when called if registry is not set
    cls.meta.registry = registry
    if database is True:
        cls.database = registry.database
    elif database is not False:
        if database == "keep":
            if getattr(cls, "database", None) is None:
                cls.database = registry.database

        else:
            cls.database = database
    meta = cls.meta
    if name:
        cls.__name__ = name

    # Making sure it does not generate models if abstract or a proxy
    if not meta.abstract and not cls.__is_proxy_model__:
        if getattr(cls, "__reflected__", False):
            registry.reflected[cls.__name__] = cls
        else:
            registry.models[cls.__name__] = cls
        # after registrating the own model
        for value in list(meta.fields.values()):
            if isinstance(value, BaseManyToManyForeignKeyField):
                m2m_registry: Registry = value.target_registry
                with contextlib.suppress(Exception):
                    m2m_registry = cast("Registry", value.target.registry)

                def create_through_model(x: Any, field: "BaseFieldType" = value) -> None:
                    # we capture with field = ... the variable
                    field.create_through_model()

                m2m_registry.register_callback(value.to, create_through_model, one_time=True)
        # Sets the foreign key fields
        if meta.foreign_key_fields:
            _set_related_name_for_foreign_keys(
                meta, cls, replace_related_field=replace_related_field
            )
        registry.execute_model_callbacks(cls)

    # finalize
    cls.model_rebuild(force=True)

get_active_instance_schema

get_active_instance_schema(check_schema=True, check_tenant=True)

PARAMETER	DESCRIPTION
check_schema	TYPE: bool DEFAULT: True
check_tenant	TYPE: bool DEFAULT: True

Source code in edgy/core/db/models/mixins/db.py

def get_active_instance_schema(
    self, check_schema: bool = True, check_tenant: bool = True
) -> Union[str, None]:
    if self.__using_schema__ is not Undefined:
        return cast(Union[str, None], self.__using_schema__)
    return self.__class__.get_active_class_schema(
        check_schema=check_schema, check_tenant=check_tenant
    )

get_active_class_schema classmethod

get_active_class_schema(check_schema=True, check_tenant=True)

PARAMETER	DESCRIPTION
check_schema	TYPE: bool DEFAULT: True
check_tenant	TYPE: bool DEFAULT: True

Source code in edgy/core/db/models/mixins/db.py

@classmethod
def get_active_class_schema(cls, check_schema: bool = True, check_tenant: bool = True) -> str:
    if cls.__using_schema__ is not Undefined:
        return cast(Union[str, None], cls.__using_schema__)
    if check_schema:
        schema = get_schema(check_tenant=check_tenant)
        if schema is not None:
            return schema
    db_schema: Optional[str] = cls.get_db_schema()
    # sometime "" is ok, sometimes not, sqlalchemy logic
    return db_schema or None

copy_edgy_model classmethod

copy_edgy_model(registry=None, name='', **kwargs)

Copy the model class and optionally add it to another registry.

PARAMETER	DESCRIPTION
registry	TYPE: Optional[Registry] DEFAULT: None
name	TYPE: str DEFAULT: ''
**kwargs	TYPE: Any DEFAULT: {}

Source code in edgy/core/db/models/mixins/db.py

@classmethod
def copy_edgy_model(
    cls: type["Model"], registry: Optional["Registry"] = None, name: str = "", **kwargs: Any
) -> type["Model"]:
    """Copy the model class and optionally add it to another registry."""
    # removes private pydantic stuff, except the prefixed ones
    attrs = {
        key: val for key, val in cls.__dict__.items() if key not in cls._removed_copy_keys
    }
    # managers and fields are gone, we have to readd them with the correct data
    attrs.update(
        (
            (field_name, field)
            for field_name, field in cls.meta.fields.items()
            if not field.no_copy
        )
    )
    attrs.update(cls.meta.managers)
    _copy = cast(
        type["Model"],
        type(cls.__name__, cls.__bases__, attrs, skip_registry=True, **kwargs),
    )
    for field_name in _copy.meta.foreign_key_fields:
        # we need to unreference and check if both models are in the same registry
        if cls.meta.fields[field_name].target.meta.registry is cls.meta.registry:
            _copy.meta.fields[field_name].target = cls.meta.fields[field_name].target.__name__
        else:
            # otherwise we need to disable backrefs
            _copy.meta.fields[field_name].target.related_name = False
    if name:
        _copy.__name__ = name
    if registry is not None:
        # replace when old class otherwise old references can lead to issues
        _copy.add_to_registry(registry, replace_related_field=cls)
    return _copy

_update async

_update(**kwargs)

Update operation of the database fields.

PARAMETER	DESCRIPTION
**kwargs	TYPE: Any DEFAULT: {}

Source code in edgy/core/db/models/mixins/db.py

async def _update(self: "Model", **kwargs: Any) -> Any:
    """
    Update operation of the database fields.
    """
    await self.meta.signals.pre_update.send_async(self.__class__, instance=self)
    column_values = self.extract_column_values(
        extracted_values=kwargs,
        is_partial=True,
        is_update=True,
        phase="prepare_update",
        instance=self,
        model_instance=self,
    )
    # empty updates shouldn't cause an error. E.g. only model references are updated
    clauses = self.identifying_clauses()
    token = CURRENT_INSTANCE.set(self)
    try:
        if column_values and clauses:
            check_db_connection(self.database, stacklevel=4)
            async with self.database as database, database.transaction():
                # can update column_values
                column_values.update(
                    await self.execute_pre_save_hooks(
                        column_values, kwargs, force_insert=False
                    )
                )
                expression = self.table.update().values(**column_values).where(*clauses)
                await database.execute(expression)

            # Update the model instance.
            new_kwargs = self.transform_input(
                column_values, phase="post_update", instance=self
            )
            self.__dict__.update(new_kwargs)

        # updates aren't required to change the db, they can also just affect the meta fields
        await self.execute_post_save_hooks(
            cast(Sequence[str], kwargs.keys()), force_insert=False
        )

    finally:
        CURRENT_INSTANCE.reset(token)
    if column_values or kwargs:
        # Ensure on access refresh the results is active
        self._loaded_or_deleted = False
    await self.meta.signals.post_update.send_async(self.__class__, instance=self)

_insert async

_insert(**kwargs)

Performs the save instruction.

PARAMETER	DESCRIPTION
**kwargs	TYPE: Any DEFAULT: {}

Source code in edgy/core/db/models/mixins/db.py

async def _insert(self: "Model", **kwargs: Any) -> "Model":
    """
    Performs the save instruction.
    """
    column_values: dict[str, Any] = self.extract_column_values(
        extracted_values=kwargs,
        is_partial=False,
        is_update=False,
        phase="prepare_insert",
        instance=self,
        model_instance=self,
    )
    check_db_connection(self.database, stacklevel=4)
    token = CURRENT_INSTANCE.set(self)
    try:
        async with self.database as database, database.transaction():
            # can update column_values
            column_values.update(
                await self.execute_pre_save_hooks(column_values, kwargs, force_insert=True)
            )
            expression = self.table.insert().values(**column_values)
            autoincrement_value = await database.execute(expression)
        # sqlalchemy supports only one autoincrement column
        if autoincrement_value:
            column = self.table.autoincrement_column
            if column is not None and hasattr(autoincrement_value, "_mapping"):
                autoincrement_value = autoincrement_value._mapping[column.key]
            # can be explicit set, which causes an invalid value returned
            if column is not None and column.key not in column_values:
                column_values[column.key] = autoincrement_value

        new_kwargs = self.transform_input(column_values, phase="post_insert", instance=self)
        self.__dict__.update(new_kwargs)

        if self.meta.post_save_fields:
            await self.execute_post_save_hooks(
                cast(Sequence[str], kwargs.keys()), force_insert=True
            )
    finally:
        CURRENT_INSTANCE.reset(token)
    # Ensure on access refresh the results is active
    self._loaded_or_deleted = False

    return self

_get_unique_constraints classmethod

_get_unique_constraints(fields)

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
fields	TYPE: Union[Sequence, str, UniqueConstraint]

Source code in edgy/core/db/models/mixins/db.py

@classmethod
def _get_unique_constraints(
    cls, fields: Union[Sequence, str, sqlalchemy.UniqueConstraint]
) -> 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(fields, str):
        return sqlalchemy.UniqueConstraint(*cls.meta.field_to_column_names[fields])
    elif isinstance(fields, UniqueConstraint):
        return sqlalchemy.UniqueConstraint(
            *chain.from_iterable(
                cls.meta.field_to_column_names[field] for field in fields.fields
            ),
            name=fields.name,
            deferrable=fields.deferrable,
            initially=fields.initially,
        )
    return sqlalchemy.UniqueConstraint(
        *chain.from_iterable(cls.meta.field_to_column_names[field] for field in fields)
    )

_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/mixins/db.py

@classmethod
def _get_indexes(cls, index: Index) -> Optional[sqlalchemy.Index]:
    """
    Creates the index based on the Index fields
    """
    return sqlalchemy.Index(
        index.name,
        *chain.from_iterable(
            [field]
            if isinstance(field, sqlalchemy.TextClause)
            else cls.meta.field_to_column_names[field]
            for field in index.fields
        ),
    )

declarative classmethod

declarative()

Source code in edgy/core/db/models/mixins/generics.py

@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

@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.meta.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

can_load_from_row classmethod

can_load_from_row(row, table)

Check if a model_class can be loaded from a row for the table.

PARAMETER	DESCRIPTION
row	TYPE: Row
table	TYPE: Table

Source code in edgy/core/db/models/mixins/row.py

@classmethod
def can_load_from_row(cls: type["Model"], row: "Row", table: "Table") -> bool:
    """Check if a model_class can be loaded from a row for the table."""

    return bool(
        cls.meta.registry
        and not cls.meta.abstract
        and all(row._mapping.get(f"{table.name}_{col}") is not None for col in cls.pkcolumns)
    )

from_sqla_row async classmethod

from_sqla_row(row, tables_and_models, select_related=None, prefetch_related=None, only_fields=None, is_defer_fields=False, exclude_secrets=False, using_schema=None, database=None, prefix='', old_select_related_value=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
tables_and_models	TYPE: dict[str, tuple[Table, type[BaseModelType]]]
select_related	TYPE: Optional[Sequence[Any]] DEFAULT: None
prefetch_related	TYPE: Optional[Sequence[Prefetch]] DEFAULT: None
only_fields	TYPE: Sequence[str] DEFAULT: None
is_defer_fields	TYPE: bool DEFAULT: False
exclude_secrets	TYPE: bool DEFAULT: False
using_schema	TYPE: Optional[str] DEFAULT: None
database	TYPE: Optional[Database] DEFAULT: None
prefix	TYPE: str DEFAULT: ''
old_select_related_value	TYPE: Optional[Model] DEFAULT: None

Source code in edgy/core/db/models/mixins/row.py

@classmethod
async def from_sqla_row(
    cls: type["Model"],
    row: "Row",
    # contain the mappings used for select
    tables_and_models: dict[str, tuple["Table", type["BaseModelType"]]],
    select_related: Optional[Sequence[Any]] = None,
    prefetch_related: Optional[Sequence["Prefetch"]] = None,
    only_fields: Sequence[str] = None,
    is_defer_fields: bool = False,
    exclude_secrets: bool = False,
    using_schema: Optional[str] = None,
    database: Optional["Database"] = None,
    prefix: str = "",
    old_select_related_value: Optional["Model"] = None,
) -> Optional["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_columns: set[str] = set()
    if exclude_secrets:
        for name in cls.meta.secret_fields:
            secret_columns.update(cls.meta.field_to_column_names[name])

    for related in select_related:
        field_name = related.split("__", 1)[0]
        try:
            field = cls.meta.fields[field_name]
        except KeyError:
            raise QuerySetError(
                detail=f'Selected field "{field_name}cast("Model", " does not exist on {cls}.'
            ) from None
        if isinstance(field, RelationshipField):
            model_class, _, remainder = field.traverse_field(related)
        else:
            raise QuerySetError(
                detail=f'Selected field "{field_name}" is not a RelationshipField on {cls}.'
            ) from None

        _prefix = field_name if not prefix else f"{prefix}__{field_name}"
        # stop selecting when None. Related models are not available.
        if not model_class.can_load_from_row(
            row,
            tables_and_models[_prefix][0],
        ):
            continue

        if remainder:
            # don't pass table, it is only for the main model_class
            item[field_name] = await model_class.from_sqla_row(
                row,
                tables_and_models=tables_and_models,
                select_related=[remainder],
                prefetch_related=prefetch_related,
                exclude_secrets=exclude_secrets,
                is_defer_fields=is_defer_fields,
                using_schema=using_schema,
                database=database,
                prefix=_prefix,
                old_select_related_value=item.get(field_name),
            )
        else:
            # don't pass table, it is only for the main model_class
            item[field_name] = await model_class.from_sqla_row(
                row,
                tables_and_models=tables_and_models,
                exclude_secrets=exclude_secrets,
                is_defer_fields=is_defer_fields,
                using_schema=using_schema,
                database=database,
                prefix=_prefix,
                old_select_related_value=item.get(field_name),
            )
    # don't overwrite, update with new values and return
    if old_select_related_value:
        for k, v in item.items():
            setattr(old_select_related_value, k, v)
        return old_select_related_value
    table_columns = tables_and_models[prefix][0].columns
    # 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 in cls.meta.foreign_key_fields:
        foreign_key = cls.meta.fields[related]
        ignore_related: bool = cls.__should_ignore_related_name(related, select_related)
        if ignore_related or related in cls.meta.secret_fields:
            continue
        if related in item:
            continue

        if exclude_secrets and foreign_key.secret:
            continue
        columns_to_check = foreign_key.get_column_names(related)

        model_related = foreign_key.target

        child_item = {}
        for column_name in columns_to_check:
            column = getattr(table_columns, column_name, None)
            if column_name is None:
                continue
            columnkeyhash = column_name
            if prefix:
                columnkeyhash = f"{tables_and_models[prefix][0].name}_{column.key}"

            if columnkeyhash in row._mapping:
                child_item[foreign_key.from_fk_field_name(related, column_name)] = (
                    row._mapping[columnkeyhash]
                )
        # 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.
        proxy_model = model_related.proxy_model(**child_item)
        proxy_database = database if model_related.database is cls.database else None
        # don't pass a table. It is not in the row (select related path) and has not an explicit table
        proxy_model = apply_instance_extras(
            proxy_model,
            model_related,
            using_schema,
            database=proxy_database,
        )
        proxy_model.identifying_db_fields = foreign_key.related_columns

        item[related] = proxy_model

    # Check for the only_fields
    # Pull out the regular column values.
    for column in table_columns:
        if (
            only_fields
            and prefix not in only_fields
            and (f"{prefix}__{column.key}" if prefix else column.key) not in only_fields
        ):
            continue
        if column.key in secret_columns:
            continue
        if column.key not in cls.meta.columns_to_field:
            continue
        # set if not of an foreign key with one column
        if column.key in item:
            continue
        columnkeyhash = column.key
        if prefix:
            columnkeyhash = f"{tables_and_models[prefix][0].name}_{columnkeyhash}"

        if columnkeyhash in row._mapping:
            item[column.key] = row._mapping[columnkeyhash]
    model: Model = (
        cls.proxy_model(**item, __phase__="init_db")  # type: ignore
        if exclude_secrets or is_defer_fields or only_fields
        else cls(**item, __phase__="init_db")
    )
    # Apply the schema to the model
    model = apply_instance_extras(
        model,
        cls,
        using_schema,
        database=database,
        table=tables_and_models[prefix][0],
    )

    if prefetch_related:
        # Handle prefetch related fields.
        await cls.__handle_prefetch_related(
            row=row,
            prefix=prefix,
            model=model,
            tables_and_models=tables_and_models,
            prefetch_related=prefetch_related,
        )
    assert model.pk is not None, model
    return model

__should_ignore_related_name classmethod

__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/mixins/row.py

@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

create_model_key_from_sqla_row classmethod

create_model_key_from_sqla_row(row, row_prefix='')

Build a cache key for the model.

PARAMETER	DESCRIPTION
row	TYPE: Row
row_prefix	TYPE: str DEFAULT: ''

Source code in edgy/core/db/models/mixins/row.py

@classmethod
def create_model_key_from_sqla_row(cls, row: "Row", row_prefix: str = "") -> tuple:
    """
    Build a cache key for the model.
    """
    pk_key_list: list[Any] = [cls.__name__]
    for attr in cls.pkcolumns:
        pk_key_list.append(str(row._mapping[f"{row_prefix}{attr}"]))
    return tuple(pk_key_list)

__set_prefetch async classmethod

__set_prefetch(row, model, row_prefix, related)

PARAMETER	DESCRIPTION
row	TYPE: Row
model	TYPE: Model
row_prefix	TYPE: str
related	TYPE: Prefetch

Source code in edgy/core/db/models/mixins/row.py

@classmethod
async def __set_prefetch(
    cls,
    row: "Row",
    model: "Model",
    row_prefix: str,
    related: "Prefetch",
) -> None:
    model_key = ()
    if related._is_finished:
        # when force_rollback
        # we can only bake after all rows are retrieved
        # this is why it is here
        await related.init_bake(type(model))
        model_key = model.create_model_key()
    if model_key in related._baked_results:
        setattr(model, related.to_attr, related._baked_results[model_key])
    else:
        crawl_result = crawl_relationship(
            model.__class__, related.related_name, traverse_last=True
        )
        if crawl_result.reverse_path is False:
            QuerySetError(
                detail=("Creating a reverse path is not possible, unidirectional fields used.")
            )
        if crawl_result.cross_db_remainder:
            raise NotImplementedError(
                "Cannot prefetch from other db yet. Maybe in future this feature will be added."
            )
        queryset = related.queryset
        if related._is_finished:
            assert queryset is not None, "Queryset is not set but _is_finished flag"
        else:
            check_prefetch_collision(model, related)
            if queryset is None:
                queryset = crawl_result.model_class.query.all()

            queryset = queryset.select_related(cast(str, crawl_result.reverse_path))
        clause = {
            f"{crawl_result.reverse_path}__{pkcol}": row._mapping[f"{row_prefix}{pkcol}"]
            for pkcol in cls.pkcolumns
        }
        setattr(model, related.to_attr, await queryset.filter(clause))

__handle_prefetch_related async classmethod

__handle_prefetch_related(row, model, prefix, tables_and_models, prefetch_related)

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: Model
prefix	TYPE: str
tables_and_models	TYPE: dict[str, tuple[Table, type[BaseModelType]]]
prefetch_related	TYPE: Sequence[Prefetch]

Source code in edgy/core/db/models/mixins/row.py

@classmethod
async def __handle_prefetch_related(
    cls,
    row: "Row",
    model: "Model",
    prefix: str,
    tables_and_models: dict[str, tuple["Table", type["BaseModelType"]]],
    prefetch_related: Sequence["Prefetch"],
) -> None:
    """
    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.
    """

    queries = []

    for related in prefetch_related:
        # Check for conflicting names
        # Check as early as possible
        check_prefetch_collision(model=model, related=related)
        row_prefix = f"{tables_and_models[prefix].name}_" if prefix else ""
        queries.append(
            cls.__set_prefetch(row=row, row_prefix=row_prefix, model=model, related=related)
        )
    if queries:
        await asyncio.gather(*queries)

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/model.py

@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.
    """
    fields = {key: copy.copy(field) for key, field in cls.meta.fields.items()}

    class MethodHolder(Model):
        pass

    ignore = set(dir(MethodHolder))

    for key in dir(cls):
        if key in ignore or key.startswith("_"):
            continue
        val = inspect.getattr_static(cls, key)
        if inspect.isfunction(val):
            setattr(MethodHolder, key, val)

    proxy_model = ProxyModel(
        name=cls.__name__,
        module=cls.__module__,
        metadata=cls.meta,
        definitions=fields,
        bases=(MethodHolder,),
    )

    proxy_model.build()
    generify_model_fields(cast(type[EdgyBaseModel], proxy_model.model))
    return cast(type[Model], proxy_model.model)