Skip to content

QuerySet class

saffier.QuerySet 

QuerySet(model_class=None, database=None, filter_clauses=None, or_clauses=None, select_related=None, prefetch_related=None, limit_count=None, limit_offset=None, order_by=None, group_by=None, distinct_on=None, only_fields=None, defer_fields=None, m2m_related=None, using_schema=None, table=None, exclude_secrets=False)

Bases: BaseQuerySet, QuerySetProtocol

QuerySet object used for query retrieving.

PARAMETER DESCRIPTION
model_class

TYPE: Union[Type[Model], None] DEFAULT: None

database

TYPE: Union[Database, None] DEFAULT: None

filter_clauses

TYPE: Any DEFAULT: None

or_clauses

TYPE: Any DEFAULT: None

select_related

TYPE: Any DEFAULT: None

prefetch_related

TYPE: Any DEFAULT: None

limit_count

TYPE: Any DEFAULT: None

limit_offset

TYPE: Any DEFAULT: None

order_by

TYPE: Any DEFAULT: None

group_by

TYPE: Any DEFAULT: None

distinct_on

TYPE: Any DEFAULT: None

only_fields

TYPE: Any DEFAULT: None

defer_fields

TYPE: Any DEFAULT: None

m2m_related

TYPE: Any DEFAULT: None

using_schema

TYPE: Any DEFAULT: None

table

TYPE: Any DEFAULT: None

exclude_secrets

TYPE: Any DEFAULT: False

Source code in saffier/core/db/querysets/base.py 
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
def __init__(
    self,
    model_class: Union[Type["Model"], None] = None,
    database: Union["Database", None] = None,
    filter_clauses: Any = None,
    or_clauses: Any = None,
    select_related: Any = None,
    prefetch_related: Any = None,
    limit_count: Any = None,
    limit_offset: Any = None,
    order_by: Any = None,
    group_by: Any = None,
    distinct_on: Any = None,
    only_fields: Any = None,
    defer_fields: Any = None,
    m2m_related: Any = None,
    using_schema: Any = None,
    table: Any = None,
    exclude_secrets: Any = False,
) -> None:
    super().__init__(model_class=model_class)
    self.model_class = cast("Type[Model]", model_class)
    self.filter_clauses = [] if filter_clauses is None else filter_clauses
    self.or_clauses = [] if or_clauses is None else or_clauses
    self.limit_count = limit_count
    self._select_related = [] if select_related is None else select_related
    self._prefetch_related = [] if prefetch_related is None else prefetch_related
    self._offset = limit_offset
    self._order_by = [] if order_by is None else order_by
    self._group_by = [] if group_by is None else group_by
    self.distinct_on = [] if distinct_on is None else distinct_on
    self._only = [] if only_fields is None else only_fields
    self._defer = [] if defer_fields is None else defer_fields
    self._expression = None
    self._cache = None
    self._m2m_related = m2m_related  # type: ignore
    self.using_schema = using_schema
    self._exclude_secrets = exclude_secrets or False
    self.extra: Dict[str, Any] = {}

    if self.is_m2m and not self._m2m_related:
        self._m2m_related = self.model_class.meta.multi_related[0]

    if table is not None:
        self.table = table
    if database is not None:
        self.database = database

model_class instance-attribute 

model_class = cast('Type[Model]', model_class)

database instance-attribute 

database = database

table instance-attribute 

table = table

is_m2m property 

is_m2m

filter_clauses instance-attribute 

filter_clauses = [] if filter_clauses is None else filter_clauses

or_clauses instance-attribute 

or_clauses = [] if or_clauses is None else or_clauses

limit_count instance-attribute 

limit_count = limit_count

_select_related = [] if select_related is None else select_related

_prefetch_related = [] if prefetch_related is None else prefetch_related

_offset instance-attribute 

_offset = limit_offset

_order_by instance-attribute 

_order_by = [] if order_by is None else order_by

_group_by instance-attribute 

_group_by = [] if group_by is None else group_by

distinct_on instance-attribute 

distinct_on = [] if distinct_on is None else distinct_on

_only instance-attribute 

_only = [] if only_fields is None else only_fields

_defer instance-attribute 

_defer = [] if defer_fields is None else defer_fields

_expression instance-attribute 

_expression = None

_cache instance-attribute 

_cache = None

using_schema instance-attribute 

using_schema = using_schema

_exclude_secrets instance-attribute 

_exclude_secrets = exclude_secrets or False

extra instance-attribute 

extra = {}

sql property writable 

sql

execute async 

execute()
Source code in saffier/core/db/querysets/protocols.py 
16
17
async def execute(self) -> typing.Any:
    raise NotImplementedError()

_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

prefetch_related(*prefetch)

Performs a reverse lookup for the foreignkeys. This is different from the select_related. Select related performs a follows the SQL foreign relation whereas the prefetch_related follows the python relations.

PARAMETER DESCRIPTION
*prefetch

TYPE: Prefetch DEFAULT: ()

Source code in saffier/core/db/querysets/prefetch.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def prefetch_related(self, *prefetch: Prefetch) -> "QuerySet":
    """
    Performs a reverse lookup for the foreignkeys. This is different
    from the select_related. Select related performs a follows the SQL foreign relation
    whereas the prefetch_related follows the python relations.
    """
    queryset: "QuerySet" = self._clone()

    if not isinstance(prefetch, (list, tuple)):
        prefetch = cast("List[Prefetch]", [prefetch])  # type: ignore

    if isinstance(prefetch, tuple):
        prefetch = list(prefetch)  # type: ignore

    if any(not isinstance(value, Prefetch) for value in prefetch):
        raise QuerySetError("The prefetch_related must have Prefetch type objects only.")

    prefetch = list(self._prefetch_related) + prefetch  # type: ignore
    queryset._prefetch_related = prefetch
    return queryset

using 

using(schema)

Enables and switches the db schema.

Generates the registry object pointing to the desired schema using the same connection.

PARAMETER DESCRIPTION
schema

TYPE: str

Source code in saffier/core/db/querysets/mixins.py 
65
66
67
68
69
70
71
72
73
def using(self, schema: str) -> "QuerySet":
    """
    Enables and switches the db schema.

    Generates the registry object pointing to the desired schema
    using the same connection.
    """
    queryset = set_queryset_schema(self, self.model_class, value=schema)
    return queryset

using_with_db 

using_with_db(connection_name, schema=None)

Enables and switches the database connection.

Generates the new queryset using the selected connection provided in the extra of the model registry.

PARAMETER DESCRIPTION
connection_name

TYPE: str

schema

TYPE: Optional[str] DEFAULT: None

Source code in saffier/core/db/querysets/mixins.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def using_with_db(self, connection_name: str, schema: Optional[str] = None) -> "QuerySet":
    """
    Enables and switches the database connection.

    Generates the new queryset using the selected connection provided in the extra of the model
    registry.
    """
    assert (
        connection_name in self.model_class.meta.registry.extra
    ), f"`{connection_name}` is not in the connections extra of the model`{self.model_class.__name__}` registry"

    connection: Type["Registry"] = self.model_class.meta.registry.extra[connection_name]
    if schema:
        return set_queryset_database(self, self.model_class, connection, schema)
    queryset = set_queryset_database(self, self.model_class, connection)
    return queryset

_build_order_by_expression 

_build_order_by_expression(order_by, expression)

Builds the order by expression

PARAMETER DESCRIPTION
order_by

TYPE: Any

expression

TYPE: Any

Source code in saffier/core/db/querysets/base.py 
91
92
93
94
95
def _build_order_by_expression(self, order_by: Any, expression: Any) -> Any:
    """Builds the order by expression"""
    order_by = list(map(self._prepare_order_by, order_by))
    expression = expression.order_by(*order_by)
    return expression

_build_group_by_expression 

_build_group_by_expression(group_by, expression)

Builds the group by expression

PARAMETER DESCRIPTION
group_by

TYPE: Any

expression

TYPE: Any

Source code in saffier/core/db/querysets/base.py 
 97
 98
 99
100
101
def _build_group_by_expression(self, group_by: Any, expression: Any) -> Any:
    """Builds the group by expression"""
    group_by = list(map(self._prepare_group_by, group_by))
    expression = expression.group_by(*group_by)
    return expression

_build_filter_clauses_expression 

_build_filter_clauses_expression(filter_clauses, expression)

Builds the filter clauses expression

PARAMETER DESCRIPTION
filter_clauses

TYPE: Any

expression

TYPE: Any

Source code in saffier/core/db/querysets/base.py 
103
104
105
106
107
108
109
110
def _build_filter_clauses_expression(self, filter_clauses: Any, expression: Any) -> Any:
    """Builds the filter clauses expression"""
    if len(filter_clauses) == 1:
        clause = filter_clauses[0]
    else:
        clause = sqlalchemy.sql.and_(*filter_clauses)
    expression = expression.where(clause)
    return expression

_build_or_clauses_expression 

_build_or_clauses_expression(or_clauses, expression)

Builds the filter clauses expression

PARAMETER DESCRIPTION
or_clauses

TYPE: Any

expression

TYPE: Any

Source code in saffier/core/db/querysets/base.py 
112
113
114
115
116
117
118
119
def _build_or_clauses_expression(self, or_clauses: Any, expression: Any) -> Any:
    """Builds the filter clauses expression"""
    if len(or_clauses) == 1:
        clause = or_clauses[0]
    else:
        clause = sqlalchemy.sql.or_(*or_clauses)
    expression = expression.where(clause)
    return expression

_build_select_distinct 

_build_select_distinct(distinct_on, expression)

Filters selects only specific fields

PARAMETER DESCRIPTION
distinct_on

TYPE: Any

expression

TYPE: Any

Source code in saffier/core/db/querysets/base.py 
121
122
123
124
125
def _build_select_distinct(self, distinct_on: Any, expression: Any) -> Any:
    """Filters selects only specific fields"""
    distinct_on = list(map(self._prepare_fields_for_distinct, distinct_on))
    expression = expression.distinct(*distinct_on)
    return expression

_is_multiple_foreign_key 

_is_multiple_foreign_key(model_class)

Checks if the table has multiple foreign keys to the same table. Iterates through all fields of type ForeignKey and OneToOneField and checks if there are multiple FKs to the same destination table.

Returns a tuple of bool, list of tuples

PARAMETER DESCRIPTION
model_class

TYPE: Union[Type[Model], Type[ReflectModel]]

Source code in saffier/core/db/querysets/base.py 
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
def _is_multiple_foreign_key(
    self, model_class: Union[Type["Model"], Type["ReflectModel"]]
) -> Tuple[bool, List[Tuple[str, str, str]]]:
    """
    Checks if the table has multiple foreign keys to the same table.
    Iterates through all fields of type ForeignKey and OneToOneField and checks if there are
    multiple FKs to the same destination table.

    Returns a tuple of bool, list of tuples
    """
    fields = model_class.fields  # type: ignore
    foreign_keys = []
    has_many = False

    counter = 0

    for key, value in fields.items():
        tablename = None

        if counter > 1:
            has_many = True

        if isinstance(value, (saffier.ForeignKey, saffier.OneToOneField)):
            tablename = value.to if isinstance(value.to, str) else value.to.__name__

            if tablename not in foreign_keys:
                foreign_keys.append((key, tablename, value.related_name))
                counter += 1
            else:
                counter += 1

    return has_many, foreign_keys  # type: ignore

_build_tables_select_from_relationship 

_build_tables_select_from_relationship()

Builds the tables relationships and joins. When a table contains more than one foreign key pointing to the same destination table, a lookup for the related field is made to understand from which foreign key the table is looked up from.

Source code in saffier/core/db/querysets/base.py 
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
def _build_tables_select_from_relationship(self) -> Any:
    """
    Builds the tables relationships and joins.
    When a table contains more than one foreign key pointing to the same
    destination table, a lookup for the related field is made to understand
    from which foreign key the table is looked up from.
    """
    queryset: "QuerySet" = self._clone()

    tables = [queryset.table]
    select_from = queryset.table
    has_many_fk_same_table = False

    # Select related
    for item in queryset._select_related:
        # For m2m relationships
        model_class = queryset.model_class
        select_from = queryset.table

        for part in item.split("__"):
            try:
                model_class = model_class.fields[part].target
            except KeyError:
                # Check related fields
                model_class = getattr(model_class, part).related_from
                has_many_fk_same_table, keys = self._is_multiple_foreign_key(model_class)

            table = model_class.table

            # If there is multiple FKs to the same table
            if not has_many_fk_same_table:
                select_from = sqlalchemy.sql.join(select_from, table)  # type: ignore
            else:
                lookup_field = None

                # Extract the table field from the related
                # Assign to a lookup_field
                for values in keys:
                    field, _, related_name = values
                    if related_name == part:
                        lookup_field = field
                        break

                select_from = sqlalchemy.sql.join(  # type: ignore
                    select_from,
                    table,
                    select_from.c.id == getattr(table.c, lookup_field),
                )

            tables.append(table)
    return tables, select_from

_validate_only_and_defer 

_validate_only_and_defer()
Source code in saffier/core/db/querysets/base.py 
212
213
214
def _validate_only_and_defer(self) -> None:
    if self._only and self._defer:
        raise QuerySetError("You cannot use .only() and .defer() at the same time.")

_secret_recursive_names 

_secret_recursive_names(model_class, columns=None)

Recursively gets the names of the fields excluding the secrets.

PARAMETER DESCRIPTION
model_class

TYPE: Any

columns

TYPE: Union[List[str], None] DEFAULT: None

Source code in saffier/core/db/querysets/base.py 
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
def _secret_recursive_names(
    self, model_class: Any, columns: Union[List[str], None] = None
) -> List[str]:
    """
    Recursively gets the names of the fields excluding the secrets.
    """
    if columns is None:
        columns = []

    for name, field in model_class.fields.items():
        if isinstance(field, saffier_fields.ForeignKey):
            # Making sure the foreign key is always added unless is a secret
            if not field.secret:
                columns.append(name)
                columns.extend(
                    self._secret_recursive_names(model_class=field.target, columns=columns)
                )
            continue
        if not field.secret:
            columns.append(name)

    columns = list(set(columns))
    return columns

_build_select 

_build_select()

Builds the query select based on the given parameters and filters.

Source code in saffier/core/db/querysets/base.py 
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
def _build_select(self) -> Any:
    """
    Builds the query select based on the given parameters and filters.
    """
    queryset: "QuerySet" = self._clone()

    queryset._validate_only_and_defer()
    tables, select_from = queryset._build_tables_select_from_relationship()
    expression = sqlalchemy.sql.select(*tables)
    expression = expression.select_from(select_from)

    if queryset._only:
        expression = expression.with_only_columns(*queryset._only)

    if queryset._defer:
        columns = [
            column for column in select_from.columns if column.name not in queryset._defer
        ]
        expression = expression.with_only_columns(*columns)

    if queryset._exclude_secrets:
        model_columns = queryset._secret_recursive_names(model_class=queryset.model_class)
        columns = [column for column in select_from.columns if column.name in model_columns]
        expression = expression.with_only_columns(*columns)

    if queryset.filter_clauses:
        expression = queryset._build_filter_clauses_expression(
            queryset.filter_clauses, expression=expression
        )

    if queryset.or_clauses:
        expression = queryset._build_or_clauses_expression(
            queryset.or_clauses, expression=expression
        )

    if queryset._order_by:
        expression = queryset._build_order_by_expression(
            queryset._order_by, expression=expression
        )

    if queryset.limit_count:
        expression = expression.limit(queryset.limit_count)

    if queryset._offset:
        expression = expression.offset(queryset._offset)

    if queryset._group_by:
        expression = queryset._build_group_by_expression(
            queryset._group_by, expression=expression
        )

    if queryset.distinct_on:
        expression = queryset._build_select_distinct(
            queryset.distinct_on, expression=expression
        )

    queryset._expression = expression  # type: ignore
    return expression

_filter_query 

_filter_query(exclude=False, or_=False, **kwargs)
PARAMETER DESCRIPTION
exclude

TYPE: bool DEFAULT: False

or_

TYPE: bool DEFAULT: False

**kwargs

TYPE: Any DEFAULT: {}

Source code in saffier/core/db/querysets/base.py 
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
def _filter_query(self, exclude: bool = False, or_: bool = False, **kwargs: Any) -> "QuerySet":
    from saffier.core.db.models.model import Model

    clauses = []
    filter_clauses = self.filter_clauses
    or_clauses = self.or_clauses
    select_related = list(self._select_related)
    prefetch_related = list(self._prefetch_related)

    if kwargs.get("pk"):
        pk_name = self.model_class.pkname
        kwargs[pk_name] = kwargs.pop("pk")

    # Making sure for queries we use the main class and not the proxy
    # And enable the parent
    if self.model_class.is_proxy_model:
        self.model_class = self.model_class.parent

    for key, value in kwargs.items():
        if "__" in key:
            parts = key.split("__")

            # Determine if we should treat the final part as a
            # filter operator or as a related field.
            if parts[-1] in settings.filter_operators:
                op = parts[-1]
                field_name = parts[-2]
                related_parts = parts[:-2]
            else:
                op = "exact"
                field_name = parts[-1]
                related_parts = parts[:-1]

            model_class = self.model_class
            if related_parts:
                # Add any implied select_related
                related_str = "__".join(related_parts)
                if related_str not in select_related:
                    select_related.append(related_str)

                # Walk the relationships to the actual model class
                # against which the comparison is being made.
                for part in related_parts:
                    try:
                        model_class = model_class.fields[part].target
                    except KeyError:
                        model_class = getattr(model_class, part).related_from

            column = model_class.table.columns[field_name]

        else:
            op = "exact"
            try:
                column = self.table.columns[key]
            except KeyError as error:
                # Check for related fields
                # if an Attribute error is raised, we need to make sure
                # It raises the KeyError from the previous check
                try:
                    model_class = getattr(self.model_class, key).related_to
                    column = model_class.table.columns[settings.default_related_lookup_field]
                except AttributeError:
                    raise KeyError(str(error)) from error

        # Map the operation code onto SQLAlchemy's ColumnElement
        # https://docs.sqlalchemy.org/en/latest/core/sqlelement.html#sqlalchemy.sql.expression.ColumnElement
        op_attr = settings.filter_operators[op]
        has_escaped_character = False

        if op in ["contains", "icontains"]:
            has_escaped_character = any(c for c in self.ESCAPE_CHARACTERS if c in value)
            if has_escaped_character:
                # enable escape modifier
                for char in self.ESCAPE_CHARACTERS:
                    value = value.replace(char, f"\\{char}")
            value = f"%{value}%"

        if isinstance(value, Model):
            value = value.pk

        clause = getattr(column, op_attr)(value)
        clause.modifiers["escape"] = "\\" if has_escaped_character else None

        clauses.append(clause)

    if exclude:
        if not or_:
            filter_clauses.append(sqlalchemy.not_(sqlalchemy.sql.and_(*clauses)))
        else:
            or_clauses.append(sqlalchemy.not_(sqlalchemy.sql.and_(*clauses)))
    else:
        if not or_:
            filter_clauses += clauses
        else:
            or_clauses += clauses

    return cast(
        "QuerySet",
        self.__class__(
            model_class=self.model_class,
            database=self._database,
            filter_clauses=filter_clauses,
            or_clauses=or_clauses,
            select_related=select_related,
            prefetch_related=prefetch_related,
            limit_count=self.limit_count,
            limit_offset=self._offset,
            order_by=self._order_by,
            only_fields=self._only,
            defer_fields=self._defer,
            m2m_related=self.m2m_related,
            exclude_secrets=self._exclude_secrets,
            table=self.table,
            using_schema=self.using_schema,
        ),
    )

_validate_kwargs 

_validate_kwargs(**kwargs)
PARAMETER DESCRIPTION
**kwargs

TYPE: Any DEFAULT: {}

Source code in saffier/core/db/querysets/base.py 
416
417
418
419
420
421
422
423
def _validate_kwargs(self, **kwargs: Any) -> Any:
    fields = self.model_class.fields
    validator = Schema(fields={key: value.validator for key, value in fields.items()})
    kwargs = validator.check(kwargs)
    for key, value in fields.items():
        if value.validator.read_only and value.validator.has_default():
            kwargs[key] = value.validator.get_default_value()
    return kwargs

_prepare_order_by 

_prepare_order_by(order_by)
PARAMETER DESCRIPTION
order_by

TYPE: str

Source code in saffier/core/db/querysets/base.py 
425
426
427
428
429
def _prepare_order_by(self, order_by: str) -> Any:
    reverse = order_by.startswith("-")
    order_by = order_by.lstrip("-")
    order_col = self.table.columns[order_by]
    return order_col.desc() if reverse else order_col

_prepare_group_by 

_prepare_group_by(group_by)
PARAMETER DESCRIPTION
group_by

TYPE: str

Source code in saffier/core/db/querysets/base.py 
431
432
433
434
def _prepare_group_by(self, group_by: str) -> Any:
    group_by = group_by.lstrip("-")
    group_col = self.table.columns[group_by]
    return group_col

_prepare_fields_for_distinct 

_prepare_fields_for_distinct(distinct_on)
PARAMETER DESCRIPTION
distinct_on

TYPE: str

Source code in saffier/core/db/querysets/base.py 
436
437
438
def _prepare_fields_for_distinct(self, distinct_on: str) -> Any:
    _distinct_on: sqlalchemy.Column = self.table.columns[distinct_on]
    return _distinct_on

_clone 

_clone()

Return a copy of the current QuerySet that's ready for another operation.

Source code in saffier/core/db/querysets/base.py 
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
def _clone(self) -> Any:
    """
    Return a copy of the current QuerySet that's ready for another
    operation.
    """
    queryset = self.__class__.__new__(self.__class__)
    queryset.model_class = self.model_class

    # Making sure the registry schema takes precendent with
    # Any provided using
    if not self.model_class.meta.registry.db_schema:
        schema = get_schema()
        if self.using_schema is None and schema is not None:
            self.using_schema = schema
        queryset.model_class.table = self.model_class.build(self.using_schema)

    queryset.filter_clauses = copy.copy(self.filter_clauses)
    queryset.or_clauses = copy.copy(self.or_clauses)
    queryset.limit_count = self.limit_count
    queryset._select_related = copy.copy(self._select_related)
    queryset._prefetch_related = copy.copy(self._prefetch_related)
    queryset._offset = self._offset
    queryset._order_by = copy.copy(self._order_by)
    queryset._group_by = copy.copy(self._group_by)
    queryset.distinct_on = copy.copy(self.distinct_on)
    queryset._expression = self._expression
    queryset._cache = self._cache
    queryset._m2m_related = self._m2m_related
    queryset._only = copy.copy(self._only)
    queryset._defer = copy.copy(self._defer)
    queryset._database = self.database
    queryset.table = self.table
    queryset.extra = self.extra
    queryset._exclude_secrets = self._exclude_secrets
    queryset.using_schema = self.using_schema

    return queryset

__aiter__ async 

__aiter__()
Source code in saffier/core/db/querysets/base.py 
495
496
497
async def __aiter__(self) -> AsyncIterator[SaffierModel]:
    for value in await self:
        yield value

_set_query_expression 

_set_query_expression(expression)

Sets the value of the sql property to the expression used.

PARAMETER DESCRIPTION
expression

TYPE: Any

Source code in saffier/core/db/querysets/base.py 
499
500
501
502
503
504
def _set_query_expression(self, expression: Any) -> None:
    """
    Sets the value of the sql property to the expression used.
    """
    self.sql = expression
    self.model_class.raw_query = self.sql

_filter_or_exclude 

_filter_or_exclude(clause=None, exclude=False, **kwargs)

Filters or excludes a given clause for a specific QuerySet.

PARAMETER DESCRIPTION
clause

TYPE: Optional[BinaryExpression] DEFAULT: None

exclude

TYPE: bool DEFAULT: False

**kwargs

TYPE: Any DEFAULT: {}

Source code in saffier/core/db/querysets/base.py 
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
def _filter_or_exclude(
    self,
    clause: Optional[sqlalchemy.sql.expression.BinaryExpression] = None,
    exclude: bool = False,
    **kwargs: Any,
) -> "QuerySet":
    """
    Filters or excludes a given clause for a specific QuerySet.
    """
    queryset: "QuerySet" = self._clone()

    if clause is None:
        if not exclude:
            return queryset._filter_query(**kwargs)
        return queryset._filter_query(exclude=exclude, **kwargs)

    queryset.filter_clauses.append(clause)
    return queryset

filter 

filter(clause=None, **kwargs)

Filters the QuerySet by the given kwargs and clause.

PARAMETER DESCRIPTION
clause

TYPE: Optional[BinaryExpression] DEFAULT: None

**kwargs

TYPE: Any DEFAULT: {}

Source code in saffier/core/db/querysets/base.py 
525
526
527
528
529
530
531
532
533
def filter(
    self,
    clause: Optional[sqlalchemy.sql.expression.BinaryExpression] = None,
    **kwargs: Any,
) -> "QuerySet":
    """
    Filters the QuerySet by the given kwargs and clause.
    """
    return self._filter_or_exclude(clause=clause, **kwargs)

or_ 

or_(clause=None, **kwargs)

Filters the QuerySet by the OR operand.

PARAMETER DESCRIPTION
clause

TYPE: Optional[BinaryExpression] DEFAULT: None

**kwargs

TYPE: Any DEFAULT: {}

Source code in saffier/core/db/querysets/base.py 
535
536
537
538
539
540
541
542
543
544
545
def or_(
    self,
    clause: Optional[sqlalchemy.sql.expression.BinaryExpression] = None,
    **kwargs: Any,
) -> "QuerySet":
    """
    Filters the QuerySet by the OR operand.
    """
    queryset: "QuerySet" = self._clone()
    queryset = self.filter(clause=clause, or_=True, **kwargs)
    return queryset

and_ 

and_(clause=None, **kwargs)

Filters the QuerySet by the AND operand.

PARAMETER DESCRIPTION
clause

TYPE: Optional[BinaryExpression] DEFAULT: None

**kwargs

TYPE: Any DEFAULT: {}

Source code in saffier/core/db/querysets/base.py 
547
548
549
550
551
552
553
554
555
556
557
def and_(
    self,
    clause: Optional[sqlalchemy.sql.expression.BinaryExpression] = None,
    **kwargs: Any,
) -> "QuerySet":
    """
    Filters the QuerySet by the AND operand.
    """
    queryset: "QuerySet" = self._clone()
    queryset = self.filter(clause=clause, **kwargs)
    return queryset

not_ 

not_(clause=None, **kwargs)

Filters the QuerySet by the NOT operand.

PARAMETER DESCRIPTION
clause

TYPE: Optional[BinaryExpression] DEFAULT: None

**kwargs

TYPE: Any DEFAULT: {}

Source code in saffier/core/db/querysets/base.py 
559
560
561
562
563
564
565
566
567
568
569
def not_(
    self,
    clause: Optional[sqlalchemy.sql.expression.BinaryExpression] = None,
    **kwargs: Any,
) -> "QuerySet":
    """
    Filters the QuerySet by the NOT operand.
    """
    queryset: "QuerySet" = self._clone()
    queryset = queryset.exclude(clause=clause, **kwargs)
    return queryset

exclude 

exclude(clause=None, **kwargs)

Exactly the same as the filter but for the exclude.

PARAMETER DESCRIPTION
clause

TYPE: Optional[BinaryExpression] DEFAULT: None

**kwargs

TYPE: Any DEFAULT: {}

Source code in saffier/core/db/querysets/base.py 
571
572
573
574
575
576
577
578
579
580
581
def exclude(
    self,
    clause: Optional[sqlalchemy.sql.expression.BinaryExpression] = None,
    **kwargs: Any,
) -> "QuerySet":
    """
    Exactly the same as the filter but for the exclude.
    """
    queryset: "QuerySet" = self._clone()
    queryset = self._filter_or_exclude(clause=clause, exclude=True, **kwargs)
    return queryset

exclude_secrets 

exclude_secrets(clause=None, **kwargs)

Excludes any field that contains the secret=True declared from being leaked.

PARAMETER DESCRIPTION
clause

TYPE: Optional[BinaryExpression] DEFAULT: None

**kwargs

TYPE: Any DEFAULT: {}

Source code in saffier/core/db/querysets/base.py 
583
584
585
586
587
588
589
590
591
592
593
594
def exclude_secrets(
    self,
    clause: Optional[sqlalchemy.sql.expression.BinaryExpression] = None,
    **kwargs: Any,
) -> "QuerySet":
    """
    Excludes any field that contains the `secret=True` declared from being leaked.
    """
    queryset: "QuerySet" = self._clone()
    queryset._exclude_secrets = True
    queryset = queryset.filter(clause=clause, **kwargs)
    return queryset

lookup 

lookup(term)

Broader way of searching for a given term

PARAMETER DESCRIPTION
term

TYPE: Any

Source code in saffier/core/db/querysets/base.py 
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
def lookup(self, term: Any) -> "QuerySet":
    """
    Broader way of searching for a given term
    """
    queryset: "QuerySet" = self._clone()
    if not term:
        return queryset

    filter_clauses = list(queryset.filter_clauses)
    value = f"%{term}%"

    search_fields = [
        name
        for name, field in queryset.model_class.fields.items()
        if isinstance(field, (CharField, TextField))
    ]
    search_clauses = [queryset.table.columns[name].ilike(value) for name in search_fields]

    if len(search_clauses) > 1:
        filter_clauses.append(sqlalchemy.sql.or_(*search_clauses))
    else:
        filter_clauses.extend(search_clauses)

    return queryset

order_by 

order_by(*order_by)

Returns a QuerySet ordered by the given fields.

PARAMETER DESCRIPTION
*order_by

TYPE: str DEFAULT: ()

Source code in saffier/core/db/querysets/base.py 
621
622
623
624
625
626
627
def order_by(self, *order_by: str) -> "QuerySet":
    """
    Returns a QuerySet ordered by the given fields.
    """
    queryset: "QuerySet" = self._clone()
    queryset._order_by = order_by
    return queryset

limit 

limit(limit_count)

Returns a QuerySet limited by.

PARAMETER DESCRIPTION
limit_count

TYPE: int

Source code in saffier/core/db/querysets/base.py 
629
630
631
632
633
634
635
def limit(self, limit_count: int) -> "QuerySet":
    """
    Returns a QuerySet limited by.
    """
    queryset: "QuerySet" = self._clone()
    queryset.limit_count = limit_count
    return queryset

offset 

offset(offset)

Returns a Queryset limited by the offset.

PARAMETER DESCRIPTION
offset

TYPE: int

Source code in saffier/core/db/querysets/base.py 
637
638
639
640
641
642
643
def offset(self, offset: int) -> "QuerySet":
    """
    Returns a Queryset limited by the offset.
    """
    queryset: "QuerySet" = self._clone()
    queryset._offset = offset
    return queryset

group_by 

group_by(*group_by)

Returns the values grouped by the given fields.

PARAMETER DESCRIPTION
*group_by

TYPE: str DEFAULT: ()

Source code in saffier/core/db/querysets/base.py 
645
646
647
648
649
650
651
def group_by(self, *group_by: str) -> "QuerySet":
    """
    Returns the values grouped by the given fields.
    """
    queryset: "QuerySet" = self._clone()
    queryset._group_by = group_by
    return queryset

distinct 

distinct(*distinct_on)

Returns a queryset with distinct results.

PARAMETER DESCRIPTION
*distinct_on

TYPE: str DEFAULT: ()

Source code in saffier/core/db/querysets/base.py 
653
654
655
656
657
658
659
def distinct(self, *distinct_on: str) -> "QuerySet":
    """
    Returns a queryset with distinct results.
    """
    queryset: "QuerySet" = self._clone()
    queryset.distinct_on = distinct_on
    return queryset

only 

only(*fields)

Returns a list of models with the selected only fields and always the primary key.

PARAMETER DESCRIPTION
*fields

TYPE: Sequence[str] DEFAULT: ()

Source code in saffier/core/db/querysets/base.py 
661
662
663
664
665
666
667
668
669
670
671
672
def only(self, *fields: Sequence[str]) -> "QuerySet":
    """
    Returns a list of models with the selected only fields and always the primary
    key.
    """
    only_fields = [sqlalchemy.text(field) for field in fields]
    if self.model_class.pkname not in fields:
        only_fields.insert(0, sqlalchemy.text(self.model_class.pkname))

    queryset: "QuerySet" = self._clone()
    queryset._only = only_fields
    return queryset

defer 

defer(*fields)

Returns a list of models with the selected only fields and always the primary key.

PARAMETER DESCRIPTION
*fields

TYPE: Sequence[str] DEFAULT: ()

Source code in saffier/core/db/querysets/base.py 
674
675
676
677
678
679
680
681
def defer(self, *fields: Sequence[str]) -> "QuerySet":
    """
    Returns a list of models with the selected only fields and always the primary
    key.
    """
    queryset: "QuerySet" = self._clone()
    queryset._defer = fields
    return queryset

select_related(related)

Returns a QuerySet that will “follow” foreign-key relationships, selecting additional related-object data when it executes its query.

This is a performance booster which results in a single more complex query but means

later use of foreign-key relationships won’t require database queries.

PARAMETER DESCRIPTION
related

TYPE: Any

Source code in saffier/core/db/querysets/base.py 
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
def select_related(self, related: Any) -> "QuerySet":
    """
    Returns a QuerySet that will “follow” foreign-key relationships, selecting additional
    related-object data when it executes its query.

    This is a performance booster which results in a single more complex query but means

    later use of foreign-key relationships won’t require database queries.
    """
    queryset: "QuerySet" = self._clone()
    if not isinstance(related, (list, tuple)):
        related = [related]

    related = list(queryset._select_related) + related
    queryset._select_related = related
    return queryset

values async 

values(fields=None, exclude=None, exclude_none=False, flatten=False, **kwargs)

Returns the results in a python dictionary format.

PARAMETER DESCRIPTION
fields

TYPE: Union[Sequence[str], str, None] DEFAULT: None

exclude

TYPE: Union[Sequence[str], Set[str]] DEFAULT: None

exclude_none

TYPE: bool DEFAULT: False

flatten

TYPE: bool DEFAULT: False

**kwargs

TYPE: Any DEFAULT: {}

Source code in saffier/core/db/querysets/base.py 
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
async def values(
    self,
    fields: Union[Sequence[str], str, None] = None,
    exclude: Union[Sequence[str], Set[str]] = None,
    exclude_none: bool = False,
    flatten: bool = False,
    **kwargs: Any,
) -> List[Any]:
    """
    Returns the results in a python dictionary format.
    """
    fields = fields or []
    queryset: "QuerySet" = self._clone()
    rows: List[Type["Model"]] = await queryset.all()

    if not isinstance(fields, list):
        raise QuerySetError(detail="Fields must be an iterable.")

    if not fields:
        rows = [row.model_dump(exclude=exclude, exclude_none=exclude_none) for row in rows]  # type: ignore
    else:
        rows = [
            row.model_dump(exclude=exclude, exclude_none=exclude_none, include=fields)  # type: ignore
            for row in rows
        ]

    as_tuple = kwargs.pop("__as_tuple__", False)

    if not as_tuple:
        return rows

    if not flatten:
        rows = [tuple(row.values()) for row in rows]  # type: ignore
    else:
        try:
            rows = [row[fields[0]] for row in rows]  # type: ignore
        except KeyError:
            raise QuerySetError(detail=f"{fields[0]} does not exist in the results.") from None
    return rows

values_list async 

values_list(fields=None, exclude=None, exclude_none=False, flat=False)

Returns the results in a python dictionary format.

PARAMETER DESCRIPTION
fields

TYPE: Union[Sequence[str], str, None] DEFAULT: None

exclude

TYPE: Union[Sequence[str], Set[str]] DEFAULT: None

exclude_none

TYPE: bool DEFAULT: False

flat

TYPE: bool DEFAULT: False

Source code in saffier/core/db/querysets/base.py 
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
async def values_list(
    self,
    fields: Union[Sequence[str], str, None] = None,
    exclude: Union[Sequence[str], Set[str]] = None,
    exclude_none: bool = False,
    flat: bool = False,
) -> List[Any]:
    """
    Returns the results in a python dictionary format.
    """
    fields = fields or []
    if flat and len(fields) > 1:
        raise QuerySetError(
            detail=f"Maximum of 1 in fields when `flat` is enables, got {len(fields)} instead."
        ) from None

    if flat and isinstance(fields, str):
        fields = [fields]

    if isinstance(fields, str):
        fields = [fields]

    return await self.values(
        fields=fields,
        exclude=exclude,
        exclude_none=exclude_none,
        flatten=flat,
        __as_tuple__=True,
    )

exists async 

exists(**kwargs)

Returns a boolean indicating if a record exists or not.

PARAMETER DESCRIPTION
**kwargs

TYPE: Any DEFAULT: {}

Source code in saffier/core/db/querysets/base.py 
770
771
772
773
774
775
776
777
778
779
async def exists(self, **kwargs: Any) -> bool:
    """
    Returns a boolean indicating if a record exists or not.
    """
    queryset: "QuerySet" = self._clone()
    expression = queryset._build_select()
    expression = sqlalchemy.exists(expression).select()
    queryset._set_query_expression(expression)
    _exists = await queryset.database.fetch_val(expression)
    return cast("bool", _exists)

count async 

count(**kwargs)

Returns an indicating the total records.

PARAMETER DESCRIPTION
**kwargs

TYPE: Any DEFAULT: {}

Source code in saffier/core/db/querysets/base.py 
781
782
783
784
785
786
787
788
789
790
async def count(self, **kwargs: Any) -> int:
    """
    Returns an indicating the total records.
    """
    queryset: "QuerySet" = self._clone()
    expression = queryset._build_select().alias("subquery_for_count")
    expression = sqlalchemy.func.count().select().select_from(expression)
    queryset._set_query_expression(expression)
    _count = await queryset.database.fetch_val(expression)
    return cast("int", _count)

get_or_none async 

get_or_none(**kwargs)

Fetch one object matching the parameters or returns None.

PARAMETER DESCRIPTION
**kwargs

TYPE: Any DEFAULT: {}

Source code in saffier/core/db/querysets/base.py 
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
async def get_or_none(self, **kwargs: Any) -> Union[SaffierModel, None]:
    """
    Fetch one object matching the parameters or returns None.
    """
    queryset: "QuerySet" = self.filter(**kwargs)
    expression = queryset._build_select().limit(2)
    queryset._set_query_expression(expression)
    rows = await queryset.database.fetch_all(expression)

    if not rows:
        return None
    if len(rows) > 1:
        raise MultipleObjectsReturned()
    return queryset.model_class.from_query_result(
        rows[0],
        select_related=queryset._select_related,
        using_schema=queryset.using_schema,
        exclude_secrets=queryset._exclude_secrets,
    )

_all async 

_all(**kwargs)

Returns the queryset records based on specific filters

PARAMETER DESCRIPTION
**kwargs

TYPE: Any DEFAULT: {}

Source code in saffier/core/db/querysets/base.py 
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
async def _all(self, **kwargs: Any) -> List[SaffierModel]:
    """
    Returns the queryset records based on specific filters
    """
    queryset: "QuerySet" = self._clone()

    if queryset.is_m2m:
        queryset.distinct_on = [queryset.m2m_related]

    if kwargs:
        return await queryset.filter(**kwargs).all()

    expression = queryset._build_select()
    queryset._set_query_expression(expression)

    rows = await queryset.database.fetch_all(expression)

    is_only_fields = True if queryset._only else False
    is_defer_fields = True if queryset._defer else False

    # Attach the raw query to the object
    queryset.model_class.raw_query = queryset.sql

    results = [
        queryset.model_class.from_query_result(
            row,
            select_related=queryset._select_related,
            prefetch_related=queryset._prefetch_related,
            is_only_fields=is_only_fields,
            only_fields=queryset._only,
            is_defer_fields=is_defer_fields,
            using_schema=queryset.using_schema,
            exclude_secrets=queryset._exclude_secrets,
        )
        for row in rows
    ]

    if not queryset.is_m2m:
        return results

    return [getattr(result, queryset.m2m_related) for result in results]

all 

all(**kwargs)

Returns the queryset records based on specific filters

PARAMETER DESCRIPTION
**kwargs

TYPE: Any DEFAULT: {}

Source code in saffier/core/db/querysets/base.py 
854
855
856
857
858
859
860
def all(self, **kwargs: Any) -> "QuerySet":
    """
    Returns the queryset records based on specific filters
    """
    queryset: "QuerySet" = self._clone()
    queryset.extra = kwargs
    return queryset

get async 

get(**kwargs)

Returns a single record based on the given kwargs.

PARAMETER DESCRIPTION
**kwargs

TYPE: Any DEFAULT: {}

Source code in saffier/core/db/querysets/base.py 
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
async def get(self, **kwargs: Any) -> SaffierModel:
    """
    Returns a single record based on the given kwargs.
    """
    queryset: "QuerySet" = self._clone()

    if kwargs:
        return await queryset.filter(**kwargs).get()

    expression = queryset._build_select().limit(2)
    rows = await queryset.database.fetch_all(expression)
    queryset._set_query_expression(expression)

    is_only_fields = True if queryset._only else False
    is_defer_fields = True if queryset._defer else False

    if not rows:
        raise ObjectNotFound()
    if len(rows) > 1:
        raise MultipleObjectsReturned()
    return queryset.model_class.from_query_result(
        rows[0],
        select_related=queryset._select_related,
        is_only_fields=is_only_fields,
        only_fields=queryset._only,
        is_defer_fields=is_defer_fields,
        prefetch_related=queryset._prefetch_related,
        using_schema=queryset.using_schema,
        exclude_secrets=queryset._exclude_secrets,
    )

first async 

first(**kwargs)

Returns the first record of a given queryset.

PARAMETER DESCRIPTION
**kwargs

TYPE: Any DEFAULT: {}

Source code in saffier/core/db/querysets/base.py 
893
894
895
896
897
898
899
900
901
902
903
904
async def first(self, **kwargs: Any) -> Union[SaffierModel, None]:
    """
    Returns the first record of a given queryset.
    """
    queryset: "QuerySet" = self._clone()
    if kwargs:
        return await queryset.filter(**kwargs).order_by("id").get()

    rows = await queryset.limit(1).order_by("id").all()
    if rows:
        return rows[0]
    return None

last async 

last(**kwargs)

Returns the last record of a given queryset.

PARAMETER DESCRIPTION
**kwargs

TYPE: Any DEFAULT: {}

Source code in saffier/core/db/querysets/base.py 
906
907
908
909
910
911
912
913
914
915
916
917
async def last(self, **kwargs: Any) -> Union[SaffierModel, None]:
    """
    Returns the last record of a given queryset.
    """
    queryset: "QuerySet" = self._clone()
    if kwargs:
        return await queryset.filter(**kwargs).order_by("-id").get()

    rows = await queryset.order_by("-id").all()
    if rows:
        return rows[0]
    return None

create async 

create(**kwargs)

Creates a record in a specific table.

PARAMETER DESCRIPTION
**kwargs

TYPE: Any DEFAULT: {}

Source code in saffier/core/db/querysets/base.py 
919
920
921
922
923
924
925
926
927
928
async def create(self, **kwargs: Any) -> SaffierModel:
    """
    Creates a record in a specific table.
    """
    queryset: "QuerySet" = self._clone()
    kwargs = queryset._validate_kwargs(**kwargs)
    instance = queryset.model_class(**kwargs)
    instance.table = queryset.table
    instance = await instance.save(force_save=True, values=kwargs)
    return instance

bulk_create async 

bulk_create(objs)

Bulk creates records in a table

PARAMETER DESCRIPTION
objs

TYPE: List[Dict]

Source code in saffier/core/db/querysets/base.py 
930
931
932
933
934
935
936
937
938
939
async def bulk_create(self, objs: List[Dict]) -> None:
    """
    Bulk creates records in a table
    """
    queryset: "QuerySet" = self._clone()
    new_objs = [queryset._validate_kwargs(**obj) for obj in objs]

    expression = queryset.table.insert().values(new_objs)
    queryset._set_query_expression(expression)
    await queryset.database.execute(expression)

bulk_update async 

bulk_update(objs, fields)

Bulk updates records in a table.

A similar solution was suggested here: https://github.com/encode/orm/pull/148

It is thought to be a clean approach to a simple problem so it was added here and refactored to be compatible with Saffier.

PARAMETER DESCRIPTION
objs

TYPE: List[SaffierModel]

fields

TYPE: List[str]

Source code in saffier/core/db/querysets/base.py 
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
async def bulk_update(self, objs: List[SaffierModel], fields: List[str]) -> None:
    """
    Bulk updates records in a table.

    A similar solution was suggested here: https://github.com/encode/orm/pull/148

    It is thought to be a clean approach to a simple problem so it was added here and
    refactored to be compatible with Saffier.
    """
    queryset: "QuerySet" = self._clone()

    new_fields = {}
    for key, field in queryset.model_class.fields.items():
        if key in fields:
            new_fields[key] = field.validator

    validator = Schema(fields=new_fields)

    new_objs = []
    for obj in objs:
        new_obj = {}
        for key, value in obj.__dict__.items():
            if key in fields:
                new_obj[key] = value
        new_objs.append(new_obj)

    new_objs = [
        queryset._update_auto_now_fields(validator.check(obj), queryset.model_class.fields)
        for obj in new_objs
    ]

    pk = getattr(queryset.table.c, queryset.pkname)
    expression = queryset.table.update().where(
        pk == sqlalchemy.bindparam("__id" if queryset.pkname == "id" else queryset.pkname)
    )
    kwargs: Dict[str, Any] = {
        field: sqlalchemy.bindparam(field) for obj in new_objs for field in obj.keys()
    }
    pks = [
        {"__id" if queryset.pkname == "id" else queryset.pkname: getattr(obj, queryset.pkname)}
        for obj in objs
    ]

    query_list = []
    for pk, value in zip(pks, new_objs):  # noqa
        query_list.append({**pk, **value})

    expression = expression.values(kwargs)
    queryset._set_query_expression(expression)
    await queryset.database.execute(expression, query_list)

delete async 

delete()
Source code in saffier/core/db/querysets/base.py 
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
async def delete(self) -> None:
    queryset: "QuerySet" = self._clone()
    await self.model_class.signals.pre_delete.send(sender=self.__class__, instance=self)

    expression = queryset.table.delete()
    for filter_clause in queryset.filter_clauses:
        expression = expression.where(filter_clause)

    queryset._set_query_expression(expression)
    await queryset.database.execute(expression)

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

update async 

update(**kwargs)

Updates a record in a specific table with the given kwargs.

PARAMETER DESCRIPTION
**kwargs

TYPE: Any DEFAULT: {}

Source code in saffier/core/db/querysets/base.py 
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
async def update(self, **kwargs: Any) -> None:
    """
    Updates a record in a specific table with the given kwargs.
    """
    queryset: "QuerySet" = self._clone()
    fields = {
        key: field.validator
        for key, field in queryset.model_class.fields.items()
        if key in kwargs
    }

    validator = Schema(fields=fields)
    kwargs = queryset._update_auto_now_fields(
        validator.check(kwargs), queryset.model_class.fields
    )

    await self.model_class.signals.pre_update.send(
        sender=self.__class__, instance=self, kwargs=kwargs
    )

    expression = queryset.table.update().values(**kwargs)

    for filter_clause in queryset.filter_clauses:
        expression = expression.where(filter_clause)

    queryset._set_query_expression(expression)
    await queryset.database.execute(expression)

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

get_or_create async 

get_or_create(defaults, **kwargs)

Creates a record in a specific table or updates if already exists.

PARAMETER DESCRIPTION
defaults

TYPE: Dict[str, Any]

**kwargs

TYPE: Any DEFAULT: {}

Source code in saffier/core/db/querysets/base.py 
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
async def get_or_create(
    self, defaults: Dict[str, Any], **kwargs: Any
) -> Tuple[SaffierModel, bool]:
    """
    Creates a record in a specific table or updates if already exists.
    """
    queryset: "QuerySet" = self._clone()
    try:
        instance = await queryset.get(**kwargs)
        return instance, False
    except ObjectNotFound:
        kwargs.update(defaults)
        instance = await queryset.create(**kwargs)
        return instance, True

update_or_create async 

update_or_create(defaults, **kwargs)

Updates a record in a specific table or creates a new one.

PARAMETER DESCRIPTION
defaults

TYPE: Dict[str, Any]

**kwargs

TYPE: Any DEFAULT: {}

Source code in saffier/core/db/querysets/base.py 
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
async def update_or_create(
    self, defaults: Dict[str, Any], **kwargs: Any
) -> Tuple[SaffierModel, bool]:
    """
    Updates a record in a specific table or creates a new one.
    """
    queryset: "QuerySet" = self._clone()
    try:
        instance = await queryset.get(**kwargs)
        await instance.update(**defaults)
        return instance, False
    except ObjectNotFound:
        kwargs.update(defaults)
        instance = await queryset.create(**kwargs)
        return instance, True

contains async 

contains(instance)

Returns true if the QuerySet contains the provided object. False if otherwise.

PARAMETER DESCRIPTION
instance

TYPE: SaffierModel

Source code in saffier/core/db/querysets/base.py 
1066
1067
1068
1069
1070
1071
1072
1073
async def contains(self, instance: SaffierModel) -> bool:
    """Returns true if the QuerySet contains the provided object.
    False if otherwise.
    """
    queryset: "QuerySet" = self._clone()
    if getattr(instance, "pk", None) is None:
        raise ValueError("'obj' must be a model or reflect model instance.")
    return await queryset.filter(pk=instance.pk).exists()

_execute async 

_execute()
Source code in saffier/core/db/querysets/base.py 
1075
1076
1077
1078
async def _execute(self) -> Any:
    queryset: "QuerySet" = self._clone()
    records = await queryset._all(**queryset.extra)
    return records