ReflectModel class

saffier.ReflectModel

ReflectModel(**kwargs)

Bases: Model, SaffierBaseReflectModel

Reflect on async engines is not yet supported, therefore, we need to make a sync_engine call.

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

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

def __init__(self, **kwargs: Any) -> None:
    self.setup_model_fields_from_kwargs(kwargs)

is_proxy_model class-attribute

is_proxy_model = False

query class-attribute

query = Manager()

meta class-attribute

meta = MetaInfo(None)

__db_model__ class-attribute

__db_model__ = False

__raw_query__ class-attribute

__raw_query__ = None

__proxy_model__ class-attribute

__proxy_model__ = None

pk property writable

pk

raw_query property writable

raw_query

table property writable

table

proxy_model cached property

proxy_model

signals cached property

signals

Meta

The Meta class used to configure each metadata of the model. Abstract classes are not generated in the database, instead, they are simply used as a reference for field generation.

Usage:

.. code-block:: python3

class User(Model):
    ...

    class Meta:
        registry = models
        tablename = "users"

declarative classmethod

declarative()

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

@classmethod
def declarative(cls) -> typing.Any:
    return cls.generate_model_declarative()

generate_model_declarative classmethod

generate_model_declarative()

Transforms a core Saffier table into a Declarative model table.

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

@classmethod
def generate_model_declarative(cls) -> typing.Any:
    """
    Transforms a core Saffier table into a Declarative model table.
    """
    Base = cls.meta.registry.declarative_base

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

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

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

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

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

    return model_table

_update_auto_now_fields

_update_auto_now_fields(values, fields)

Updates the auto fields

PARAMETER	DESCRIPTION
values	TYPE: Any
fields	TYPE: Any

Source code in saffier/core/utils/models.py

def _update_auto_now_fields(self, values: Any, fields: Any) -> Any:
    """
    Updates the auto fields
    """
    for k, v in fields.items():
        if isinstance(v, (DateField, DateTimeField)) and v.auto_now:
            values[k] = v.validator.get_default_value()  # type: ignore
    return values

_resolve_value

_resolve_value(value)

PARAMETER	DESCRIPTION
value	TYPE: Any

Source code in saffier/core/utils/models.py

def _resolve_value(self, value: typing.Any) -> typing.Any:
    if isinstance(value, dict):
        return dumps(
            value,
            option=OPT_SERIALIZE_NUMPY | OPT_OMIT_MICROSECONDS,
        ).decode("utf-8")
    elif isinstance(value, Enum):
        return value.name
    return value

setup_model_fields_from_kwargs

setup_model_fields_from_kwargs(kwargs)

Loops and setup the kwargs of the model

PARAMETER	DESCRIPTION
kwargs	TYPE: Any

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

def setup_model_fields_from_kwargs(self, kwargs: Any) -> Any:
    """
    Loops and setup the kwargs of the model
    """
    if "pk" in kwargs:
        kwargs[self.pkname] = kwargs.pop("pk")

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

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

get_instance_name

get_instance_name()

Returns the name of the class in lowercase.

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

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

generate_proxy_model classmethod

generate_proxy_model()

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

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

@classmethod
def generate_proxy_model(cls) -> Type["Model"]:
    """
    Generates a proxy model for each model. This proxy model is a simple
    shallow copy of the original model being generated.
    """
    if cls.__proxy_model__:
        return cls.__proxy_model__

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

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

build classmethod

build(schema=None)

The inspect is done in an async manner and reflects the objects from the database.

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

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

@classmethod
def build(cls, schema: Optional[str] = None) -> sqlalchemy.Table:
    """
    The inspect is done in an async manner and reflects the objects from the database.
    """
    metadata = cast("sqlalchemy.MetaData", cls.meta.registry._metadata)  # type: ignore
    metadata.schema = schema
    tablename: str = cast("str", cls.meta.tablename)
    return cls.reflect(tablename, metadata)

_get_unique_constraints classmethod

_get_unique_constraints(columns)

Returns the unique constraints for the model.

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

PARAMETER	DESCRIPTION
columns	TYPE: Sequence

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

@classmethod
def _get_unique_constraints(cls, columns: Sequence) -> Optional[sqlalchemy.UniqueConstraint]:
    """
    Returns the unique constraints for the model.

    The columns must be a a list, tuple of strings or a UniqueConstraint object.
    """
    if isinstance(columns, str):
        return sqlalchemy.UniqueConstraint(columns)
    elif isinstance(columns, UniqueConstraint):
        return sqlalchemy.UniqueConstraint(*columns.fields)
    return sqlalchemy.UniqueConstraint(*columns)

_get_indexes classmethod

_get_indexes(index)

Creates the index based on the Index fields

PARAMETER	DESCRIPTION
index	TYPE: Index

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

@classmethod
def _get_indexes(cls, index: Index) -> Optional[sqlalchemy.Index]:
    """
    Creates the index based on the Index fields
    """
    return sqlalchemy.Index(index.name, *index.fields)  # type: ignore

update_from_dict

update_from_dict(dict_values)

Updates the current model object with the new fields

PARAMETER	DESCRIPTION
dict_values	TYPE: Dict[str, Any]

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

def update_from_dict(self, dict_values: Dict[str, Any]) -> Self:
    """Updates the current model object with the new fields"""
    for key, value in dict_values.items():
        setattr(self, key, value)
    return self

extract_db_fields

extract_db_fields()

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

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

def extract_db_fields(self) -> Dict[str, Any]:
    """
    Extacts all the db fields and excludes the related_names since those
    are simply relations.
    """
    related_names = self.meta.related_names
    return {k: v for k, v in self.__dict__.items() if k not in related_names}

__setattr__

__setattr__(key, value)

PARAMETER	DESCRIPTION
key	TYPE: Any
value	TYPE: Any

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

def __setattr__(self, key: Any, value: Any) -> Any:
    if key in self.fields:
        # Setting a relationship to a raw pk value should set a
        # fully-fledged relationship instance, with just the pk loaded.
        field = self.fields[key]

        if isinstance(field, saffier.ManyToManyField):
            value = getattr(self, settings.many_to_many_relation.format(key=key))
        else:
            value = self.fields[key].expand_relationship(value)
    super().__setattr__(key, value)

__get_instance_values

__get_instance_values(instance)

PARAMETER	DESCRIPTION
instance	TYPE: Any

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

def __get_instance_values(self, instance: Any) -> Set[Any]:
    return {
        v
        for k, v in instance.__dict__.items()
        if k in instance.fields.keys() and v is not None
    }

__eq__

__eq__(other)

PARAMETER	DESCRIPTION
other	TYPE: Any

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

def __eq__(self, other: Any) -> bool:
    if self.__class__ != other.__class__:
        return False
    original = self.__get_instance_values(instance=self)
    other_values = self.__get_instance_values(instance=other)
    if original != other_values:
        return False
    return True

get_engine cached classmethod

get_engine(url)

PARAMETER	DESCRIPTION
url	TYPE: str

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

@classmethod
@functools.lru_cache
def get_engine(cls, url: str) -> Engine:
    return sqlalchemy.create_engine(url)

reflect classmethod

reflect(tablename, metadata)

PARAMETER	DESCRIPTION
tablename	TYPE: str
metadata	TYPE: MetaData

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

@classmethod
def reflect(cls, tablename: str, metadata: sqlalchemy.MetaData) -> sqlalchemy.Table:
    try:
        return sqlalchemy.Table(
            tablename, metadata, autoload_with=cls.meta.registry.sync_engine  # type: ignore
        )
    except Exception as e:
        raise ImproperlyConfigured(
            detail=f"Table with the name {tablename} does not exist."
        ) from e

from_query_result classmethod

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

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

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

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

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

:return: Model class.

PARAMETER	DESCRIPTION
row	TYPE: Row
select_related	TYPE: Optional[Sequence[Any]] DEFAULT: None
prefetch_related	TYPE: Optional[Sequence[Prefetch]] DEFAULT: None
is_only_fields	TYPE: bool DEFAULT: False
only_fields	TYPE: Sequence[str] DEFAULT: None
is_defer_fields	TYPE: bool DEFAULT: False
exclude_secrets	TYPE: bool DEFAULT: False
using_schema	TYPE: Union[str, None] DEFAULT: None

Source code in saffier/core/db/models/row.py

@classmethod
def from_query_result(
    cls,
    row: Row,
    select_related: Optional[Sequence[Any]] = None,
    prefetch_related: Optional[Sequence["Prefetch"]] = None,
    is_only_fields: bool = False,
    only_fields: Sequence[str] = None,
    is_defer_fields: bool = False,
    exclude_secrets: bool = False,
    using_schema: Union[str, None] = None,
) -> Optional[Type["Model"]]:
    """
    Class method to convert a SQLAlchemy Row result into a SaffierModel row type.

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

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

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

    :return: Model class.
    """
    item: Dict[str, Any] = {}
    select_related = select_related or []
    prefetch_related = prefetch_related or []

    secret_fields = (
        [name for name, field in cls.fields.items() if field.secret] if exclude_secrets else []
    )

    # Instantiate any child instances first.
    for related in select_related:
        if "__" in related:
            first_part, remainder = related.split("__", 1)
            try:
                model_cls = cls.fields[first_part].target
            except KeyError:
                model_cls = getattr(cls, first_part).related_from

            item[first_part] = model_cls.from_query_result(
                row,
                select_related=[remainder],
                using_schema=using_schema,
                exclude_secrets=exclude_secrets,
            )
        else:
            try:
                model_cls = cls.fields[related].target
            except KeyError:
                model_cls = getattr(cls, related).related_from
            item[related] = model_cls.from_query_result(
                row, using_schema=using_schema, exclude_secrets=exclude_secrets
            )

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

        model_related = foreign_key.target

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

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

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

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

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

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

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

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

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

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

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

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

    if using_schema is not None:
        model.table = model.build(using_schema)  # type: ignore
    return model

__apply_schema classmethod

__apply_schema(model, schema=None)

PARAMETER	DESCRIPTION
model	TYPE: Type[Model]
schema	TYPE: Optional[str] DEFAULT: None

Source code in saffier/core/db/models/row.py

@classmethod
def __apply_schema(cls, model: Type["Model"], schema: Optional[str] = None) -> Type["Model"]:
    # Apply the schema to the model
    if schema is not None:
        model.table = model.build(schema)  # type: ignore
    return model

__should_ignore_related_name classmethod

__should_ignore_related_name(related_name, select_related)

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

PARAMETER	DESCRIPTION
related_name	TYPE: str
select_related	TYPE: Sequence[str]

Source code in saffier/core/db/models/row.py

@classmethod
def __should_ignore_related_name(
    cls, related_name: str, select_related: Sequence[str]
) -> bool:
    """
    Validates if it should populate the related field if select related is not considered.
    """
    for related_field in select_related:
        fields = related_field.split("__")
        if related_name in fields:
            return True
    return False

__handle_prefetch_related classmethod

__handle_prefetch_related(row, model, prefetch_related, parent_cls=None, original_prefetch=None, is_nested=False)

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

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

PARAMETER	DESCRIPTION
row	TYPE: Row
model	TYPE: Type[Model]
prefetch_related	TYPE: Sequence[Prefetch]
parent_cls	TYPE: Optional[Type[Model]] DEFAULT: None
original_prefetch	TYPE: Optional[Prefetch] DEFAULT: None
is_nested	TYPE: bool DEFAULT: False

Source code in saffier/core/db/models/row.py

@classmethod
def __handle_prefetch_related(
    cls,
    row: Row,
    model: Type["Model"],
    prefetch_related: Sequence["Prefetch"],
    parent_cls: Optional[Type["Model"]] = None,
    original_prefetch: Optional["Prefetch"] = None,
    is_nested: bool = False,
) -> Type["Model"]:
    """
    Handles any prefetch related scenario from the model.
    Loads in advance all the models needed for a specific record

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

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

        if original_prefetch and not is_nested:
            original_prefetch = related

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

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

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

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

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

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

            setattr(model, related.to_attr, records)
    return model

__process_nested_prefetch_related classmethod

__process_nested_prefetch_related(row, prefetch_related, parent_cls, original_prefetch, queryset)

Processes the nested prefetch related names.

PARAMETER	DESCRIPTION
row	TYPE: Row
prefetch_related	TYPE: Prefetch
parent_cls	TYPE: Type[Model]
original_prefetch	TYPE: Prefetch
queryset	TYPE: QuerySet

Source code in saffier/core/db/models/row.py

@classmethod
def __process_nested_prefetch_related(
    cls,
    row: Row,
    prefetch_related: "Prefetch",
    parent_cls: Type["Model"],
    original_prefetch: "Prefetch",
    queryset: "QuerySet",
) -> List[Type["Model"]]:
    """
    Processes the nested prefetch related names.
    """
    query_split = original_prefetch.related_name.split("__")
    query_split.reverse()
    query_split.pop(query_split.index(prefetch_related.related_name))

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

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

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

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

    # Extact foreign key value
    filter_by_pk = getattr(row, parent_cls.pkname)

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

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

__run_query async classmethod

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

Runs a specific query against a given model with filters.

PARAMETER	DESCRIPTION
model	TYPE: Optional[Type[Model]] DEFAULT: None
extra	TYPE: Optional[Dict[str, Any]] DEFAULT: None
queryset	TYPE: Optional[QuerySet] DEFAULT: None

Source code in saffier/core/db/models/row.py

@classmethod
async def __run_query(
    cls,
    model: Optional[Type["Model"]] = None,
    extra: Optional[Dict[str, Any]] = None,
    queryset: Optional["QuerySet"] = None,
) -> Union[List[Type["Model"]], Any]:
    """
    Runs a specific query against a given model with filters.
    """

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

    if extra:
        queryset.extra = extra

    return await queryset

model_dump

model_dump(include=None, exclude=None, exclude_none=False)

Dumps the model in a dict format.

PARAMETER	DESCRIPTION
include	TYPE: Union[Set[int], Set[str], Dict[int, Any], Dict[str, Any], None] DEFAULT: None
exclude	TYPE: Union[Set[int], Set[str], Dict[int, Any], Dict[str, Any], None] DEFAULT: None
exclude_none	TYPE: bool DEFAULT: False

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

def model_dump(
    self,
    include: Union[
        typing.Set[int],
        typing.Set[str],
        typing.Dict[int, typing.Any],
        typing.Dict[str, typing.Any],
        None,
    ] = None,
    exclude: Union[
        typing.Set[int],
        typing.Set[str],
        typing.Dict[int, typing.Any],
        typing.Dict[str, typing.Any],
        None,
    ] = None,
    exclude_none: bool = False,
) -> typing.Dict[str, typing.Any]:
    """
    Dumps the model in a dict format.
    """
    row_dict = {k: v for k, v in self.__dict__.items() if k in self.fields}

    if include is not None:
        row_dict = {k: v for k, v in row_dict.items() if k in include}
    if exclude is not None:
        row_dict = {k: v for k, v in row_dict.items() if k not in exclude}
    if exclude_none:
        row_dict = {k: v for k, v in row_dict.items() if v is not None}
    return row_dict

update async

update(**kwargs)

Update operation of the database fields.

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

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

async def update(self, **kwargs: typing.Any) -> typing.Any:
    """
    Update operation of the database fields.
    """
    await self.signals.pre_update.send(sender=self.__class__, instance=self)

    fields = {key: field.validator for key, field in self.fields.items() if key in kwargs}
    validator = Schema(fields=fields)
    kwargs = self._update_auto_now_fields(validator.check(kwargs), self.fields)
    pk_column = getattr(self.table.c, self.pkname)
    expression = self.table.update().values(**kwargs).where(pk_column == self.pk)
    await self.database.execute(expression)
    await self.signals.post_update.send(sender=self.__class__, instance=self)

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

    return self

delete async

delete()

Delete operation from the database

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

async def delete(self) -> None:
    """Delete operation from the database"""
    await self.signals.pre_delete.send(sender=self.__class__, instance=self)

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

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

load async

load()

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

async def load(self) -> None:
    # Build the select expression.
    pk_column = getattr(self.table.c, self.pkname)
    expression = self.table.select().where(pk_column == self.pk)

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

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

_save async

_save(**kwargs)

Performs the save instruction.

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

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

async def _save(self, **kwargs: typing.Any) -> "Model":
    """
    Performs the save instruction.
    """
    expression = self.table.insert().values(**kwargs)
    autoincrement_value = await self.database.execute(expression)
    # sqlalchemy supports only one autoincrement column
    if autoincrement_value:
        if isinstance(autoincrement_value, Row):
            assert len(autoincrement_value) == 1
            autoincrement_value = autoincrement_value[0]
        column = self.table.autoincrement_column
        # can be explicit set, which causes an invalid value returned
        if column is not None and column.key not in kwargs:
            saffier_setattr(self, column.key, autoincrement_value)
    return self

save async

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

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

PARAMETER	DESCRIPTION
force_save	TYPE: bool DEFAULT: False
values	TYPE: Any DEFAULT: None
**kwargs	TYPE: Any DEFAULT: {}

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

async def save(
    self,
    force_save: bool = False,
    values: typing.Any = None,
    **kwargs: typing.Any,
) -> Union[Type["Model"], Any]:
    """
    Performs a save of a given model instance.
    When creating a user it will make sure it can update existing or
    create a new one.
    """
    await self.signals.pre_save.send(sender=self.__class__, instance=self)

    extracted_fields = self.extract_db_fields()

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

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

    fields = {
        key: field.validator for key, field in self.fields.items() if key in extracted_fields
    }
    if values:
        kwargs = self._update_auto_now_fields(values, self.fields)
    else:
        validator = Schema(fields=fields)
        kwargs = self._update_auto_now_fields(validator.check(extracted_fields), self.fields)

    # Performs the update or the create based on a possible existing primary key
    if getattr(self, "pk", None) is None or force_save:
        await self._save(**kwargs)
    else:
        await self.signals.pre_update.send(sender=self.__class__, instance=self, kwargs=kwargs)
        await self.update(**kwargs)
        await self.signals.post_update.send(sender=self.__class__, instance=self)

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

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

__getattr__

__getattr__(name)

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

PARAMETER	DESCRIPTION
name	TYPE: str

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

def __getattr__(self, name: str) -> Any:
    """
    Run an one off query to populate any foreign key making sure
    it runs only once per foreign key avoiding multiple database calls.
    """
    if name not in self.__dict__ and name in self.fields and name != self.pkname:
        run_sync(self.load())
        return self.__dict__[name]
    return super().__getattr__(name)