Skip to content

Model class

saffier.Model 

Model(**kwargs)

Bases: ModelRow, DeclarativeMixin

The models will always have an id attribute as primery key. The primary key can be whatever desired, from IntegerField, FloatField to UUIDField as long as the id field is explicitly declared or else it defaults to BigIntegerField.

PARAMETER DESCRIPTION
**kwargs

TYPE: Any DEFAULT: {}

Source code in saffier/core/db/models/base.py 
38
39
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 
13
14
15
@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 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
@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 
26
27
28
29
30
31
32
33
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 
35
36
37
38
39
40
41
42
43
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 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
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 
101
102
103
104
105
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 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
@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)

Performs the operation of building the core SQLAlchemy Table object. Builds the constrainst, indexes, columns and metadata based on the provided Meta class object.

PARAMETER DESCRIPTION
schema

TYPE: Optional[str] DEFAULT: None

Source code in saffier/core/db/models/base.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
@classmethod
def build(cls, schema: Optional[str] = None) -> sqlalchemy.Table:
    """
    Performs the operation of building the core SQLAlchemy Table object.
    Builds the constrainst, indexes, columns and metadata based on the
    provided Meta class object.
    """
    tablename = cls.meta.tablename
    metadata: sqlalchemy.MetaData = cast("sqlalchemy.MetaData", cls.meta.registry._metadata)  # type: ignore
    metadata.schema = schema

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

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

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

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

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

_get_unique_constraints classmethod 

_get_unique_constraints(columns)

Returns the unique constraints for the model.

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

PARAMETER DESCRIPTION
columns

TYPE: Sequence

Source code in saffier/core/db/models/base.py 
162
163
164
165
166
167
168
169
170
171
172
173
@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 
175
176
177
178
179
180
@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 
182
183
184
185
186
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 
188
189
190
191
192
193
194
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 
196
197
198
199
200
201
202
203
204
205
206
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 
208
209
210
211
212
213
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 
215
216
217
218
219
220
221
222
def __eq__(self, other: Any) -> bool:
    if self.__class__ != other.__class__:
        return False
    original = self.__get_instance_values(instance=self)
    other_values = self.__get_instance_values(instance=other)
    if original != other_values:
        return False
    return True

from_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 
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
@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 
160
161
162
163
164
165
@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 
167
168
169
170
171
172
173
174
175
176
177
178
@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 
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
@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 
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
@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 
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
@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 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
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 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
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 
91
92
93
94
95
96
97
98
99
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 
101
102
103
104
105
106
107
108
109
110
111
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 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
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 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
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 
178
179
180
181
182
183
184
185
186
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)