Model class

edgy.Model

Model(*args, **kwargs)

Bases: ModelRowMixin, DeclarativeMixin, DatabaseMixin, AdminMixin, DumpMixin, EdgyBaseModel

The Model class represents an Edgy ORM model, serving as the foundation for database table mapping and interactions.

This class combines various mixins to provide a comprehensive set of functionalities, including row-level operations (ModelRowMixin), declarative capabilities for SQLAlchemy model generation (DeclarativeMixin), database connectivity (DatabaseMixin), administrative features (AdminMixin), and data dumping capabilities (DumpMixin). It inherits from EdgyBaseModel for core model features and uses BaseModelMeta as its metaclass to handle model registration and setup.

Models defined inheriting from this class can be automatically converted into SQLAlchemy declarative models, facilitating database schema generation and ORM operations.

Example

import edgy
from edgy import Database, Registry

# Initialize a database connection and a registry for models.
database = Database("sqlite:///db.sqlite")
models = Registry(database=database)


class User(edgy.Model):
    '''
    The User model represents a 'users' table in the database.
    If no table name is explicitly provided in the Meta class,
    Edgy automatically infers it (e.g., "users" from "User").
    '''

    # Define model fields using Edgy's field types.
    id: int = edgy.IntegerField(primary_key=True, autoincrement=True)
    is_active: bool = edgy.BooleanField(default=False)

    class Meta:
        # Associate the model with a registry for database operations.
        registry = models

Initializes the DatabaseMixin.

PARAMETER	DESCRIPTION
*args	Variable length argument list. TYPE: Any DEFAULT: ()
**kwargs	Arbitrary keyword arguments. TYPE: Any DEFAULT: {}

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

def __init__(self, *args: Any, **kwargs: Any) -> None:
    """
    Initializes the DatabaseMixin.

    Args:
        *args: Variable length argument list.
        **kwargs: Arbitrary keyword arguments.
    """
    super().__init__(*args, **kwargs)
    self.transaction = self.not_set_transaction

columns class-attribute

columns

database class-attribute

database = None

table deletable property writable

table

Returns the SQLAlchemy table associated with the model instance.

If the table is not already set on the instance, it will be built dynamically based on the active schema.

RETURNS	DESCRIPTION
Table	The SQLAlchemy Table object.

pkcolumns property

pkcolumns

Returns the names of the primary key columns for the model instance.

If not already cached, it builds them based on the model's table.

RETURNS	DESCRIPTION
Sequence[str]	A sequence of strings representing the primary key column names.

pknames property

pknames

Returns the logical names of the primary key fields for the model.

RETURNS	DESCRIPTION
Sequence[str]	A sequence of strings representing the primary key field names.

transaction instance-attribute

transaction = not_set_transaction

__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

identifying_db_fields cached property

identifying_db_fields

Returns the columns used for loading the model instance. Defaults to primary key columns.

can_load property

can_load

Checks if the model instance can be loaded from the database. Requires a registry, not to be abstract, and all identifying fields to have values.

:return: True if the model can be loaded, False otherwise.

_edgy_private_attrs class-attribute

_edgy_private_attrs = PrivateAttr(default={'__show_pk__', '__using_schema__', '__no_load_trigger_attrs__', '__deletion_with_signals__', 'database', 'transaction'})

_edgy_namespace class-attribute instance-attribute

_edgy_namespace = _edgy_namespace

__proxy_model__ class-attribute

__proxy_model__ = None

__show_pk__ class-attribute

__show_pk__ = False

__using_schema__ class-attribute

__using_schema__ = Undefined

__deletion_with_signals__ class-attribute

__deletion_with_signals__ = False

__no_load_trigger_attrs__ class-attribute

__no_load_trigger_attrs__ = _empty

_db_loaded class-attribute instance-attribute

_db_loaded = _db_loaded

_db_deleted class-attribute instance-attribute

_db_deleted = _db_deleted

_db_schemas class-attribute

_db_schemas

__pydantic_extra__ instance-attribute

__pydantic_extra__ = None

_db_loaded_or_deleted property

_db_loaded_or_deleted

Indicates if the model instance is loaded from the database or marked as deleted.

:return: True if loaded or deleted, False otherwise.

_loaded_or_deleted property

_loaded_or_deleted

Deprecated: Use _db_loaded_or_deleted instead.

Indicates if the model instance is loaded from the database or marked as deleted.

:return: True if loaded or deleted, False otherwise.

signals property

signals

Deprecated: Use meta.signals instead.

Returns the broadcaster for signals.

fields property

fields

Deprecated: Use meta.fields instead.

Returns a dictionary of the model's fields.

_removed_copy_keys class-attribute

_removed_copy_keys = _removed_copy_keys

proxy_model class-attribute

proxy_model

__parent__ class-attribute

__parent__

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

Inner Meta class for configuring model-specific options.

ATTRIBUTE	DESCRIPTION
abstract	If True, the model is considered abstract and will not be created as a database table. Defaults to True. TYPE: bool
registry	If False, prevents the model from inheriting a registry from its base classes, ensuring it's not automatically added to a global registry unless explicitly specified. Defaults to False. TYPE: bool

abstract class-attribute instance-attribute

abstract = True

registry class-attribute instance-attribute

registry = False

get_columns_for_name

get_columns_for_name(name)

Retrieves the SQLAlchemy columns associated with a given field name.

PARAMETER	DESCRIPTION
name	The name of the field. TYPE: str

RETURNS	DESCRIPTION
Sequence[Column]	A sequence of SQLAlchemy Column objects.

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

def get_columns_for_name(self: Model, name: str) -> Sequence[sqlalchemy.Column]:
    """
    Retrieves the SQLAlchemy columns associated with a given field name.

    Args:
        name: The name of the field.

    Returns:
        A sequence of SQLAlchemy `Column` objects.
    """
    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='')

Generates SQLAlchemy clauses for identifying the current model instance.

These clauses are typically used in WHERE conditions for update and delete operations, based on the model's identifying database fields (usually primary keys).

PARAMETER	DESCRIPTION
prefix	An optional prefix to apply to column names in the clauses. (Currently, this feature is not fully implemented and will raise a NotImplementedError if used.) TYPE: str DEFAULT: ''

RETURNS	DESCRIPTION
list[Any]	A list of SQLAlchemy clause elements.

RAISES	DESCRIPTION
NotImplementedError	If a prefix is provided.

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

def identifying_clauses(self, prefix: str = "") -> list[Any]:
    """
    Generates SQLAlchemy clauses for identifying the current model instance.

    These clauses are typically used in WHERE conditions for update and delete
    operations, based on the model's identifying database fields (usually primary keys).

    Args:
        prefix: An optional prefix to apply to column names in the clauses.
                (Currently, this feature is not fully implemented and will
                raise a `NotImplementedError` if used.)

    Returns:
        A list of SQLAlchemy clause elements.

    Raises:
        NotImplementedError: If a prefix is provided.
    """
    # 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)

Loads the current model instance's data from the database.

This method fetches the record corresponding to the instance's identifying clauses and updates the instance's attributes. If only_needed is True, it skips loading if the instance is already loaded or marked as deleted.

PARAMETER	DESCRIPTION
only_needed	If True, loads data only if the instance is not already loaded or deleted. TYPE: bool DEFAULT: False

RAISES	DESCRIPTION
ObjectNotFound	If no row is found in the database corresponding to the instance's identifying clauses.

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

async def load(self, only_needed: bool = False) -> None:
    """
    Loads the current model instance's data from the database.

    This method fetches the record corresponding to the instance's identifying
    clauses and updates the instance's attributes. If `only_needed` is True,
    it skips loading if the instance is already loaded or marked as deleted.

    Args:
        only_needed: If True, loads data only if the instance is not already
                     loaded or deleted.

    Raises:
        ObjectNotFound: If no row is found in the database corresponding to
                        the instance's identifying clauses.
    """
    if only_needed and self._db_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:
        self._db_deleted = True
        self._db_loaded = True
        raise ObjectNotFound("row does not exist anymore")
    # Update the instance.
    self.__dict__.update(self.transform_input(dict(row._mapping), phase="load", instance=self))
    self._db_deleted = False
    self._db_loaded = True

update async

update(**kwargs)

Updates the current model instance in the database with the provided keyword arguments.

This method triggers pre-update and post-update signals.

PARAMETER	DESCRIPTION
**kwargs	Keyword arguments representing the fields and their new values to update. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
Self	The updated model instance.

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

async def update(self: Model, **kwargs: Any) -> Self:
    """
    Updates the current model instance in the database with the provided keyword arguments.

    This method triggers pre-update and post-update signals.

    Args:
        **kwargs: Keyword arguments representing the fields and their new values
                  to update.

    Returns:
        The updated model instance.
    """
    token = EXPLICIT_SPECIFIED_VALUES.set(set(kwargs.keys()))
    token2 = CURRENT_INSTANCE.set(self)
    try:
        # assume always partial
        await self._update(
            True,
            kwargs,
            pre_fn=partial(
                self.meta.signals.pre_update.send_async, is_update=True, is_migration=False
            ),
            post_fn=partial(
                self.meta.signals.post_update.send_async, is_update=True, is_migration=False
            ),
            instance=self,
        )
    finally:
        EXPLICIT_SPECIFIED_VALUES.reset(token)
        CURRENT_INSTANCE.reset(token2)
    return cast("Self", self)

real_save async

real_save(force_insert, values)

Performs the actual save operation, determining whether to insert or update.

This method checks for the existence of the instance in the database and decides whether to perform an _insert or _update operation. It also handles pre-save and post-save signals.

PARAMETER	DESCRIPTION
force_insert	If True, forces an insert operation regardless of whether the instance already exists in the database. TYPE: bool
values	A dictionary of specific values to save, or a set of field names to explicitly mark as modified for a partial update. If None, all extracted database fields are considered. TYPE: dict[str, Any] | set[str] | None

RETURNS	DESCRIPTION
Self	The saved model instance.

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

async def real_save(
    self,
    force_insert: bool,
    values: dict[str, Any] | set[str] | None,
) -> Self:
    """
    Performs the actual save operation, determining whether to insert or update.

    This method checks for the existence of the instance in the database and
    decides whether to perform an `_insert` or `_update` operation. It also
    handles pre-save and post-save signals.

    Args:
        force_insert: If True, forces an insert operation regardless of whether
                      the instance already exists in the database.
        values: A dictionary of specific values to save, or a set of field
                names to explicitly mark as modified for a partial update.
                If None, all extracted database fields are considered.

    Returns:
        The saved model instance.
    """
    instance: BaseModelType | QuerySet = CURRENT_INSTANCE.get()
    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 type(self).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/DateTime with primary_key set
            if field is not None:
                if getattr(field, "increment_on_save", 0) != 0 or getattr(
                    field, "auto_now", False
                ):
                    # we create a new revision.
                    force_insert = True
                elif getattr(field, "auto_now_add", False):  # noqa: SIM102
                    # force_insert if auto_now_add field is empty
                    if value is None:
                        force_insert = True

        # check if it exists
        if not force_insert and not await self.check_exist_in_db(only_needed=True):
            force_insert = True
    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(
                bool(values),
                extracted_fields,
                pre_fn=partial(
                    self.meta.signals.pre_save.send_async, is_update=False, is_migration=False
                ),
                post_fn=partial(
                    self.meta.signals.post_save.send_async, is_update=False, is_migration=False
                ),
                instance=instance,
            )
        else:
            await self._update(
                # assume partial when values are None
                values is not None,
                extracted_fields if values is None else values,
                pre_fn=partial(
                    self.meta.signals.pre_save.send_async, is_update=True, is_migration=False
                ),
                post_fn=partial(
                    self.meta.signals.post_save.send_async, is_update=True, is_migration=False
                ),
                instance=instance,
            )
    finally:
        EXPLICIT_SPECIFIED_VALUES.reset(token2)
    return self

save async

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

Saves the current model instance to the database.

This method acts as a public entry point for saving, encapsulating the logic for deciding between an insert or update operation. It also handles context variable management for the save process.

PARAMETER	DESCRIPTION
force_insert	If True, forces an insert operation even if the instance might already exist. TYPE: bool DEFAULT: False
values	A dictionary of specific values to save, or a set of field names to explicitly mark as modified for a partial update. If None, all extracted database fields are considered. TYPE: dict[str, Any] | set[str] | None DEFAULT: None
force_save	Deprecated. Use force_insert instead. TYPE: bool | None DEFAULT: None

RETURNS	DESCRIPTION
Model	The saved model instance.

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

async def save(
    self: Model,
    force_insert: bool = False,
    values: dict[str, Any] | set[str] | None = None,
    force_save: bool | None = None,
) -> Model:
    """
    Saves the current model instance to the database.

    This method acts as a public entry point for saving, encapsulating the
    logic for deciding between an insert or update operation. It also handles
    context variable management for the save process.

    Args:
        force_insert: If True, forces an insert operation even if the instance
                      might already exist.
        values: A dictionary of specific values to save, or a set of field
                names to explicitly mark as modified for a partial update.
                If None, all extracted database fields are considered.
        force_save: Deprecated. Use `force_insert` instead.

    Returns:
        The saved model instance.
    """
    if force_save is not None:
        warnings.warn(
            "'force_save' is deprecated in favor of 'force_insert'",
            DeprecationWarning,
            stacklevel=2,
        )
        force_insert = force_save
    token = CURRENT_INSTANCE.set(self)
    try:
        return await self.real_save(force_insert=force_insert, values=values)
    finally:
        CURRENT_INSTANCE.reset(token)

raw_delete async

raw_delete(*, skip_post_delete_hooks, remove_referenced_call)

Performs the low-level delete operation from the database.

This method handles pre-delete signals, cascades deletions (if configured), and post-delete cleanup.

PARAMETER	DESCRIPTION
skip_post_delete_hooks	If True, post-delete hooks will not be executed. TYPE: bool
remove_referenced_call	A boolean or string indicating if the deletion is triggered by a referenced call (e.g., cascade delete). If a string, it represents the field name that triggered it. TYPE: bool | str

RETURNS	DESCRIPTION
int	The number of rows deleted.

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

async def raw_delete(
    self: Model, *, skip_post_delete_hooks: bool, remove_referenced_call: bool | str
) -> int:
    """
    Performs the low-level delete operation from the database.

    This method handles pre-delete signals, cascades deletions (if configured),
    and post-delete cleanup.

    Args:
        skip_post_delete_hooks: If True, post-delete hooks will not be executed.
        remove_referenced_call: A boolean or string indicating if the deletion
                                is triggered by a referenced call (e.g., cascade delete).
                                If a string, it represents the field name that triggered it.

    Returns:
        The number of rows deleted.
    """
    if self._db_deleted:
        return 0
    instance = CURRENT_INSTANCE.get()
    real_class = self.get_real_class()
    # remove_referenced_call = called from a deleter of another field or model
    with_signals = self.__deletion_with_signals__ and (
        instance is not self or remove_referenced_call
    )
    if with_signals:
        await self.meta.signals.pre_delete.send_async(
            real_class, instance=instance, model_instance=self
        )
    ignore_fields: set[str] = set()
    if remove_referenced_call and isinstance(remove_referenced_call, str):
        ignore_fields.add(remove_referenced_call)
    # get values before deleting
    field_values: dict[str, Any] = {}
    if not skip_post_delete_hooks and self.meta.post_delete_fields.difference(ignore_fields):
        token = MODEL_GETATTR_BEHAVIOR.set("coro")
        try:
            for field_name in self.meta.post_delete_fields.difference(ignore_fields):
                try:
                    field_value = getattr(self, field_name)
                except AttributeError:
                    # already deleted
                    continue
                if inspect.isawaitable(field_value):
                    try:
                        field_value = await field_value
                    except AttributeError:
                        # already deleted
                        continue
                field_values[field_name] = field_value
        finally:
            MODEL_GETATTR_BEHAVIOR.reset(token)
    clauses = self.identifying_clauses()
    row_count = 0
    if clauses:
        expression = self.table.delete().where(*clauses)
        check_db_connection(self.database)
        async with self.database as database:
            row_count = cast(int, await database.execute(expression))
    # we cannot load anymore afterwards
    self._db_deleted = True
    # now cleanup with the saved values
    if field_values:
        token_instance = CURRENT_MODEL_INSTANCE.set(self)
        field_dict: FIELD_CONTEXT_TYPE = cast("FIELD_CONTEXT_TYPE", {})
        token_field_ctx = CURRENT_FIELD_CONTEXT.set(field_dict)
        try:
            for field_name, value in field_values.items():
                field = self.meta.fields[field_name]
                field_dict.clear()
                field_dict["field"] = field
                await field.post_delete_callback(value)
        finally:
            CURRENT_FIELD_CONTEXT.reset(token_field_ctx)
            CURRENT_MODEL_INSTANCE.reset(token_instance)
    if with_signals:
        await self.meta.signals.post_delete.send_async(
            real_class,
            instance=CURRENT_INSTANCE.get(),
            model_instance=self,
            row_count=row_count,
        )
    return row_count

delete async

delete(skip_post_delete_hooks=False)

Deletes the current model instance from the database.

This method triggers pre-delete and post-delete signals.

PARAMETER	DESCRIPTION
skip_post_delete_hooks	If True, post-delete hooks will not be executed. TYPE: bool DEFAULT: False

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

async def delete(self: Model, skip_post_delete_hooks: bool = False) -> None:
    """
    Deletes the current model instance from the database.

    This method triggers pre-delete and post-delete signals.

    Args:
        skip_post_delete_hooks: If True, post-delete hooks will not be executed.
    """
    real_class = self.get_real_class()
    await self.meta.signals.pre_delete.send_async(
        real_class, instance=self, model_instance=self
    )
    token = CURRENT_INSTANCE.set(self)
    try:
        row_count = await self.raw_delete(
            skip_post_delete_hooks=skip_post_delete_hooks,
            remove_referenced_call=False,
        )
    finally:
        CURRENT_INSTANCE.reset(token)
    await self.meta.signals.post_delete.send_async(
        real_class, instance=self, model_instance=self, row_count=row_count
    )

load_recursive async

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

Recursively loads related model instances.

:param only_needed: If True, only load if the instance is not already loaded. :param only_needed_nest: If True, stop recursion for nested instances if already loaded. :param _seen: A set of seen model keys to prevent infinite recursion.

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

async def load_recursive(
    self,
    only_needed: bool = False,
    only_needed_nest: bool = False,
    _seen: set[Any] | None = None,
) -> None:
    """
    Recursively loads related model instances.

    :param only_needed: If True, only load if the instance is not already loaded.
    :param only_needed_nest: If True, stop recursion for nested instances if already loaded.
    :param _seen: A set of seen model keys to prevent infinite recursion.
    """
    # Initialize _seen set if not provided.
    if _seen is None:
        _seen = {self.create_model_key()}
    else:
        model_key = self.create_model_key()
        # If the model key has been seen, return to prevent infinite recursion.
        if model_key in _seen:
            return
        else:
            _seen.add(model_key)
    _db_loaded_or_deleted = self._db_loaded_or_deleted
    # Load the current instance if it can be loaded.
    if self.can_load:
        await self.load(only_needed)
    # If only_needed_nest is True and the instance is already loaded or deleted, return.
    if only_needed_nest and _db_loaded_or_deleted:
        return
    # Recursively load foreign key fields.
    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 further loading for it.
            await value.load_recursive(
                only_needed=only_needed, only_needed_nest=True, _seen=_seen
            )

model_dump

model_dump(show_pk=None, **kwargs)

An updated and enhanced version of the Pydantic model_dump method.

This method provides fine-grained control over how the model's data is serialized into a dictionary. It specifically addresses: - Enforcing the inclusion of primary key fields. - Correctly handling fields marked for exclusion. - Applying custom logic for fields that retrieve their values via getters or require special serialization (e.g., related models, composite fields).

PARAMETER	DESCRIPTION
self	The instance of the Pydantic BaseModel on which model_dump is called. TYPE: BaseModel
show_pk	An optional boolean flag. If True, the primary key field(s) will always be included in the dumped dictionary, even if they are otherwise excluded or not explicitly included. If None, the default behavior defined by self.__show_pk__ will be used. TYPE: bool | None DEFAULT: None
**kwargs	Arbitrary keyword arguments that are passed directly to the underlying Pydantic super().model_dump method. These can include exclude, include, mode, etc. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
dict[str, Any]	A dict representing the serialized model data, with applied
dict[str, Any]	inclusions, exclusions, and special field handling.

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

def model_dump(self: BaseModel, show_pk: bool | None = None, **kwargs: Any) -> dict[str, Any]:
    """
    An updated and enhanced version of the Pydantic `model_dump` method.

    This method provides fine-grained control over how the model's data is
    serialized into a dictionary. It specifically addresses:
    -   Enforcing the inclusion of primary key fields.
    -   Correctly handling fields marked for exclusion.
    -   Applying custom logic for fields that retrieve their values via getters
        or require special serialization (e.g., related models, composite fields).

    Args:
        self: The instance of the Pydantic `BaseModel` on which `model_dump` is called.
        show_pk: An optional boolean flag. If `True`, the primary key field(s) will
                 always be included in the dumped dictionary, even if they are
                 otherwise excluded or not explicitly included. If `None`, the
                 default behavior defined by `self.__show_pk__` will be used.
        **kwargs: Arbitrary keyword arguments that are passed directly to the
                  underlying Pydantic `super().model_dump` method. These can
                  include `exclude`, `include`, `mode`, etc.

    Returns:
        A `dict` representing the serialized model data, with applied
        inclusions, exclusions, and special field handling.
    """
    meta = self.meta
    # Retrieve the 'exclude' argument, defaulting to None if not provided.
    # This argument can be a set of field names, a dictionary mapping field
    # names to boolean exclusion flags, or None.
    _exclude: set[str] | dict[str, Any] | None = kwargs.pop("exclude", None)

    # Initialize variables for managing field exclusions.
    # `initial_full_field_exclude` will contain names of fields to be fully excluded.
    # `exclude_passed` is a dictionary used for the first pass of `super().model_dump`.
    # `exclude_second_pass` is used for fields processed in the second pass (e.g., getters).
    if _exclude is None:
        initial_full_field_exclude: set[str] = _empty
        # Must be a writable dictionary to allow modifications.
        exclude_passed: dict[str, Any] = {}
        exclude_second_pass: dict[str, Any] = {}
    elif isinstance(_exclude, dict):
        # If `_exclude` is a dictionary, extract fields marked for full exclusion (value is True).
        initial_full_field_exclude = {k for k, v in _exclude.items() if v is True}
        # Create a copy for `exclude_passed` to avoid modifying the original `_exclude`.
        exclude_passed = copy.copy(_exclude)
        exclude_second_pass = _exclude
    else:
        # If `_exclude` is a set or list, convert it to a set for consistency.
        initial_full_field_exclude = set(_exclude)
        # Create dictionaries where all initially excluded fields are marked `True`.
        exclude_passed = dict.fromkeys(initial_full_field_exclude, True)
        exclude_second_pass = exclude_passed.copy()

    # `need_second_pass` will store field names that require a second processing pass.
    # These are typically fields with custom getters or foreign keys to models
    # that need special serialization.
    need_second_pass: set[str] = set()

    # Process fields that have special getter methods defined.
    for field_name in meta.special_getter_fields:
        # Temporarily exclude these fields from the initial `model_dump` pass.
        exclude_passed[field_name] = True
        # If a special getter field was not explicitly excluded and is not marked
        # for exclusion in its `MetaInfo`, add it to `need_second_pass`.
        if (
            field_name not in initial_full_field_exclude
            and not meta.fields[field_name].exclude
        ):
            need_second_pass.add(field_name)

    # Process foreign key fields.
    for field_name in meta.foreign_key_fields:
        field = meta.fields[field_name]
        # If the foreign key field is already fully excluded, skip further processing.
        if field_name in initial_full_field_exclude or field.exclude:
            continue
        # If the target model of the foreign key needs special serialization,
        # temporarily exclude it from the first pass and add to `need_second_pass`.
        if field.target.meta.needs_special_serialization:
            exclude_passed[field_name] = True
            need_second_pass.add(field_name)

    # Retrieve the 'include' argument, defaulting to None.
    include: set[str] | dict[str, Any] | None = kwargs.pop("include", None)
    # Determine the serialization mode ('json' or 'python').
    mode: Literal["json", "python"] = kwargs.pop("mode", "python")

    # Determine if the primary key should be shown, preferring the `show_pk`
    # argument over the model's internal `__show_pk__` attribute.
    should_show_pk = self.__show_pk__ if show_pk is None else show_pk

    # Perform the initial `model_dump` using Pydantic's default implementation.
    # This will handle most fields and apply the initial exclusion rules.
    result_dict: dict[str, Any] = super().model_dump(
        exclude=exclude_passed, include=include, mode=mode, **kwargs
    )

    # Set a context variable to control the behavior of `getattr` during the
    # second pass, often used to prevent recursive database queries.
    token = MODEL_GETATTR_BEHAVIOR.set("passdown")
    try:
        # Process fields identified in `need_second_pass`.
        for field_name in need_second_pass:
            # Skip the field if it's not a primary key (and `show_pk` is true)
            # or if it's not explicitly included (and `include` is not None).
            if not (
                (should_show_pk and field_name in self.pknames)
                or include is None
                or field_name in include
            ):
                continue

            field = meta.fields[field_name]
            try:
                # Attempt to get the value of the field, which might trigger a getter.
                retval = getattr(self, field_name)
            except AttributeError:
                # If the attribute doesn't exist (e.g., not loaded), skip it.
                continue

            sub_include = None
            # If `include` is a dictionary, extract specific inclusion rules for the sub-field.
            if isinstance(include, dict):
                sub_include = include.get(field_name, None)
                # If the sub-field is explicitly included with `True`, treat it as no specific
                # sub-inclusion (i.e., include all sub-fields by default).
                if sub_include is True:
                    sub_include = None

            # Get specific exclusion rules for the sub-field.
            sub_exclude = exclude_second_pass.get(field_name, None)
            # Ensure that a field marked for full exclusion in the first pass is not
            # unexpectedly processed in the second pass.
            assert sub_exclude is not True, "field should have been excluded"

            # If the retrieved value is another `BaseModel` (e.g., a related object),
            # recursively call `model_dump` on it.
            if isinstance(retval, BaseModel):
                retval = retval.model_dump(
                    include=sub_include, exclude=sub_exclude, mode=mode, **kwargs
                )
            else:
                # For non-BaseModel values, `sub_include` and `sub_exclude` should not be present
                # as they are only applicable to nested Pydantic models.
                assert sub_include is None, "sub include filters for no pydantic model"
                assert sub_exclude is None, "sub exclude filters for no pydantic model"
                # If the mode is 'json' and the field is not marked for unsafe JSON serialization,
                # skip it. This prevents non-serializable types from breaking JSON output.
                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

            # Determine the alias for the field in the dumped dictionary.
            # Prioritize `serialization_alias`, then `alias`, otherwise use the field name.
            alias: str = field_name
            if getattr(field, "serialization_alias", None):
                alias = cast(str, field.serialization_alias)
            elif getattr(field, "alias", None):
                alias = field.alias
            # Add the processed field and its value to the `result_dict`.
            result_dict[alias] = retval
    finally:
        # Reset the `MODEL_GETATTR_BEHAVIOR` context variable to its previous state.
        MODEL_GETATTR_BEHAVIOR.reset(token)
    return result_dict

build classmethod

build(schema=None, metadata=None)

Builds and returns the SQLAlchemy table representation for the model.

This method constructs the sqlalchemy.Table object, including columns, unique constraints, indexes, and global constraints, based on the model's meta information and specified schema.

PARAMETER	DESCRIPTION
schema	An optional schema name to apply to the table. TYPE: str | None DEFAULT: None
metadata	An optional sqlalchemy.MetaData object to use. If None, the registry's metadata is used. TYPE: MetaData | None DEFAULT: None

RETURNS	DESCRIPTION
Table	The constructed SQLAlchemy Table object.

RAISES	DESCRIPTION
AssertionError	If the model's registry is not set.

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

@classmethod
def build(
    cls,
    schema: str | None = None,
    metadata: sqlalchemy.MetaData | None = None,
) -> sqlalchemy.Table:
    """
    Builds and returns the SQLAlchemy table representation for the model.

    This method constructs the `sqlalchemy.Table` object, including columns,
    unique constraints, indexes, and global constraints, based on the model's
    meta information and specified schema.

    Args:
        schema: An optional schema name to apply to the table.
        metadata: An optional `sqlalchemy.MetaData` object to use. If None,
                  the registry's metadata is used.

    Returns:
        The constructed SQLAlchemy `Table` object.

    Raises:
        AssertionError: If the model's registry is not set.
    """
    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)
        if not NO_GLOBAL_FIELD_CONSTRAINTS.get():
            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, is_update)

Executes post-save hooks for relevant fields.

:param fields: A sequence of field names that were affected by the save operation. :param is_update: True if the operation was an update, False for creation.

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

async def execute_post_save_hooks(self, fields: Sequence[str], is_update: bool) -> None:
    """
    Executes post-save hooks for relevant fields.

    :param fields: A sequence of field names that were affected by the save operation.
    :param is_update: True if the operation was an update, False for creation.
    """
    # Determine affected fields by intersecting with post-save fields.
    affected_fields = self.meta.post_save_fields.intersection(fields)
    if affected_fields:
        # Set MODEL_GETATTR_BEHAVIOR to "passdown" to prevent triggering loads.
        token = MODEL_GETATTR_BEHAVIOR.set("passdown")
        token2 = CURRENT_MODEL_INSTANCE.set(self)
        field_dict: FIELD_CONTEXT_TYPE = cast("FIELD_CONTEXT_TYPE", {})
        token_field_ctx = CURRENT_FIELD_CONTEXT.set(field_dict)
        try:
            for field_name in affected_fields:
                field = self.meta.fields[field_name]
                try:
                    # Attempt to get the field value.
                    value = getattr(self, field_name)
                except AttributeError:
                    # Skip if the attribute is not found.
                    continue
                field_dict.clear()
                field_dict["field"] = field
                # Execute post-save callback for the field.
                await field.post_save_callback(value, is_update=is_update)
        finally:
            # Reset context variables.
            CURRENT_FIELD_CONTEXT.reset(token_field_ctx)
            MODEL_GETATTR_BEHAVIOR.reset(token)
            CURRENT_MODEL_INSTANCE.reset(token2)

execute_pre_save_hooks async

execute_pre_save_hooks(column_values, original, is_update)

Executes pre-save hooks for relevant fields.

:param column_values: Dictionary of new column values. :param original: Dictionary of original column values. :param is_update: True if the operation is an update, False for creation. :return: A dictionary of values returned by pre-save callbacks.

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], is_update: bool
) -> dict[str, Any]:
    """
    Executes pre-save hooks for relevant fields.

    :param column_values: Dictionary of new column values.
    :param original: Dictionary of original column values.
    :param is_update: True if the operation is an update, False for creation.
    :return: A dictionary of values returned by pre-save callbacks.
    """
    # also handle defaults
    # Combine keys from new and original column values to identify affected fields.
    keys = {*column_values.keys(), *original.keys()}
    affected_fields = self.meta.pre_save_fields.intersection(keys)
    retdict: dict[str, Any] = {}
    if affected_fields:
        # Set MODEL_GETATTR_BEHAVIOR to "passdown" to prevent triggering loads.
        token = MODEL_GETATTR_BEHAVIOR.set("passdown")
        token2 = CURRENT_MODEL_INSTANCE.set(self)
        field_dict: FIELD_CONTEXT_TYPE = cast("FIELD_CONTEXT_TYPE", {})
        token_field_ctx = CURRENT_FIELD_CONTEXT.set(field_dict)
        try:
            for field_name in affected_fields:
                # Skip if the field is not in new or original values.
                if field_name not in column_values and field_name not in original:
                    continue
                field = self.meta.fields[field_name]
                field_dict.clear()
                field_dict["field"] = field
                # Execute pre-save callback for the field.
                retdict.update(
                    await field.pre_save_callback(
                        column_values.get(field_name),
                        original.get(field_name),
                        is_update=is_update,
                    )
                )
        finally:
            # Reset context variables.
            CURRENT_FIELD_CONTEXT.reset(token_field_ctx)
            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, evaluate_values=False)

Extracts and validates column values, applying transformations and defaults.

:param extracted_values: Dictionary of values to extract and validate. :param is_update: True if the operation is an update. :param is_partial: True if it's a partial update/creation. :param phase: The current phase of extraction. :param instance: The current instance being processed. :param model_instance: The model instance context. :param evaluate_values: If True, callable values in extracted_values will be evaluated. :return: A dictionary of validated column values.

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: BaseModelType | QuerySet | None = None,
    model_instance: BaseModelType | None = None,
    evaluate_values: bool = False,
) -> dict[str, Any]:
    """
    Extracts and validates column values, applying transformations and defaults.

    :param extracted_values: Dictionary of values to extract and validate.
    :param is_update: True if the operation is an update.
    :param is_partial: True if it's a partial update/creation.
    :param phase: The current phase of extraction.
    :param instance: The current instance being processed.
    :param model_instance: The model instance context.
    :param evaluate_values: If True, callable values in `extracted_values` will be
                             evaluated.
    :return: A dictionary of validated column values.
    """
    validated: dict[str, Any] = {}
    # Set context variables for phase, current instance, and model instance.
    token = CURRENT_PHASE.set(phase)
    token2 = CURRENT_INSTANCE.set(instance)
    token3 = CURRENT_MODEL_INSTANCE.set(model_instance)
    field_dict: FIELD_CONTEXT_TYPE = cast("FIELD_CONTEXT_TYPE", {})
    token_field_ctx = CURRENT_FIELD_CONTEXT.set(field_dict)

    try:
        # Phase 1: Evaluate callable values if `evaluate_values` is True.
        if evaluate_values:
            new_extracted_values = {}
            for k, v in extracted_values.items():
                if callable(v):
                    field_dict.clear()
                    field_dict["field"] = cast("BaseFieldType", cls.meta.fields.get(k))
                    v = v()
                new_extracted_values[k] = v
            extracted_values = new_extracted_values
        else:
            extracted_values = {**extracted_values}
        # Phase 2: Apply input modifying fields.
        if cls.meta.input_modifying_fields:
            for field_name in cls.meta.input_modifying_fields:
                cls.meta.fields[field_name].modify_input(field_name, extracted_values)
        # Phase 3: Validate fields and set defaults for read-only fields.
        need_second_pass: list[BaseFieldType] = []
        for field_name, field in cls.meta.fields.items():
            field_dict.clear()
            field_dict["field"] = field
            if field.read_only:
                # If read-only, and not a partial update or inject_default_on_partial_update,
                # and has a default, apply default values.
                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
                # Clean and update validated values.
                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 fields with defaults to a second pass if not partial
                # or if inject_default_on_partial_update is set for updates.
                need_second_pass.append(field)

        # Phase 4: Set defaults for remaining fields if necessary.
        if need_second_pass:
            for field in need_second_pass:
                field_dict.clear()
                field_dict["field"] = field
                # Check if field appeared (e.g., by composite) before setting default.
                if field.name not in validated:
                    validated.update(field.get_default_values(field.name, validated))
    finally:
        # Reset context variables.
        CURRENT_FIELD_CONTEXT.reset(token_field_ctx)
        CURRENT_MODEL_INSTANCE.reset(token3)
        CURRENT_INSTANCE.reset(token2)
        CURRENT_PHASE.reset(token)
    return validated

get_real_class classmethod

get_real_class()

Returns the concrete (non-proxy) class of the model instance.

If the current instance is a proxy model, it returns its parent (the original model class). Otherwise, it returns the class itself.

RETURNS	DESCRIPTION
type[BaseModelType]	type[BaseModelType]: The real, non-proxy class of the model.

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

@classmethod
def get_real_class(cls) -> type[BaseModelType]:
    """
    Returns the concrete (non-proxy) class of the model instance.

    If the current instance is a proxy model, it returns its parent (the original
    model class). Otherwise, it returns the class itself.

    Returns:
        type[BaseModelType]: The real, non-proxy class of the model.
    """
    # Return the parent class if this is a proxy model, otherwise return the class itself.
    return cls.__parent__ if cls.__is_proxy_model__ else cls

extract_db_fields

extract_db_fields(only=None)

Extracts and returns a dictionary of database-related fields from the model instance.

This includes direct model fields and SQLAlchemy column attributes, but excludes related fields which are handled separately due to their disjoint nature.

PARAMETER	DESCRIPTION
only	An optional container of field names to include in the extraction. If None, all relevant fields are extracted. Defaults to None. TYPE: Container[str] | None DEFAULT: None

RETURNS	DESCRIPTION
dict[str, Any]	dict[str, Any]: A dictionary where keys are field names and values are their corresponding database values.

RAISES	DESCRIPTION
AssertionError	If only contains field names that do not exist in the model's fields or as SQLAlchemy columns.

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

def extract_db_fields(self, only: Container[str] | None = None) -> dict[str, Any]:
    """
    Extracts and returns a dictionary of database-related fields from the model instance.

    This includes direct model fields and SQLAlchemy column attributes, but excludes
    related fields which are handled separately due to their disjoint nature.

    Args:
        only (Container[str] | None): An optional container of field names to include
                                      in the extraction. If `None`, all relevant fields
                                      are extracted. Defaults to `None`.

    Returns:
        dict[str, Any]: A dictionary where keys are field names and values are their
                        corresponding database values.

    Raises:
        AssertionError: If `only` contains field names that do not exist in the model's
                        fields or as SQLAlchemy columns.
    """
    # Get all defined fields from the model's meta information.
    fields = self.meta.fields
    # Get the SQLAlchemy columns associated with the model's table.
    columns = self.table.columns

    # If `only` is specified, filter the dictionary to include only the specified keys.
    # An assertion ensures that all keys in `only` are valid fields or column attributes.
    if only is not None:
        assert all(k in fields or hasattr(columns, k) for k in only), (
            f'"only" includes invalid fields, {only}'
        )
        return {k: v for k, v in self.__dict__.items() if k in only}

    # If `only` is not specified, return all attributes that are either model fields
    # or SQLAlchemy column attributes.
    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 lowercase name of the model's class.

This is typically used for generating default table names or for identification purposes.

RETURNS	DESCRIPTION
str	The lowercase name of the model's class. TYPE: str

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

def get_instance_name(self) -> str:
    """
    Returns the lowercase name of the model's class.

    This is typically used for generating default table names or for
    identification purposes.

    Returns:
        str: The lowercase name of the model's class.
    """
    return type(self).__name__.lower()

create_model_key

create_model_key()

Generates a unique cache key for the model instance.

The key is composed of the model's class name and the string representation of its primary key column values. This key can be used for caching model instances to improve performance.

RETURNS	DESCRIPTION
tuple	A tuple representing the unique cache key for the model instance. TYPE: tuple

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

def create_model_key(self) -> tuple:
    """
    Generates a unique cache key for the model instance.

    The key is composed of the model's class name and the string representation
    of its primary key column values. This key can be used for caching model
    instances to improve performance.

    Returns:
        tuple: A tuple representing the unique cache key for the model instance.
    """
    # Start the key with the model's class name.
    pk_key_list: list[Any] = [type(self).__name__]
    # Iterate over primary key column names and append their string values to the key list.
    # Note: `pkcolumns` contains column names, not column objects.
    for attr in self.pkcolumns:
        pk_key_list.append(str(getattr(self, attr)))
    # Convert the list to a tuple to make it hashable for use as a dictionary key.
    return tuple(pk_key_list)

transform_input classmethod

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

Transforms input keyword arguments by applying field-specific modifications and to_model transformations.

:param kwargs: The input keyword arguments to transform. :param phase: The current phase of transformation. :param instance: The model instance being transformed. :param drop_extra_kwargs: If True, extra kwargs not defined in model fields will be dropped. :return: The transformed keyword arguments.

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

@classmethod
def transform_input(
    cls,
    kwargs: dict[str, Any],
    phase: str = "",
    instance: BaseModelType | None = None,
    drop_extra_kwargs: bool = False,
) -> Any:
    """
    Transforms input keyword arguments by applying field-specific modifications
    and `to_model` transformations.

    :param kwargs: The input keyword arguments to transform.
    :param phase: The current phase of transformation.
    :param instance: The model instance being transformed.
    :param drop_extra_kwargs: If True, extra kwargs not defined in model fields
                               will be dropped.
    :return: The transformed keyword arguments.
    """

    # Create a copy of kwargs to avoid modifying the original input.
    kwargs = kwargs.copy()
    new_kwargs: dict[str, Any] = {}

    fields = cls.meta.fields
    # Set context variables for the current instance, model instance, and phase.
    token = CURRENT_INSTANCE.set(instance)
    token2 = CURRENT_MODEL_INSTANCE.set(instance)
    token3 = CURRENT_PHASE.set(phase)
    try:
        # Phase 1: Apply input modifying fields.
        for field_name in cls.meta.input_modifying_fields:
            fields[field_name].modify_input(field_name, kwargs)
        # Phase 2: Apply `to_model` transformations.
        for key, value in kwargs.items():
            field = fields.get(key, None)
            if field is not None:
                # If a field exists, apply its to_model transformation.
                new_kwargs.update(**field.to_model(key, value))
            elif not drop_extra_kwargs:
                # If no field and not dropping extra kwargs, keep the value.
                new_kwargs[key] = value
    finally:
        # Reset context variables.
        CURRENT_PHASE.reset(token3)
        CURRENT_MODEL_INSTANCE.reset(token2)
        CURRENT_INSTANCE.reset(token)
    return new_kwargs

join_identifiers_to_string

join_identifiers_to_string(*, sep=', ', sep_inner='=')

Joins the identifying database fields and their values into a string.

:param sep: Separator for multiple identifier-value pairs. :param sep_inner: Separator between identifier and its value. :return: A string representation of identifying fields.

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

def join_identifiers_to_string(self, *, sep: str = ", ", sep_inner: str = "=") -> str:
    """
    Joins the identifying database fields and their values into a string.

    :param sep: Separator for multiple identifier-value pairs.
    :param sep_inner: Separator between identifier and its value.
    :return: A string representation of identifying fields.
    """
    pkl = []
    # Set MODEL_GETATTR_BEHAVIOR to "passdown" to prevent triggering loads.
    token = MODEL_GETATTR_BEHAVIOR.set("passdown")
    try:
        for identifier in self.identifying_db_fields:
            # Append identifier and its value to the list.
            pkl.append(f"{identifier}{sep_inner}{getattr(self, identifier, None)}")
    except AttributeError:
        # Handle abstract embedded cases where identifiers might not be directly available.
        pass
    finally:
        # Reset MODEL_GETATTR_BEHAVIOR.
        MODEL_GETATTR_BEHAVIOR.reset(token)
    return sep.join(pkl)

__setattr__

__setattr__(key, value)

Custom __setattr__ method to handle specific attribute assignments.

If the __using_schema__ attribute is set, it clears the cached _table to ensure the table is rebuilt with the new schema.

PARAMETER	DESCRIPTION
key	The name of the attribute to set. TYPE: str
value	The value to assign to the attribute. TYPE: Any

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

def __setattr__(self, key: str, value: Any) -> None:
    """
    Custom `__setattr__` method to handle specific attribute assignments.

    If the `__using_schema__` attribute is set, it clears the cached
    `_table` to ensure the table is rebuilt with the new schema.

    Args:
        key: The name of the attribute to set.
        value: The value to assign to the attribute.
    """
    if key == "__using_schema__":
        self._edgy_namespace.pop("_table", None)
    super().__setattr__(key, value)

_agetattr_helper async

_agetattr_helper(name, getter)

Asynchronous helper for getattr to load the model and retrieve attributes.

:param name: The name of the attribute to retrieve. :param getter: The getter method for the attribute, if available. :return: The value of the attribute. :raises AttributeError: If the attribute is not found.

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

async def _agetattr_helper(self, name: str, getter: Any) -> Any:
    """
    Asynchronous helper for __getattr__ to load the model and retrieve attributes.

    :param name: The name of the attribute to retrieve.
    :param getter: The getter method for the attribute, if available.
    :return: The value of the attribute.
    :raises AttributeError: If the attribute is not found.
    """
    # Load the model data asynchronously.
    await self.load()
    if getter is not None:
        # If a getter is provided, use it, handling awaitable results.
        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:
        # Attempt to retrieve from __dict__.
        return self.__dict__[name]
    except KeyError:
        raise AttributeError(f"Attribute: {name} not found") from None

__getattribute__

__getattribute__(name)

Custom getter for model attributes.

Handles retrieving Edgy's private attributes from _edgy_namespace.

:param name: The name of the attribute to retrieve. :return: The value of the attribute. :raises AttributeError: If the attribute is not found in private namespace.

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

def __getattribute__(self, name: str) -> Any:
    """
    Custom getter for model attributes.

    Handles retrieving Edgy's private attributes from `_edgy_namespace`.

    :param name: The name of the attribute to retrieve.
    :return: The value of the attribute.
    :raises AttributeError: If the attribute is not found in private namespace.
    """
    # If the attribute is an Edgy private attribute and not the private attributes set itself,
    # try to retrieve it from _edgy_namespace.
    if name != "_edgy_private_attrs" and name in self._edgy_private_attrs:
        try:
            return self._edgy_namespace[name]
        except KeyError as exc:
            raise AttributeError from exc
    # For all other attributes, use the default __getattribute__ behavior.
    return super().__getattribute__(name)

__getattr__

__getattr__(name)

Custom getter for model attributes when not found through normal lookup.

This method handles: 1. Initialization of managers on first access. 2. Redirection of attribute access to getter fields. 3. Triggering a one-off database query to populate foreign key relationships, ensuring it runs only once per foreign key.

:param name: The name of the attribute to retrieve. :return: The value of the attribute.

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

def __getattr__(self, name: str) -> Any:
    """
    Custom getter for model attributes when not found through normal lookup.

    This method handles:
    1. Initialization of managers on first access.
    2. Redirection of attribute access to getter fields.
    3. Triggering a one-off database query to populate foreign key relationships,
       ensuring it runs only once per foreign key.

    :param name: The name of the attribute to retrieve.
    :return: The value of the attribute.
    """
    # Attributes exempted from triggering special __getattr__ logic.
    if name in _excempted_attrs or name in self._edgy_private_attrs:
        return super().__getattr__(name)

    behavior = MODEL_GETATTR_BEHAVIOR.get()
    manager = self.meta.managers.get(name)
    if manager is not None:
        # Initialize and cache manager instances on first access.
        if name not in self._edgy_namespace:
            manager = copy.copy(manager)
            manager.instance = self
            self._edgy_namespace[name] = manager
        return self._edgy_namespace[name]

    field = self.meta.fields.get(name)
    if field is not None:
        token_field_ctx = CURRENT_FIELD_CONTEXT.set(
            cast("FIELD_CONTEXT_TYPE", {"field": field})
        )
    try:
        getter: Any = None
        if field is not None and hasattr(field, "__get__"):
            getter = field.__get__
            # If behavior is "coro" or "passdown", return the getter result directly.
            if behavior == "coro" or behavior == "passdown":
                return field.__get__(self, self.__class__)
            else:
                # Otherwise, set "passdown" behavior and try to get the field value.
                token = MODEL_GETATTR_BEHAVIOR.set("passdown")
                try:
                    return field.__get__(self, self.__class__)
                except AttributeError:
                    # If AttributeError, forward to the load routine.
                    pass
                finally:
                    MODEL_GETATTR_BEHAVIOR.reset(token)
        # If the attribute is not in __dict__, not in "passdown" behavior,
        # not already loaded/deleted, and is a loadable field, trigger a load.
        if (
            name not in self.__dict__
            and behavior != "passdown"
            # is already loaded or deleted
            and not self._db_loaded_or_deleted
            # only load when it is a field except for reflected
            and (field is not None or self.__reflected__)
            # exclude attr names from triggering load
            and name not in getattr(self, "__no_load_trigger_attrs__", _empty)
            and name not in self.identifying_db_fields
            and self.can_load
        ):
            coro = self._agetattr_helper(name, getter)
            # If behavior is "coro", return the coroutine directly.
            if behavior == "coro":
                return coro
            # Otherwise, run the coroutine synchronously.
            return run_sync(coro)
        # If none of the above, use the default __getattr__ behavior (will raise AttributeError).
        return super().__getattr__(name)
    finally:
        # Reset CURRENT_FIELD_CONTEXT if a field was involved.
        if field:
            CURRENT_FIELD_CONTEXT.reset(token_field_ctx)

__delattr__

__delattr__(name)

Custom deleter for model attributes.

Handles deleting Edgy's private attributes from _edgy_namespace.

:param name: The name of the attribute to delete. :raises AttributeError: If the attribute is not found in private namespace.

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

def __delattr__(self, name: str) -> None:
    """
    Custom deleter for model attributes.

    Handles deleting Edgy's private attributes from `_edgy_namespace`.

    :param name: The name of the attribute to delete.
    :raises AttributeError: If the attribute is not found in private namespace.
    """
    # If the attribute is an Edgy private attribute, try to delete it from _edgy_namespace.
    if name in self._edgy_private_attrs:
        try:
            del self._edgy_namespace[name]
            return
        except KeyError as exc:
            raise AttributeError from exc
    # For all other attributes, use the default __delattr__ behavior.
    super().__delattr__(name)

__eq__

__eq__(other)

Compares two EdgyBaseModel instances for equality.

Equality is determined by comparing their table name, registry, and the values of their identifying database fields.

:param other: The other object to compare with. :return: True if the instances are equal, False otherwise.

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

def __eq__(self, other: Any) -> bool:
    """
    Compares two EdgyBaseModel instances for equality.

    Equality is determined by comparing their table name, registry,
    and the values of their identifying database fields.

    :param other: The other object to compare with.
    :return: True if the instances are equal, False otherwise.
    """
    # If the other object is not an instance of EdgyBaseModel, they are not equal.
    if not isinstance(other, EdgyBaseModel):
        return False
    # Compare registry and table name for quick inequality checks.
    if self.meta.registry is not other.meta.registry:
        return False
    if self.meta.tablename != other.meta.tablename:
        return False

    # Extract identifying column values for comparison, handling partial extraction.
    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,
    )
    # Get all unique keys from both dictionaries.
    key_set = {*self_dict.keys(), *other_dict.keys()}
    # Compare values for each key. If any mismatch, return False.
    for field_name in key_set:
        if self_dict.get(field_name) != other_dict.get(field_name):
            return False
    # If all identifying field values match, the instances are considered equal.
    return True

model_dump_json

model_dump_json(**kwargs)

Dumps the model data into a JSON string.

This method leverages model_dump with mode="json" and then uses orjson for efficient JSON serialization, which is faster than Python's built-in json module.

PARAMETER	DESCRIPTION
self	The instance of the BaseModel to be dumped.
**kwargs	Arbitrary keyword arguments passed to model_dump. These can control inclusions, exclusions, and other dumping behaviors. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
str	A str representing the JSON serialization of the model data.

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

def model_dump_json(self, **kwargs: Any) -> str:
    """
    Dumps the model data into a JSON string.

    This method leverages `model_dump` with `mode="json"` and then uses `orjson`
    for efficient JSON serialization, which is faster than Python's built-in `json` module.

    Args:
        self: The instance of the `BaseModel` to be dumped.
        **kwargs: Arbitrary keyword arguments passed to `model_dump`. These can
                  control inclusions, exclusions, and other dumping behaviors.

    Returns:
        A `str` representing the JSON serialization of the model data.
    """
    return orjson.dumps(self.model_dump(mode="json", **kwargs)).decode()

get_admin_marshall_config classmethod

get_admin_marshall_config(*, phase, for_schema)

Generates a dictionary representing the marshall configuration for the admin interface.

This configuration dictates how model fields are handled during different administrative operations (e.g., creation, update) and schema generation.

PARAMETER	DESCRIPTION
cls	The Edgy Model class for which the marshall configuration is being generated. TYPE: type[Model]
phase	The current phase of the administrative operation, typically 'create', 'update', or 'read'. This influences field exclusions and read-only states. TYPE: str
for_schema	A boolean indicating whether the configuration is intended for schema generation (e.g., OpenAPI documentation) or for actual data marshalling. TYPE: bool

RETURNS	DESCRIPTION
dict[str, Any]	A dictionary containing the marshall configuration, including
dict[str, Any]	specifications for fields, read-only exclusions, primary key
dict[str, Any]	read-only status, and autoincrement exclusions.

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

@classmethod
def get_admin_marshall_config(
    cls: type[Model], *, phase: str, for_schema: bool
) -> dict[str, Any]:
    """
    Generates a dictionary representing the marshall configuration for
    the admin interface.

    This configuration dictates how model fields are handled during different
    administrative operations (e.g., creation, update) and schema generation.

    Args:
        cls: The Edgy Model class for which the marshall configuration is
            being generated.
        phase: The current phase of the administrative operation, typically
            'create', 'update', or 'read'. This influences field exclusions
            and read-only states.
        for_schema: A boolean indicating whether the configuration is
            intended for schema generation (e.g., OpenAPI documentation)
            or for actual data marshalling.

    Returns:
        A dictionary containing the marshall configuration, including
        specifications for fields, read-only exclusions, primary key
        read-only status, and autoincrement exclusions.
    """
    return {
        "fields": ["__all__"],  # Include all fields by default.
        # Exclude read-only fields during 'create' or 'update' phases.
        "exclude_read_only": phase in {"create", "update"},
        # Primary key is read-only unless in the 'create' phase.
        "primary_key_read_only": phase != "create",
        # Exclude autoincrement fields only when creating a new instance.
        "exclude_autoincrement": phase == "create",
    }

get_admin_marshall_class classmethod

get_admin_marshall_class(*, phase, for_schema=False)

Generates a dynamic Marshall class specifically for the admin interface.

This allows for custom marshalling behavior based on the current administrative operation phase and whether it's for schema generation.

PARAMETER	DESCRIPTION
cls	The Edgy Model class for which the admin marshall class is being generated. TYPE: type[Model]
phase	The current phase of the administrative operation (e.g., 'create', 'update', 'read'). This phase is used to configure the underlying marshall_config. TYPE: str
for_schema	A boolean indicating whether the generated marshall class is intended for schema generation. If True, additional properties are forbidden, aligning with strict schema definitions. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
type[Marshall]	A dynamically created subclass of marshalls.Marshall configured
type[Marshall]	with the appropriate ConfigDict and marshall_config for the
type[Marshall]	admin interface.

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

@classmethod
def get_admin_marshall_class(
    cls: type[Model], *, phase: str, for_schema: bool = False
) -> type[marshalls.Marshall]:
    """
    Generates a dynamic Marshall class specifically for the admin interface.

    This allows for custom marshalling behavior based on the current
    administrative operation phase and whether it's for schema generation.

    Args:
        cls: The Edgy Model class for which the admin marshall class is
            being generated.
        phase: The current phase of the administrative operation (e.g.,
            'create', 'update', 'read'). This phase is used to configure
            the underlying marshall_config.
        for_schema: A boolean indicating whether the generated marshall
            class is intended for schema generation. If True, additional
            properties are forbidden, aligning with strict schema definitions.

    Returns:
        A dynamically created subclass of `marshalls.Marshall` configured
        with the appropriate `ConfigDict` and `marshall_config` for the
        admin interface.
    """

    class AdminMarshall(marshalls.Marshall):
        # Configure Pydantic model behavior.
        # 'title' is set to the model's name for clarity in schemas.
        # 'extra="forbid"' prevents unknown fields when generating schemas.
        model_config: ClassVar[ConfigDict] = ConfigDict(
            title=cls.__name__, extra="forbid" if for_schema else None
        )
        # Initialize the marshall configuration using the model and
        # the admin-specific configuration.
        marshall_config = marshalls.ConfigMarshall(
            model=cls,
            **cls.get_admin_marshall_config(phase=phase, for_schema=for_schema),  # type: ignore
        )

    return AdminMarshall

get_admin_marshall_for_save classmethod

get_admin_marshall_for_save(instance=None, /, **kwargs)

Generates a Marshall instance for saving (creating or updating) an Edgy model through the admin interface.

This method determines the appropriate phase ('create' or 'update') based on whether an instance is provided, and then creates a corresponding AdminMarshall instance.

PARAMETER	DESCRIPTION
cls	The Edgy Model class for which the marshall instance is being generated. TYPE: type[Model]
instance	An optional existing model instance. If provided, the operation is considered an 'update'; otherwise, it's a 'create'. TYPE: Model | None DEFAULT: None
kwargs	Additional keyword arguments to pass to the AdminMarshall constructor. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
Marshall	An instance of AdminMarshall prepared for either creating a new
Marshall	model record or updating an existing one.

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

@classmethod
def get_admin_marshall_for_save(
    cls: type[Model], instance: Model | None = None, /, **kwargs: Any
) -> marshalls.Marshall:
    """
    Generates a Marshall instance for saving (creating or updating)
    an Edgy model through the admin interface.

    This method determines the appropriate phase ('create' or 'update')
    based on whether an instance is provided, and then creates a
    corresponding `AdminMarshall` instance.

    Args:
        cls: The Edgy Model class for which the marshall instance is
            being generated.
        instance: An optional existing model instance. If provided, the
            operation is considered an 'update'; otherwise, it's a 'create'.
        kwargs: Additional keyword arguments to pass to the `AdminMarshall`
            constructor.

    Returns:
        An instance of `AdminMarshall` prepared for either creating a new
        model record or updating an existing one.
    """
    # Determine the phase based on whether an instance is provided.
    # If an instance exists, it's an 'update' operation; otherwise, it's 'create'.
    phase = "update" if instance is not None else "create"
    # Get the appropriate AdminMarshall class for the determined phase.
    # 'for_schema' is set to False as this is for an instance, not a schema.
    AdminMarshallClass = cls.get_admin_marshall_class(phase=phase, for_schema=False)
    # Return an instance of the AdminMarshallClass, passing the model instance
    # and any additional keyword arguments.
    return AdminMarshallClass(instance, **kwargs)

add_to_registry classmethod

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

A public wrapper for real_add_to_registry.

This method provides a convenient interface for registering a model class with a given registry, forwarding all parameters to real_add_to_registry.

PARAMETER	DESCRIPTION
registry	The Registry instance to which the model will be added. TYPE: Registry
name	An optional name to assign to the model in the registry. TYPE: str DEFAULT: ''
database	Specifies how the database connection should be handled. TYPE: bool | Database | Literal['keep'] DEFAULT: 'keep'
replace_related_field	Indicates whether existing related fields should be replaced. TYPE: bool | type[BaseModelType] | tuple[type[BaseModelType], ...] | list[type[BaseModelType]] DEFAULT: False
on_conflict	Defines the behavior when a model with the same name is already registered. TYPE: Literal['keep', 'replace', 'error'] DEFAULT: 'error'

RETURNS	DESCRIPTION
type[BaseModelType]	The registered model class.

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

@classmethod
def add_to_registry(
    cls,
    registry: Registry,
    name: str = "",
    database: bool | Database | Literal["keep"] = "keep",
    *,
    replace_related_field: bool
    | type[BaseModelType]
    | tuple[type[BaseModelType], ...]
    | list[type[BaseModelType]] = False,
    on_conflict: Literal["keep", "replace", "error"] = "error",
) -> type[BaseModelType]:
    """
    A public wrapper for `real_add_to_registry`.

    This method provides a convenient interface for registering a model class
    with a given registry, forwarding all parameters to `real_add_to_registry`.

    Args:
        registry: The `Registry` instance to which the model will be added.
        name: An optional name to assign to the model in the registry.
        database: Specifies how the database connection should be handled.
        replace_related_field: Indicates whether existing related fields should
                               be replaced.
        on_conflict: Defines the behavior when a model with the same name is
                     already registered.

    Returns:
        The registered model class.
    """
    return cls.real_add_to_registry(
        registry=registry,
        name=name,
        database=database,
        replace_related_field=replace_related_field,
        on_conflict=on_conflict,
    )

get_active_instance_schema

get_active_instance_schema(check_schema=True, check_tenant=True)

Retrieves the active schema for the current model instance.

This method first checks if a schema is explicitly set for the instance. If not, it defers to get_active_class_schema to determine the schema based on class-level configurations and global context.

PARAMETER	DESCRIPTION
check_schema	If True, checks for a global schema in the context. TYPE: bool DEFAULT: True
check_tenant	If True, considers tenant-specific schemas. TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
str | None	The active schema as a string, or None if no schema is found.

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

def get_active_instance_schema(
    self, check_schema: bool = True, check_tenant: bool = True
) -> str | None:
    """
    Retrieves the active schema for the current model instance.

    This method first checks if a schema is explicitly set for the instance.
    If not, it defers to `get_active_class_schema` to determine the schema
    based on class-level configurations and global context.

    Args:
        check_schema: If True, checks for a global schema in the context.
        check_tenant: If True, considers tenant-specific schemas.

    Returns:
        The active schema as a string, or None if no schema is found.
    """
    if self._edgy_namespace["__using_schema__"] is not Undefined:
        return cast(str | None, self._edgy_namespace["__using_schema__"])
    return type(self).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)

Retrieves the active schema for the model class.

This method determines the schema based on the class's __using_schema__ attribute, global schema context, and the model's configured database schema.

PARAMETER	DESCRIPTION
check_schema	If True, checks for a global schema in the context. TYPE: bool DEFAULT: True
check_tenant	If True, considers tenant-specific schemas when checking the global schema. TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
str	The active schema as a string.

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:
    """
    Retrieves the active schema for the model class.

    This method determines the schema based on the class's `__using_schema__`
    attribute, global schema context, and the model's configured database schema.

    Args:
        check_schema: If True, checks for a global schema in the context.
        check_tenant: If True, considers tenant-specific schemas when checking
                      the global schema.

    Returns:
        The active schema as a string.
    """
    if cls.__using_schema__ is not Undefined:
        return cast(str | None, cls.__using_schema__)
    if check_schema:
        schema = get_schema(check_tenant=check_tenant)
        if schema is not None:
            return schema
    db_schema: str | None = 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='', unlink_same_registry=True, on_conflict='error', **kwargs)

Copies the model class and optionally registers it with another registry.

This method creates a deep copy of the model, including its fields and managers. It can also reconfigure foreign key and many-to-many relationships to point to models within a new registry or disable backreferences.

PARAMETER	DESCRIPTION
registry	An optional Registry instance to which the copied model should be added. If None, the model is not added to any registry. TYPE: Registry | None DEFAULT: None
name	An optional new name for the copied model. If not provided, the original model's name is used. TYPE: str DEFAULT: ''
unlink_same_registry	If True, and the registry is different from the original model's registry, foreign key targets that point to models within the original registry will be unreferenced, forcing them to be resolved within the new registry. TYPE: bool DEFAULT: True
on_conflict	Defines the behavior if a model with the same name already exists in the target registry (if registry is provided). TYPE: Literal['keep', 'replace', 'error'] DEFAULT: 'error'
**kwargs	Additional keyword arguments to pass to create_edgy_model. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
type[Self]	The newly created and copied model class.

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

@classmethod
def copy_edgy_model(
    cls: type[Model],
    registry: Registry | None = None,
    name: str = "",
    unlink_same_registry: bool = True,
    on_conflict: Literal["keep", "replace", "error"] = "error",
    **kwargs: Any,
) -> type[Self]:
    """
    Copies the model class and optionally registers it with another registry.

    This method creates a deep copy of the model, including its fields and
    managers. It can also reconfigure foreign key and many-to-many relationships
    to point to models within a new registry or disable backreferences.

    Args:
        registry: An optional `Registry` instance to which the copied model
                  should be added. If `None`, the model is not added to any registry.
        name: An optional new name for the copied model. If not provided,
              the original model's name is used.
        unlink_same_registry: If True, and the `registry` is different from the
                              original model's registry, foreign key targets
                              that point to models within the original registry
                              will be unreferenced, forcing them to be resolved
                              within the new registry.
        on_conflict: Defines the behavior if a model with the same name already
                     exists in the target registry (if `registry` is provided).
        **kwargs: Additional keyword arguments to pass to `create_edgy_model`.

    Returns:
        The newly created and copied model class.
    """
    # 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 = create_edgy_model(
        __name__=name or cls.__name__,
        __module__=cls.__module__,
        __definitions__=attrs,
        __metadata__=cls.meta,
        __bases__=cls.__bases__,
        __type_kwargs__={**kwargs, "skip_registry": True},
    )
    # should also allow masking database with None
    if hasattr(cls, "database"):
        _copy.database = cls.database
    replaceable_models: list[type[BaseModelType]] = [cls]
    if cls.meta.registry:
        for field_name in list(_copy.meta.fields):
            src_field = cls.meta.fields.get(field_name)
            if not isinstance(src_field, BaseForeignKey):
                continue
            # we use the target of source
            replaceable_models.append(src_field.target)

            if src_field.target_registry is cls.meta.registry:
                # clear target_registry, for obvious registries
                del _copy.meta.fields[field_name].target_registry
            if unlink_same_registry and src_field.target_registry is cls.meta.registry:
                # we need to unreference so the target is retrieved from the new registry

                _copy.meta.fields[field_name].target = src_field.target.__name__
            else:
                # otherwise we need to disable backrefs
                _copy.meta.fields[field_name].related_name = False

            if isinstance(src_field, BaseManyToManyForeignKeyField):
                _copy.meta.fields[field_name].through = src_field.through_original
                # clear through registry, we need a copy in the new registry
                del _copy.meta.fields[field_name].through_registry
                if (
                    isinstance(_copy.meta.fields[field_name].through, type)
                    and issubclass(_copy.meta.fields[field_name].through, BaseModelType)
                    and not _copy.meta.fields[field_name].through.meta.abstract
                ):
                    # unreference
                    _copy.meta.fields[field_name].through = through_model = _copy.meta.fields[
                        field_name
                    ].through.copy_edgy_model()
                    # we want to set the registry explicit
                    through_model.meta.registry = False
                    if src_field.from_foreign_key in through_model.meta.fields:
                        # explicit set
                        through_model.meta.fields[src_field.from_foreign_key].target = _copy
                        through_model.meta.fields[
                            src_field.from_foreign_key
                        ].related_name = cast(
                            BaseManyToManyForeignKeyField,
                            cast(type[BaseModelType], src_field.through).meta.fields[
                                src_field.from_foreign_key
                            ],
                        ).related_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=replaceable_models,
            on_conflict=on_conflict,
            database=(
                "keep"
                if cls.meta.registry is False or cls.database is not cls.meta.registry.database
                else True
            ),
        )
    return cast("type[Self]", _copy)

_update async

_update(is_partial, kwargs, pre_fn, post_fn, instance)

Internal method to perform an update operation on a model instance in the database.

This method handles the extraction of column values, execution of pre-save hooks, database interaction for updating records, and post-save hook execution.

PARAMETER	DESCRIPTION
is_partial	A boolean indicating if this is a partial update. TYPE: bool
kwargs	A dictionary of key-value pairs representing the fields to update. TYPE: dict[str, Any]
pre_fn	An asynchronous callable to be executed before the update. TYPE: Callable[..., Awaitable[Any]]
post_fn	An asynchronous callable to be executed after the update. TYPE: Callable[..., Awaitable[Any]]
instance	The model instance or queryset initiating the update. TYPE: BaseModelType | QuerySet

RETURNS	DESCRIPTION
int | None	The number of rows updated, or None if no update was performed.

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

async def _update(
    self: Model,
    is_partial: bool,
    kwargs: dict[str, Any],
    pre_fn: Callable[..., Awaitable[Any]],
    post_fn: Callable[..., Awaitable[Any]],
    instance: BaseModelType | QuerySet,
) -> int | None:
    """
    Internal method to perform an update operation on a model instance in the database.

    This method handles the extraction of column values, execution of pre-save hooks,
    database interaction for updating records, and post-save hook execution.

    Args:
        is_partial: A boolean indicating if this is a partial update.
        kwargs: A dictionary of key-value pairs representing the fields to update.
        pre_fn: An asynchronous callable to be executed before the update.
        post_fn: An asynchronous callable to be executed after the update.
        instance: The model instance or queryset initiating the update.

    Returns:
        The number of rows updated, or None if no update was performed.
    """
    real_class = self.get_real_class()
    column_values = self.extract_column_values(
        extracted_values=kwargs,
        is_partial=is_partial,
        is_update=True,
        phase="prepare_update",
        instance=self,
        model_instance=self,
        evaluate_values=True,
    )
    await pre_fn(
        real_class,
        model_instance=self,
        instance=instance,
        values=kwargs,
        column_values=column_values,
    )
    # empty updates shouldn't cause an error. E.g. only model references are updated
    clauses = self.identifying_clauses()
    row_count: int | None = None
    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, is_update=True)
            )
            expression = self.table.update().values(**column_values).where(*clauses)
            row_count = cast(int, 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()), is_update=True)

    if column_values or kwargs:
        # Ensure on access refresh the results is active
        self._db_deleted = False if row_count is None else row_count == 0
        self._db_loaded = False
    await post_fn(
        real_class,
        model_instance=self,
        instance=instance,
        values=kwargs,
        column_values=column_values,
    )
    return row_count

check_exist_in_db async

check_exist_in_db(only_needed=False)

Checks if the current model instance exists in the database.

PARAMETER	DESCRIPTION
only_needed	If True, performs the check only if the instance's loaded or deleted status is not conclusive. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
bool	True if the instance exists in the database, False otherwise.

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

async def check_exist_in_db(self, only_needed: bool = False) -> bool:
    """
    Checks if the current model instance exists in the database.

    Args:
        only_needed: If True, performs the check only if the instance's loaded
                     or deleted status is not conclusive.

    Returns:
        True if the instance exists in the database, False otherwise.
    """
    if only_needed:
        if self._db_deleted:
            return False
        if self._db_loaded:
            return True
    clauses = self.identifying_clauses()
    if not clauses:
        return False

    # Build the select expression.
    expression = self.table.select().where(*clauses).exists().select()

    # Perform the fetch.
    check_db_connection(self.database)
    async with self.database as database:
        result = cast(bool, await database.fetch_val(expression))
        self._db_deleted = not result
        return result

_insert async

_insert(evaluate_values, kwargs, pre_fn, post_fn, instance)

Internal method to perform an insert operation for a model instance into the database.

This method handles the extraction of column values, execution of pre-save hooks, database insertion, and post-save hook execution.

PARAMETER	DESCRIPTION
evaluate_values	A boolean indicating whether values should be evaluated before insertion (e.g., for default values). TYPE: bool
kwargs	A dictionary of key-value pairs representing the fields to insert. TYPE: dict[str, Any]
pre_fn	An asynchronous callable to be executed before the insert. TYPE: Callable[..., Awaitable[Any]]
post_fn	An asynchronous callable to be executed after the insert. TYPE: Callable[..., Awaitable[Any]]
instance	The model instance or queryset initiating the insert. TYPE: BaseModelType | QuerySet

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

async def _insert(
    self: Model,
    evaluate_values: bool,
    kwargs: dict[str, Any],
    pre_fn: Callable[..., Awaitable[Any]],
    post_fn: Callable[..., Awaitable[Any]],
    instance: BaseModelType | QuerySet,
) -> None:
    """
    Internal method to perform an insert operation for a model instance into the database.

    This method handles the extraction of column values, execution of pre-save hooks,
    database insertion, and post-save hook execution.

    Args:
        evaluate_values: A boolean indicating whether values should be evaluated
                         before insertion (e.g., for default values).
        kwargs: A dictionary of key-value pairs representing the fields to insert.
        pre_fn: An asynchronous callable to be executed before the insert.
        post_fn: An asynchronous callable to be executed after the insert.
        instance: The model instance or queryset initiating the insert.
    """
    real_class = self.get_real_class()
    column_values: dict[str, Any] = self.extract_column_values(
        extracted_values=kwargs,
        is_partial=False,
        is_update=False,
        phase="prepare_insert",
        instance=instance,
        model_instance=self,
        evaluate_values=evaluate_values,
    )
    await pre_fn(
        real_class,
        model_instance=self,
        instance=instance,
        column_values=column_values,
        values=kwargs,
    )
    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, is_update=False)
        )
        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()), is_update=False)
    # Ensure on access refresh the results is active
    self._db_loaded = False
    self._db_deleted = False
    await post_fn(
        real_class,
        model_instance=self,
        instance=instance,
        column_values=column_values,
        values=kwargs,
    )

add_global_field_constraints classmethod

add_global_field_constraints(schema=None, metadata=None)

Adds global constraints to an existing SQLAlchemy table.

This method is particularly useful for applying schema-specific or tenant-specific constraints to a table that has already been built.

PARAMETER	DESCRIPTION
schema	An optional schema name associated with the table. TYPE: str | None DEFAULT: None
metadata	An optional sqlalchemy.MetaData object. If None, the registry's metadata is used. TYPE: MetaData | None DEFAULT: None

RETURNS	DESCRIPTION
Table	The SQLAlchemy Table object with the added constraints.

RAISES	DESCRIPTION
AssertionError	If the model's registry is not set.

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

@classmethod
def add_global_field_constraints(
    cls,
    schema: str | None = None,
    metadata: sqlalchemy.MetaData | None = None,
) -> sqlalchemy.Table:
    """
    Adds global constraints to an existing SQLAlchemy table.

    This method is particularly useful for applying schema-specific or
    tenant-specific constraints to a table that has already been built.

    Args:
        schema: An optional schema name associated with the table.
        metadata: An optional `sqlalchemy.MetaData` object. If None,
                  the registry's metadata is used.

    Returns:
        The SQLAlchemy `Table` object with the added constraints.

    Raises:
        AssertionError: If the model's registry is not set.
    """
    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)
    table = metadata.tables[tablename if not schema else f"{schema}.{tablename}"]
    for name, field in cls.meta.fields.items():
        current_columns: list[sqlalchemy.Column] = []
        for column_name in cls.meta.field_to_column_names[name]:
            current_columns.append(table.columns[column_name])
        for constraint in field.get_global_constraints(name, current_columns, schemes):
            table.append_constraint(constraint)
    return table

_get_unique_constraints classmethod

_get_unique_constraints(fields)

Constructs and returns a SQLAlchemy UniqueConstraint object.

This method handles different input types for defining unique constraints, including a single field name, a collection of field names, or a UniqueConstraint object. It also generates a unique name for the constraint if not explicitly provided.

PARAMETER	DESCRIPTION
fields	The fields (or a UniqueConstraint object) for which to create the unique constraint. TYPE: Collection[str] | str | UniqueConstraint

RETURNS	DESCRIPTION
UniqueConstraint | None	A SQLAlchemy UniqueConstraint object, or None if no fields are provided.

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

@classmethod
def _get_unique_constraints(
    cls, fields: Collection[str] | str | UniqueConstraint
) -> sqlalchemy.UniqueConstraint | None:
    """
    Constructs and returns a SQLAlchemy `UniqueConstraint` object.

    This method handles different input types for defining unique constraints,
    including a single field name, a collection of field names, or a
    `UniqueConstraint` object. It also generates a unique name for the constraint
    if not explicitly provided.

    Args:
        fields: The fields (or a `UniqueConstraint` object) for which to create
                the unique constraint.

    Returns:
        A SQLAlchemy `UniqueConstraint` object, or None if no fields are provided.
    """
    if isinstance(fields, str):
        return sqlalchemy.UniqueConstraint(
            *cls.meta.field_to_column_names[fields],
            name=hash_names([fields], inner_prefix=cls.__name__, outer_prefix="uc"),
        )
    elif isinstance(fields, UniqueConstraint):
        return sqlalchemy.UniqueConstraint(
            *chain.from_iterable(
                # deduplicate and extract columns
                cls.meta.field_to_column_names[field]
                for field in set(fields.fields)
            ),
            name=fields.name,
            deferrable=fields.deferrable,
            initially=fields.initially,
        )
    # deduplicate
    fields = set(fields)
    return sqlalchemy.UniqueConstraint(
        *chain.from_iterable(cls.meta.field_to_column_names[field] for field in fields),
        name=hash_names(fields, inner_prefix=cls.__name__, outer_prefix="uc"),
    )

_get_indexes classmethod

_get_indexes(index)

Constructs and returns a SQLAlchemy Index object based on an Index definition.

PARAMETER	DESCRIPTION
index	The Index object containing the fields and name for the index. TYPE: Index

RETURNS	DESCRIPTION
Index | None	A SQLAlchemy Index object.

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

@classmethod
def _get_indexes(cls, index: Index) -> sqlalchemy.Index | None:
    """
    Constructs and returns a SQLAlchemy `Index` object based on an `Index` definition.

    Args:
        index: The `Index` object containing the fields and name for the index.

    Returns:
        A SQLAlchemy `Index` object.
    """
    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
        ),
    )

not_set_transaction

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

Returns a database transaction for the assigned database.

This method is designed to be assigned as the transaction property for model instances, allowing them to initiate database transactions.

PARAMETER	DESCRIPTION
force_rollback	If True, forces the transaction to roll back. TYPE: bool DEFAULT: False
**kwargs	Additional keyword arguments to pass to the database's transaction method. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
Transaction	A Transaction object.

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

def not_set_transaction(self, *, force_rollback: bool = False, **kwargs: Any) -> Transaction:
    """
    Returns a database transaction for the assigned database.

    This method is designed to be assigned as the `transaction` property for
    model instances, allowing them to initiate database transactions.

    Args:
        force_rollback: If True, forces the transaction to roll back.
        **kwargs: Additional keyword arguments to pass to the database's
                  transaction method.

    Returns:
        A `Transaction` object.
    """
    return cast(
        "Transaction",
        self.database.transaction(force_rollback=force_rollback, **kwargs),
    )

declarative classmethod

declarative()

Returns the SQLAlchemy Declarative model representation of the Edgy model.

This is a convenience class method that acts as an entry point to obtain the declarative version of the current Edgy model. It delegates the actual transformation process to generate_model_declarative.

PARAMETER	DESCRIPTION
cls	The Edgy model class (e.g., User, Product) for which the declarative model is to be generated. TYPE: type[EdgyBaseModel]

RETURNS	DESCRIPTION
Any	An Any type representing the dynamically generated SQLAlchemy Declarative
Any	model class, which can then be used with SQLAlchemy ORM functionalities.

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

@classmethod
def declarative(cls: type[EdgyBaseModel]) -> Any:
    """
    Returns the SQLAlchemy Declarative model representation of the Edgy model.

    This is a convenience class method that acts as an entry point to obtain
    the declarative version of the current Edgy model. It delegates the actual
    transformation process to `generate_model_declarative`.

    Args:
        cls: The Edgy model class (e.g., `User`, `Product`) for which the
             declarative model is to be generated.

    Returns:
        An `Any` type representing the dynamically generated SQLAlchemy Declarative
        model class, which can then be used with SQLAlchemy ORM functionalities.
    """
    # Calls the internal method to perform the transformation.
    return cls.generate_model_declarative()

generate_model_declarative classmethod

generate_model_declarative()

Transforms an Edgy model's SQLAlchemy Core Table into an SQLAlchemy Declarative model.

This method dynamically creates a new class that inherits from SQLAlchemy's Declarative Base. It assigns the Edgy model's SQLAlchemy Core Table to the __table__ attribute of the new class, making it a declarative model. Additionally, it identifies and configures SQLAlchemy ORM relationships for any foreign key fields defined in the Edgy model.

PARAMETER	DESCRIPTION
cls	The Edgy model class (e.g., User, Product) to be converted into a SQLAlchemy Declarative model. TYPE: type[EdgyBaseModel]

RETURNS	DESCRIPTION
Any	An Any type representing the dynamically generated SQLAlchemy Declarative
Any	model class, complete with mapped relationships.

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

@classmethod
def generate_model_declarative(cls: type[EdgyBaseModel]) -> Any:
    """
    Transforms an Edgy model's SQLAlchemy Core Table into an SQLAlchemy Declarative model.

    This method dynamically creates a new class that inherits from SQLAlchemy's
    Declarative Base. It assigns the Edgy model's SQLAlchemy Core Table to the
    `__table__` attribute of the new class, making it a declarative model.
    Additionally, it identifies and configures SQLAlchemy ORM relationships
    for any foreign key fields defined in the Edgy model.

    Args:
        cls: The Edgy model class (e.g., `User`, `Product`) to be converted
             into a SQLAlchemy Declarative model.

    Returns:
        An `Any` type representing the dynamically generated SQLAlchemy Declarative
        model class, complete with mapped relationships.
    """
    # orm is heavy, so keep it in the function
    from sqlalchemy.orm import Mapped, relationship

    # Retrieve the declarative base from the Edgy model's metadata registry.
    # This `Base` is the foundation for the new declarative model.
    Base = cls.meta.registry.declarative_base  # type: ignore

    # Define a dictionary to hold the attributes for the new declarative model class.
    # The `__table__` attribute is crucial for linking the declarative model
    # to the existing SQLAlchemy Core Table defined by Edgy.
    fields: dict[str, Any] = {"__table__": cls.table}

    # Dynamically create the new declarative model class using `type()`.
    # The class name is derived from the original Edgy model's name.
    model_table = type(cls.__name__, (Base,), fields)

    # Iterate through all columns in the Edgy model's SQLAlchemy Core Table.
    # This loop identifies foreign key columns that need corresponding ORM relationships.
    for column in cls.table.columns:
        # Skip columns that do not have foreign key constraints.
        if not column.foreign_keys:
            continue

        # Retrieve the field definition from the Edgy model's metadata using the column name.
        field = cls.meta.fields.get(column.name)

        # Determine the target model's name for the relationship.
        # If `field.to` is a class, use its name; otherwise, use `field.to` directly.
        to: str = field.to.__name__ if inspect.isclass(field.to) else field.to

        # Create an SQLAlchemy ORM relationship.
        # `Mapped[to]` provides type hinting for the relationship.
        # `relationship(to)` establishes the ORM link to the target model.
        mapped_model: Mapped[Any] = relationship(to)
        # The `type: ignore` is used because `Mapped[to]` expects a class,
        # but `to` is a string here. SQLAlchemy handles this internally.

        # Add the newly created relationship as a property to the declarative model's mapper.
        # The property name is constructed by appending `_relation` to the column name,
        # ensuring a distinct name for the ORM relationship attribute.
        model_table.__mapper__.add_property(f"{column.name}_relation", mapped_model)

    # Return the fully constructed SQLAlchemy Declarative model class.
    return model_table

can_load_from_row classmethod

can_load_from_row(row, table)

Checks if a model class can be instantiated and populated from a given SQLAlchemy row and table.

This method verifies if the model's registry exists, if it's not an abstract model, and if all primary key columns for the model are present and not None in the provided row's mapping.

PARAMETER	DESCRIPTION
row	The SQLAlchemy row object containing the data. TYPE: Row
table	The SQLAlchemy table object associated with the row. TYPE: Table

RETURNS	DESCRIPTION
bool	True if the model can be loaded from the row, False otherwise. TYPE: bool

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

@classmethod
def can_load_from_row(cls: type[Model], row: Row, table: Table) -> bool:
    """
    Checks if a model class can be instantiated and populated from a given SQLAlchemy row
    and table.

    This method verifies if the model's registry exists, if it's not an abstract model,
    and if all primary key columns for the model are present and not None in the
    provided row's mapping.

    Args:
        row (Row): The SQLAlchemy row object containing the data.
        table (Table): The SQLAlchemy table object associated with the row.

    Returns:
        bool: True if the model can be loaded from the row, False otherwise.
    """
    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, reference_select=None)

Converts a SQLAlchemy Row object into an Edgy Model instance.

This is a class method that processes a SQLAlchemy row, populating the model's fields, including handling select_related and prefetch_related relationships. It intelligently constructs the model instance by iterating through selected fields, managing prefixes for joined tables, and applying deferred or secret field exclusions.

PARAMETER	DESCRIPTION
row	The SQLAlchemy row result to convert. TYPE: Row
tables_and_models	A dictionary mapping prefixes to tuples of SQLAlchemy Table objects and Edgy Model types, representing the tables and models involved in the query. TYPE: dict[str, tuple[Table, type[BaseModelType]]]
select_related	An optional sequence of relationship names to eager-load. These relationships will be joined in the main query. TYPE: Sequence[Any] | None DEFAULT: None
prefetch_related	An optional sequence of Prefetch objects for pre-fetching related data in separate queries. TYPE: Sequence[Prefetch] | None DEFAULT: None
only_fields	An optional sequence of field names to include in the model instance. If specified, only these fields will be populated. TYPE: Sequence[str] | None DEFAULT: None
is_defer_fields	A boolean indicating whether fields are deferred. If True, the model instance will be a proxy model with deferred field loading. TYPE: bool DEFAULT: False
exclude_secrets	A boolean indicating whether secret fields should be excluded from the populated model instance. TYPE: bool DEFAULT: False
using_schema	An optional schema name to use for the model. TYPE: str | None DEFAULT: None
database	An optional database instance to associate with the model. TYPE: Database | None DEFAULT: None
prefix	An optional prefix used for columns in the row mapping, typically for joined tables in select_related. TYPE: str DEFAULT: ''
old_select_related_value	An optional existing model instance to update with the new row data, used in recursive select_related calls. TYPE: Model | None DEFAULT: None
reference_select	An optional dictionary specifying how to map specific columns from the row to model fields, especially for aliased columns or complex selects. TYPE: reference_select_type | None DEFAULT: None

RETURNS	DESCRIPTION
Model | None	Model | None: A fully populated Edgy Model instance, or None if the model
Model | None	cannot be loaded from the row due to missing primary key values in joined
Model | None	relationships.

RAISES	DESCRIPTION
QuerySetError	If a field specified in select_related does not exist on the model or is not a RelationshipField.
NotImplementedError	If prefetching from other databases is attempted, as this feature is not yet supported.

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

@classmethod
async def from_sqla_row(
    cls: type[Model],
    row: Row,
    tables_and_models: dict[str, tuple[Table, type[BaseModelType]]],
    select_related: Sequence[Any] | None = None,
    prefetch_related: Sequence[Prefetch] | None = None,
    only_fields: Sequence[str] | None = None,
    is_defer_fields: bool = False,
    exclude_secrets: bool = False,
    using_schema: str | None = None,
    database: Database | None = None,
    prefix: str = "",
    old_select_related_value: Model | None = None,
    reference_select: reference_select_type | None = None,
) -> Model | None:
    """
    Converts a SQLAlchemy `Row` object into an Edgy `Model` instance.

    This is a class method that processes a SQLAlchemy row, populating the model's
    fields, including handling `select_related` and `prefetch_related` relationships.
    It intelligently constructs the model instance by iterating through selected
    fields, managing prefixes for joined tables, and applying deferred or secret
    field exclusions.

    Args:
        row (Row): The SQLAlchemy row result to convert.
        tables_and_models (dict[str, tuple[Table, type[BaseModelType]]]): A dictionary
            mapping prefixes to tuples of SQLAlchemy Table objects and Edgy Model types,
            representing the tables and models involved in the query.
        select_related (Sequence[Any] | None): An optional sequence of relationship
            names to eager-load. These relationships will be joined in the main query.
        prefetch_related (Sequence[Prefetch] | None): An optional sequence of `Prefetch`
            objects for pre-fetching related data in separate queries.
        only_fields (Sequence[str] | None): An optional sequence of field names to
            include in the model instance. If specified, only these fields will be
            populated.
        is_defer_fields (bool): A boolean indicating whether fields are deferred. If
            True, the model instance will be a proxy model with deferred field loading.
        exclude_secrets (bool): A boolean indicating whether secret fields should be
            excluded from the populated model instance.
        using_schema (str | None): An optional schema name to use for the model.
        database (Database | None): An optional database instance to associate with
            the model.
        prefix (str): An optional prefix used for columns in the row mapping,
            typically for joined tables in `select_related`.
        old_select_related_value (Model | None): An optional existing model instance
            to update with the new row data, used in recursive `select_related` calls.
        reference_select (reference_select_type | None): An optional dictionary
            specifying how to map specific columns from the row to model fields,
            especially for aliased columns or complex selects.

    Returns:
        Model | None: A fully populated Edgy Model instance, or None if the model
        cannot be loaded from the row due to missing primary key values in joined
        relationships.

    Raises:
        QuerySetError: If a field specified in `select_related` does not exist on
            the model or is not a `RelationshipField`.
        NotImplementedError: If prefetching from other databases is attempted, as
            this feature is not yet supported.
    """
    # Initialize reference_select if not provided.
    _reference_select: reference_select_type = (
        reference_select if reference_select is not None else {}
    )
    item: dict[str, Any] = {}  # Dictionary to store the model's attributes.
    select_related = select_related or []
    prefetch_related = prefetch_related or []
    secret_columns: set[str] = set()

    # If exclude_secrets is True, gather all column names corresponding to secret fields.
    if exclude_secrets:
        for name in cls.meta.secret_fields:
            secret_columns.update(cls.meta.field_to_column_names[name])

    # Process select_related relationships.
    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):
            # Traverse the field to get the related model class and any remaining path.
            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}"

        # If the related model cannot be loaded from the current row (e.g., all FKs
        # are None, indicating no join match), skip processing this relationship.
        if not model_class.can_load_from_row(
            row,
            tables_and_models[_prefix][0],
        ):
            continue

        # Get the nested reference_select for the current related field.
        reference_select_sub = _reference_select.get(field_name)
        if not isinstance(reference_select_sub, dict):
            reference_select_sub = {}

        if remainder:
            # Recursively call from_sqla_row for nested select_related.
            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),
                reference_select=reference_select_sub,
            )
        else:
            # Call from_sqla_row for the direct related model.
            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),
                reference_select=reference_select_sub,
            )

    # If an `old_select_related_value` (an existing model instance) is provided,
    # update its attributes with the newly populated `item` and return it.
    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 foreign key related names (lazy-loaded relationships).
    for related in cls.meta.foreign_key_fields:
        foreign_key = cls.meta.fields[related]

        # Determine if this related field should be ignored (e.g., if it's already
        # handled by select_related or is a secret field).
        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:  # Skip if already populated by select_related.
            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 = {}

        # Collect foreign key column values from the row mapping.
        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]
                )

        # Process nested reference selects for the child model.
        reference_select_child = _reference_select.get(related)
        extra_no_trigger_child: set[str] = set()
        if isinstance(reference_select_child, dict):
            for (
                reference_target_child,
                reference_source_child,
            ) in cast("reference_select_type", reference_select_child).items():
                if isinstance(reference_source_child, dict) or not reference_source_child:
                    continue
                extra_no_trigger_child.add(reference_target_child)
                if isinstance(reference_source_child, str):
                    reference_source_child_parts = reference_source_child.rsplit("__", 1)
                    if (
                        len(reference_source_child_parts) == 2
                        and reference_source_child_parts[0] in tables_and_models
                    ):
                        reference_source_child = (
                            f"{tables_and_models[reference_source_child_parts[0]][0].name}_"
                            f"{reference_source_child_parts[1]}"
                        )
                child_item[reference_target_child] = row._mapping[reference_source_child]

        # Create a proxy model for the related field, representing a lazy-loaded
        # instance containing only the foreign key(s).
        proxy_model = model_related.proxy_model(**child_item)
        proxy_database = database if model_related.database is cls.database else None

        # Apply instance extras (schema, database, etc.) to the proxy model.
        proxy_model = apply_instance_extras(
            proxy_model,
            model_related,
            using_schema,
            database=proxy_database,
        )
        proxy_model.identifying_db_fields = foreign_key.related_columns
        proxy_model.__no_load_trigger_attrs__.update(extra_no_trigger_child)
        if exclude_secrets:
            proxy_model.__no_load_trigger_attrs__.update(model_related.meta.secret_fields)

        item[related] = proxy_model

    # Populate the regular column values for the main model.
    class_columns = cls.table.columns
    for column in table_columns:
        # Skip if only_fields is specified and the column is not in it.
        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:  # Skip if the column is a secret.
            continue
        if column.key not in class_columns:  # Skip if the column is not part of the model.
            continue
        if column.key in item:  # Skip if already populated (e.g., by select_related).
            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]

    # Apply any explicit column mappings from `reference_select`.
    for reference_target_main, reference_source_main in _reference_select.items():
        if isinstance(reference_source_main, dict) or not reference_source_main:
            continue

        if isinstance(reference_source_main, str):
            reference_source_main_parts = reference_source_main.rsplit("__", 1)
            if (
                len(reference_source_main_parts) == 2
                and reference_source_main_parts[0] in tables_and_models
            ):
                reference_source_main = (
                    f"{tables_and_models[reference_source_main_parts[0]][0].name}_"
                    f"{reference_source_main_parts[1]}"
                )
        # Overwrite existing item with the value from reference_select.
        item[reference_target_main] = row._mapping[reference_source_main]

    # Instantiate the model (either as a proxy or a full model).
    model: Model = (
        cls.proxy_model(**item, __phase__="init_db")
        if exclude_secrets or is_defer_fields or only_fields
        else cls(**item, __phase__="init_db")
    )

    # Mark the model as fully loaded if no deferred or only_fields are active.
    if not is_defer_fields and not only_fields:
        model._db_deleted = False
        model._db_loaded = True

    # If excluding secrets, ensure these attributes do not trigger a load.
    if exclude_secrets:
        model.__no_load_trigger_attrs__.update(cls.meta.secret_fields)

    # Apply instance extras (schema, database, table, etc.) to the main model.
    model = apply_instance_extras(
        model,
        cls,
        using_schema,
        database=database,
        table=tables_and_models[prefix][0],
    )

    # Handle prefetch_related fields if specified.
    if prefetch_related:
        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  # Ensure the primary key is not None.
    return model

__should_ignore_related_name classmethod

__should_ignore_related_name(related_name, select_related)

Determines whether a foreign key related name should be ignored during model population, typically if it's already covered by a select_related statement.

PARAMETER	DESCRIPTION
related_name	The name of the foreign key relationship. TYPE: str
select_related	A sequence of strings representing the select_related relationships. TYPE: Sequence[str]

RETURNS	DESCRIPTION
bool	True if the related name should be ignored, False otherwise. TYPE: bool

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:
    """
    Determines whether a foreign key related name should be ignored during model
    population, typically if it's already covered by a `select_related` statement.

    Args:
        related_name (str): The name of the foreign key relationship.
        select_related (Sequence[str]): A sequence of strings representing the
            `select_related` relationships.

    Returns:
        bool: True if the related name should be ignored, False otherwise.
    """
    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='')

Builds a unique cache key for a model instance based on its class name and primary key values extracted from a SQLAlchemy row.

PARAMETER	DESCRIPTION
row	The SQLAlchemy row object from which to extract primary key values. TYPE: Row
row_prefix	An optional prefix for column names in the row mapping, used when dealing with joined tables. TYPE: str DEFAULT: ''

RETURNS	DESCRIPTION
tuple	A tuple representing the unique key for the model instance. TYPE: tuple

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:
    """
    Builds a unique cache key for a model instance based on its class name and
    primary key values extracted from a SQLAlchemy row.

    Args:
        row (Row): The SQLAlchemy row object from which to extract primary key values.
        row_prefix (str): An optional prefix for column names in the row mapping,
            used when dealing with joined tables.

    Returns:
        tuple: A tuple representing the unique key for the model instance.
    """
    pk_key_list: list[Any] = [cls.__name__]
    for attr in cls.pkcolumns:
        # Append the primary key value from the row to the key list.
        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)

Sets a prefetched relationship on a model instance. This method handles the logic of retrieving and associating the prefetched data.

PARAMETER	DESCRIPTION
row	The SQLAlchemy row from which the main model was constructed. TYPE: Row
model	The Edgy Model instance to which the prefetched data will be attached. TYPE: Model
row_prefix	The prefix used for columns in the SQLAlchemy row, representing the main model's table. TYPE: str
related	The Prefetch object specifying the relationship to prefetch. TYPE: Prefetch

RAISES	DESCRIPTION
QuerySetError	If creating a reverse path is not possible (e.g., for unidirectional fields).
NotImplementedError	If prefetching from other databases is attempted.

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:
    """
    Sets a prefetched relationship on a model instance. This method handles the logic
    of retrieving and associating the prefetched data.

    Args:
        row (Row): The SQLAlchemy row from which the main model was constructed.
        model (Model): The Edgy Model instance to which the prefetched data will be
            attached.
        row_prefix (str): The prefix used for columns in the SQLAlchemy row,
            representing the main model's table.
        related (Prefetch): The Prefetch object specifying the relationship to prefetch.

    Raises:
        QuerySetError: If creating a reverse path is not possible (e.g., for
            unidirectional fields).
        NotImplementedError: If prefetching from other databases is attempted.
    """
    model_key = ()
    if related._is_finished:
        # If the prefetch operation is marked as finished (meaning all rows for this
        # prefetch have been collected), then bake the results. This allows for
        # efficient retrieval of prefetched data.
        await related.init_bake(type(model))
        model_key = model.create_model_key()

    # If the model's key exists in the baked results, retrieve and set the prefetched
    # data directly.
    if model_key in related._baked_results:
        setattr(model, related.to_attr, related._baked_results[model_key])
    else:
        # If not in baked results, or not finished, proceed with fetching.
        # Crawl the relationship path to get details about the related model and
        # reverse path.
        crawl_result = crawl_relationship(
            model.__class__, related.related_name, traverse_last=True
        )
        if crawl_result.reverse_path is False:
            raise 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 for potential conflicts with existing attributes on the model.
            check_prefetch_collision(model, related)
            if queryset is None:
                # If no specific queryset is provided for prefetch, default to all.
                queryset = crawl_result.model_class.query.all()

            # Ensure the reverse path is selected to link back to the main model.
            queryset = queryset.all()
            queryset._select_related.add(crawl_result.reverse_path)
            queryset._cached_select_related_expression = None

        # Construct the filter clause for the prefetched query using the main model's
        # primary key(s).
        clause = {
            f"{crawl_result.reverse_path}__{pkcol}": row._mapping[f"{row_prefix}{pkcol}"]
            for pkcol in cls.pkcolumns
        }
        # Execute the prefetched query and set the result on the model instance.
        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)

Manages the execution of all prefetch_related queries for a given model instance. This method iterates through the specified prefetch relationships, checks for collisions, and initiates the asynchronous loading of related data.

PARAMETER	DESCRIPTION
row	The SQLAlchemy row from which the main model was constructed. TYPE: Row
model	The Edgy Model instance for which prefetch relationships are to be handled. TYPE: Model
prefix	The prefix used for columns in the SQLAlchemy row, representing the main model's table. TYPE: str
tables_and_models	A dictionary mapping prefixes to tuples of SQLAlchemy Table objects and Edgy Model types, representing the tables and models involved in the query. TYPE: dict[str, tuple[Table, type[BaseModelType]]]
prefetch_related	A sequence of Prefetch objects to process. TYPE: Sequence[Prefetch]

RAISES	DESCRIPTION
QuerySetError	If a conflicting attribute is found that would be overwritten by a prefetch operation.

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:
    """
    Manages the execution of all `prefetch_related` queries for a given model instance.
    This method iterates through the specified prefetch relationships, checks for
    collisions, and initiates the asynchronous loading of related data.

    Args:
        row (Row): The SQLAlchemy row from which the main model was constructed.
        model (Model): The Edgy Model instance for which prefetch relationships are
            to be handled.
        prefix (str): The prefix used for columns in the SQLAlchemy row,
            representing the main model's table.
        tables_and_models (dict[str, tuple[Table, type[BaseModelType]]]): A dictionary
            mapping prefixes to tuples of SQLAlchemy Table objects and Edgy Model types,
            representing the tables and models involved in the query.
        prefetch_related (Sequence[Prefetch]): A sequence of `Prefetch` objects to
            process.

    Raises:
        QuerySetError: If a conflicting attribute is found that would be
            overwritten by a prefetch operation.
    """
    queries = []

    for related in prefetch_related:
        # Check for conflicting names early to prevent unexpected overwrites.
        check_prefetch_collision(model=model, related=related)
        row_prefix = f"{tables_and_models[prefix][0].name}_" if prefix else ""
        queries.append(
            cls.__set_prefetch(row=row, row_prefix=row_prefix, model=model, related=related)
        )
    # Execute all prefetch queries concurrently if there are any.
    if queries:
        await asyncio.gather(*queries)

real_add_to_registry classmethod

real_add_to_registry(**kwargs)

Adds the model to its associated registry, handling admin model registration.

This method is called during the model's initialization process to register it with the Registry instance specified in its Meta class. It also conditionally adds the model to the admin's list of registered models if in_admin is not False and a registry is present.

PARAMETER	DESCRIPTION
**kwargs	Arbitrary keyword arguments passed to the superclass method. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
type[Model]	type[Model]: The model class after being added to the registry.

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

@classmethod
def real_add_to_registry(cls, **kwargs: Any) -> type[Model]:
    """
    Adds the model to its associated registry, handling admin model registration.

    This method is called during the model's initialization process to register
    it with the `Registry` instance specified in its `Meta` class. It also
    conditionally adds the model to the admin's list of registered models
    if `in_admin` is not `False` and a registry is present.

    Args:
        **kwargs (Any): Arbitrary keyword arguments passed to the superclass method.

    Returns:
        type[Model]: The model class after being added to the registry.
    """
    # Store the current `in_admin` status from the model's meta information.
    in_admin = cls.meta.in_admin
    # Call the superclass method to perform the actual addition to the registry.
    result = cast(type[Model], super().real_add_to_registry(**kwargs))

    # If `in_admin` is not explicitly `False` and a registry is available,
    # add the model's name to the registry's admin_models set.
    if in_admin is not False and result.meta.registry:
        result.meta.registry.admin_models.add(result.__name__)
    return result

generate_proxy_model classmethod

generate_proxy_model()

Generates a lightweight proxy model based on the current model.

A proxy model is a shallow copy of the original model. It is not added to the registry and typically has its field references replaced to be generic (e.g., edgy.fields.BigIntegerField becomes int). This is useful for internal operations or when a stripped-down version of the model is needed without affecting the main registry.

RETURNS	DESCRIPTION
type[Model]	type[Model]: A new model class representing the proxy model.

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

@classmethod
def generate_proxy_model(cls) -> type[Model]:
    """
    Generates a lightweight proxy model based on the current model.

    A proxy model is a shallow copy of the original model. It is not added to
    the registry and typically has its field references replaced to be generic
    (e.g., `edgy.fields.BigIntegerField` becomes `int`). This is useful for
    internal operations or when a stripped-down version of the model is needed
    without affecting the main registry.

    Returns:
        type[Model]: A new model class representing the proxy model.
    """
    # Initialize a dictionary to hold attributes for the new proxy model.
    # It excludes certain internal pydantic-related keys.
    attrs: dict[str, Any] = {
        key: val for key, val in cls.__dict__.items() if key not in cls._removed_copy_keys
    }

    # Managers and fields are specifically re-added as they are essential
    # for the proxy model's functionality. The 'no_copy' flags are not
    # honored here as it is a specific type of copy for internal use.
    attrs.update(cls.meta.fields)
    attrs.update(cls.meta.managers)
    # Mark this model as a proxy model for internal identification.
    attrs["__is_proxy_model__"] = True

    # Create the new Edgy model using the collected attributes and metadata.
    # `skip_registry` is set to True to ensure this proxy model is not
    # inadvertently added to the global model registry.
    _copy = create_edgy_model(
        __name__=cls.__name__,
        __module__=cls.__module__,
        __definitions__=attrs,
        __metadata__=cls.meta,
        __bases__=cls.__bases__,
        # The registry is still copied from meta but `skip_registry` prevents
        # it from being added to the global registry by the metaclass.
        __type_kwargs__={"skip_registry": True},
    )
    # Assign the database connection from the original model to the proxy model.
    _copy.database = cls.database
    # Convert specific Edgy field types in the proxy model to their generic Python types.
    generify_model_fields(_copy)
    return _copy