Skip to content

QuerySet class

edgy.QuerySet 

QuerySet(model_class=None, *, database=None, filter_clauses=_empty_set, select_related=_empty_set, prefetch_related=_empty_set, limit_count=None, limit=None, limit_offset=None, offset=None, batch_size=None, order_by=_empty_set, group_by=_empty_set, distinct_on=None, distinct=None, only_fields=None, only=_empty_set, defer_fields=None, defer=_empty_set, embed_parent=None, embed_parent_filters=None, using_schema=Undefined, table=None, exclude_secrets=False, extra_select=None, reference_select=None)

Bases: BaseQuerySet

QuerySet object used for query retrieving. Public interface

Source code in edgy/core/db/querysets/base.py 
 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
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
def __init__(
    self,
    model_class: Union[type[BaseModelType], None] = None,
    *,
    database: Union[Database, None] = None,
    filter_clauses: Iterable[Any] = _empty_set,
    select_related: Iterable[str] = _empty_set,
    prefetch_related: Iterable[Prefetch] = _empty_set,
    limit_count: Optional[int] = None,
    limit: Optional[int] = None,
    limit_offset: Optional[int] = None,
    offset: Optional[int] = None,
    batch_size: Optional[int] = None,
    order_by: Iterable[str] = _empty_set,
    group_by: Iterable[str] = _empty_set,
    distinct_on: Union[None, Literal[True], Iterable[str]] = None,
    distinct: Union[None, Literal[True], Iterable[str]] = None,
    only_fields: Optional[Iterable[str]] = None,
    only: Iterable[str] = _empty_set,
    defer_fields: Optional[Sequence[str]] = None,
    defer: Iterable[str] = _empty_set,
    embed_parent: Optional[tuple[str, Union[str, str]]] = None,
    embed_parent_filters: Optional[tuple[str, str]] = None,
    using_schema: Union[str, None, Any] = Undefined,
    table: Optional[sqlalchemy.Table] = None,
    exclude_secrets: bool = False,
    extra_select: Optional[Iterable[sqlalchemy.expression.ClauseElement]] = None,
    reference_select: Optional[reference_select_type] = None,
) -> None:
    # Making sure for queries we use the main class and not the proxy
    # And enable the parent
    if model_class.__is_proxy_model__:
        model_class = model_class.__parent__

    super().__init__(model_class=model_class)
    self.filter_clauses: list[Any] = list(filter_clauses)
    self.or_clauses: list[Any] = []
    self._aliases: dict[str, sqlalchemy.Alias] = {}
    if limit_count is not None:
        warnings.warn(
            "`limit_count` is deprecated use `limit`", DeprecationWarning, stacklevel=2
        )
        limit = limit_count
    self.limit_count = limit
    if limit_offset is not None:
        warnings.warn(
            "`limit_offset` is deprecated use `limit`", DeprecationWarning, stacklevel=2
        )
        offset = limit_offset
    self._offset = offset

    self._select_related = set(select_related)
    self._prefetch_related = list(prefetch_related)
    self._batch_size = batch_size
    self._order_by: tuple[str, ...] = tuple(order_by)
    self._group_by: tuple[str, ...] = tuple(group_by)
    if distinct_on is not None:
        warnings.warn(
            "`distinct_on` is deprecated use `distinct`", DeprecationWarning, stacklevel=2
        )
        distinct = distinct_on

    if distinct is True:
        distinct = _empty_set
    self.distinct_on = list(distinct) if distinct is not None else None
    if only_fields is not None:
        warnings.warn(
            "`only_fields` is deprecated use `only`", DeprecationWarning, stacklevel=2
        )
        only = only_fields
    self._only = set(only)
    if defer_fields is not None:
        warnings.warn(
            "`defer_fields` is deprecated use `defer`", DeprecationWarning, stacklevel=2
        )
        defer = defer_fields
    self._defer = set(defer)
    self.embed_parent = embed_parent
    self.embed_parent_filters = embed_parent_filters
    self.using_schema = using_schema
    self._extra_select = list(extra_select) if extra_select is not None else []
    self._reference_select = (
        reference_select.copy() if isinstance(reference_select, dict) else {}
    )
    self._exclude_secrets = exclude_secrets
    # cache should not be cloned
    self._cache = QueryModelResultCache(attrs=self.model_class.pkcolumns)
    # is empty
    self._clear_cache(False)
    # this is not cleared, because the expression is immutable
    self._cached_select_related_expression: Optional[
        tuple[
            Any,
            dict[str, tuple[sqlalchemy.Table, type[BaseModelType]]],
        ]
    ] = None
    # initialize
    self.active_schema = self.get_schema()

    # Making sure the queryset always starts without any schema associated unless specified

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

model_class instance-attribute 

model_class = model_class

_table class-attribute instance-attribute 

_table = None

_database class-attribute instance-attribute 

_database = None

database instance-attribute 

database = database

table instance-attribute 

table = table

pkcolumns property 

pkcolumns

filter_clauses instance-attribute 

filter_clauses = list(filter_clauses)

or_clauses instance-attribute 

or_clauses = []

_aliases instance-attribute 

_aliases = {}

limit_count instance-attribute 

limit_count = limit

_offset instance-attribute 

_offset = offset

_select_related = set(select_related)

_prefetch_related = list(prefetch_related)

_batch_size instance-attribute 

_batch_size = batch_size

_order_by instance-attribute 

_order_by = tuple(order_by)

_group_by instance-attribute 

_group_by = tuple(group_by)

distinct_on instance-attribute 

distinct_on = list(distinct) if distinct is not None else None

_only instance-attribute 

_only = set(only)

_defer instance-attribute 

_defer = set(defer)

embed_parent instance-attribute 

embed_parent = embed_parent

embed_parent_filters instance-attribute 

embed_parent_filters = embed_parent_filters

using_schema instance-attribute 

using_schema = using_schema

_extra_select instance-attribute 

_extra_select = list(extra_select) if extra_select is not None else []

_reference_select instance-attribute 

_reference_select = copy() if isinstance(reference_select, dict) else {}

_exclude_secrets instance-attribute 

_exclude_secrets = exclude_secrets

_cache instance-attribute 

_cache = QueryModelResultCache(attrs=pkcolumns)

_cached_select_related_expression = None

active_schema instance-attribute 

active_schema = get_schema()

sql cached property 

sql

Get SQL select query as string with inserted blanks. For debugging only!

build_where_clause async 

build_where_clause(_=None, tables_and_models=None)

Build a where clause from the filters which can be passed in a where function.

Source code in edgy/core/db/querysets/base.py 
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
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
async def build_where_clause(
    self, _: Any = None, tables_and_models: Optional[tables_and_models_type] = None
) -> Any:
    """Build a where clause from the filters which can be passed in a where function."""
    joins: Optional[Any] = None
    if tables_and_models is None:
        joins, tables_and_models = self._build_tables_join_from_relationship()
    # ignored args for passing build_where_clause in filter_clauses
    where_clauses: list[Any] = []
    if self.or_clauses:
        where_clauses.append(
            await clauses_mod.parse_clause_arg(
                clauses_mod.or_(*self.or_clauses, no_select_related=True),
                self,
                tables_and_models,
            )
        )

    if self.filter_clauses:
        # we AND by default
        where_clauses.extend(
            await clauses_mod.parse_clause_args(self.filter_clauses, self, tables_and_models)
        )
    # for nicer unpacking
    if joins is None or len(tables_and_models) == 1:
        return clauses_mod.and_sqlalchemy(*where_clauses)
    expression = sqlalchemy.sql.select(
        *(
            getattr(tables_and_models[""][0].c, col)
            for col in tables_and_models[""][1].pkcolumns
        ),
    ).set_label_style(sqlalchemy.LABEL_STYLE_NONE)
    idtuple = sqlalchemy.tuple_(
        *(
            getattr(tables_and_models[""][0].c, col)
            for col in tables_and_models[""][1].pkcolumns
        )
    )
    expression = expression.select_from(joins)
    return idtuple.in_(
        expression.where(
            *where_clauses,
        )
    )

using 

using(_positional=_sentinel, *, database=Undefined, schema=Undefined)

Enables and switches the db schema and/or db.

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

Use schema=False to unset the schema to the active default schema. Use database=None to use the default database again.

Source code in edgy/core/db/querysets/mixins.py 
 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
def using(
    self,
    _positional: Any = _sentinel,
    *,
    database: Union[str, Any, None, "Database"] = Undefined,
    schema: Union[str, Any, None, Literal[False]] = Undefined,
) -> "QuerySet":
    """
    Enables and switches the db schema and/or db.

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

    Use schema=False to unset the schema to the active default schema.
    Use database=None to use the default database again.
    """
    if _positional is not _sentinel:
        warnings.warn(
            "Passing positional arguments to using is deprecated. Use schema= instead.",
            DeprecationWarning,
            stacklevel=2,
        )
        schema = _positional
        if schema is Undefined:
            schema = False

    queryset = cast("QuerySet", self._clone())
    if database is not Undefined:
        if isinstance(database, Database):
            connection: Database = database
        elif database is None:
            connection = self.model_class.meta.registry.database
        else:
            assert (
                database is None or database in self.model_class.meta.registry.extra
            ), f"`{database}` is not in the connections extra of the model`{self.model_class.__name__}` registry"

            connection = self.model_class.meta.registry.extra[database]
        queryset.database = connection
    if schema is not Undefined:
        queryset.using_schema = schema if schema is not False else Undefined
        new_schema = queryset.get_schema()
        if new_schema != queryset.active_schema:
            queryset.active_schema = new_schema
            queryset.table = None

    return queryset

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.

Source code in edgy/core/db/querysets/prefetch.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
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 any(not isinstance(value, Prefetch) for value in prefetch):
        raise QuerySetError("The prefetch_related must have Prefetch type objects only.")

    queryset._prefetch_related = [*self._prefetch_related, *prefetch]
    return queryset

using_with_db 

using_with_db(connection_name, schema=Undefined)

Switches the database connection and schema

Source code in edgy/core/db/querysets/mixins.py 
109
110
111
112
113
114
115
116
117
118
119
120
def using_with_db(
    self, connection_name: str, schema: Union[str, Any, None, Literal[False]] = Undefined
) -> "QuerySet":
    """
    Switches the database connection and schema
    """
    warnings.warn(
        "'using_with_db' is deprecated in favor of 'using' with schema, database arguments.",
        DeprecationWarning,
        stacklevel=2,
    )
    return self.using(database=connection_name, schema=schema)

_clone 

_clone()

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

Source code in edgy/core/db/querysets/base.py 
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
def _clone(self) -> QuerySet:
    """
    Return a copy of the current QuerySet that's ready for another
    operation.
    """
    queryset = self.__class__(
        self.model_class,
        database=getattr(self, "_database", None),
        filter_clauses=self.filter_clauses,
        select_related=self._select_related,
        prefetch_related=self._prefetch_related,
        limit=self.limit_count,
        offset=self._offset,
        batch_size=self._batch_size,
        order_by=self._order_by,
        group_by=self._group_by,
        distinct=self.distinct_on,
        only=self._only,
        defer=self._defer,
        embed_parent=self.embed_parent,
        embed_parent_filters=self.embed_parent_filters,
        using_schema=self.using_schema,
        table=getattr(self, "_table", None),
        exclude_secrets=self._exclude_secrets,
        reference_select=self._reference_select,
        extra_select=self._extra_select,
    )
    queryset.or_clauses.extend(self.or_clauses)
    queryset._cached_select_related_expression = self._cached_select_related_expression
    return cast("QuerySet", queryset)

_clear_cache 

_clear_cache(keep_result_cache=False)
Source code in edgy/core/db/querysets/base.py 
207
208
209
210
211
212
213
214
215
216
217
218
219
220
def _clear_cache(self, keep_result_cache: bool = False) -> None:
    if not keep_result_cache:
        self._cache.clear()
    self._cached_select_with_tables: Optional[
        tuple[Any, dict[str, tuple[sqlalchemy.Table, type[BaseModelType]]]]
    ] = None
    self._cache_count: Optional[int] = None
    self._cache_first: Optional[tuple[BaseModelType, BaseModelType]] = None
    self._cache_last: Optional[tuple[BaseModelType, BaseModelType]] = None
    # fetch all is in cache
    self._cache_fetch_all: bool = False
    # get current row during iteration. Used for prefetching.
    # Bad style but no other way currently possible
    self._cache_current_row: Optional[sqlalchemy.Row] = None

_build_order_by_expression 

_build_order_by_expression(order_by, expression)

Builds the order by expression

Source code in edgy/core/db/querysets/base.py 
222
223
224
225
def _build_order_by_expression(self, order_by: Any, expression: Any) -> Any:
    """Builds the order by expression"""
    expression = expression.order_by(*(self._prepare_order_by(entry) for entry in order_by))
    return expression

_build_group_by_expression 

_build_group_by_expression(group_by, expression)

Builds the group by expression

Source code in edgy/core/db/querysets/base.py 
227
228
229
230
def _build_group_by_expression(self, group_by: Any, expression: Any) -> Any:
    """Builds the group by expression"""
    expression = expression.group_by(*(self._prepare_order_by(entry) for entry in group_by))
    return expression

_build_select_distinct 

_build_select_distinct(distinct_on, expression)

Filters selects only specific fields. Leave empty to use simple distinct

Source code in edgy/core/db/querysets/base.py 
277
278
279
280
281
282
283
def _build_select_distinct(self, distinct_on: Optional[Sequence[str]], expression: Any) -> Any:
    """Filters selects only specific fields. Leave empty to use simple distinct"""
    # using with columns is not supported by all databases
    if distinct_on:
        return expression.distinct(*map(self._prepare_fields_for_distinct, distinct_on))
    else:
        return expression.distinct()

_join_table_helper classmethod 

_join_table_helper(join_clause, current_transition, *, transitions, tables_and_models)
Source code in edgy/core/db/querysets/base.py 
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
@classmethod
def _join_table_helper(
    cls,
    join_clause: Any,
    current_transition: tuple[str, str, str],
    *,
    transitions: dict[tuple[str, str, str], tuple[Any, Optional[tuple[str, str, str]], str]],
    tables_and_models: dict[str, tuple[sqlalchemy.Table, type[BaseModelType]]],
) -> Any:
    if current_transition not in transitions:
        return join_clause
    transition_value = transitions.pop(current_transition)

    if transition_value[1] is not None:
        join_clause = cls._join_table_helper(
            join_clause,
            transition_value[1],
            transitions=transitions,
            tables_and_models=tables_and_models,
        )

    return sqlalchemy.sql.join(
        join_clause,
        tables_and_models[transition_value[2]][0],
        transition_value[0],
        isouter=True,
    )

_build_tables_join_from_relationship 

_build_tables_join_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 edgy/core/db/querysets/base.py 
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
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
def _build_tables_join_from_relationship(
    self,
) -> tuple[Any, tables_and_models_type]:
    """
    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.
    """

    # How does this work?
    # First we build a transitions tree (maintable is root) by providing optionally a dependency.
    # Resolving a dependency automatically let's us resolve the tree.
    # At last we iter through the transisitions and build their dependencies first
    # We pop out the transitions so a path is not taken 2 times

    # Why left outer join? It is possible and legal for a relation to not exist we check that already in filtering.

    if self._cached_select_related_expression is None:
        maintable = self.table
        select_from = maintable
        tables_and_models: tables_and_models_type = {"": (select_from, self.model_class)}
        _select_tables_and_models: tables_and_models_type = {
            "": (select_from, self.model_class)
        }
        transitions: dict[
            tuple[str, str, str], tuple[Any, Optional[tuple[str, str, str]], str]
        ] = {}

        # Select related
        for select_path in self._select_related:
            # For m2m relationships
            model_class = self.model_class
            former_table = maintable
            former_transition = None
            prefix: str = ""
            # prefix which expands m2m fields
            _select_prefix: str = ""
            # False (default): add prefix regularly.
            # True: skip adding existing prefix to tables_and_models and skip adding the next field to the
            #       public prefix.
            # string: add a custom prefix instead of the calculated one and skip adding the next field to the
            #         public prefix.

            injected_prefix: Union[bool, str] = False
            model_database: Optional[Database] = self.database
            while select_path:
                field_name = select_path.split("__", 1)[0]
                try:
                    field = model_class.meta.fields[field_name]
                except KeyError:
                    raise QuerySetError(
                        detail=f'Selected field "{field_name}" does not exist on {model_class}.'
                    ) from None
                field = model_class.meta.fields[field_name]
                if isinstance(field, RelationshipField):
                    model_class, reverse_part, select_path = field.traverse_field(select_path)
                else:
                    raise QuerySetError(
                        detail=f'Selected field "{field_name}" is not a RelationshipField on {model_class}.'
                    )
                if isinstance(field, BaseForeignKey):
                    foreign_key = field
                    reverse = False
                else:
                    foreign_key = model_class.meta.fields[reverse_part]
                    reverse = True
                if foreign_key.is_cross_db(model_database):
                    raise QuerySetError(
                        detail=f'Selected model "{field_name}" is on another database.'
                    )
                # now use the one of the model_class itself
                model_database = None
                if injected_prefix:
                    injected_prefix = False
                else:
                    prefix = f"{prefix}__{field_name}" if prefix else f"{field_name}"
                _select_prefix = (
                    f"{_select_prefix}__{field_name}" if _select_prefix else f"{field_name}"
                )
                if foreign_key.is_m2m and foreign_key.embed_through != "":  # type: ignore
                    # we need to inject the through model for the select
                    model_class = foreign_key.through
                    if foreign_key.embed_through is False:
                        injected_prefix = True
                    else:
                        injected_prefix = f"{prefix}__{foreign_key.embed_through}"
                    if reverse:
                        select_path = f"{foreign_key.from_foreign_key}__{select_path}"
                    else:
                        select_path = f"{foreign_key.to_foreign_key}__{select_path}"
                    # if select_path is empty
                    select_path = select_path.removesuffix("__")
                    if reverse:
                        foreign_key = model_class.meta.fields[foreign_key.to_foreign_key]
                    else:
                        foreign_key = model_class.meta.fields[foreign_key.from_foreign_key]
                        reverse = True
                if _select_prefix in _select_tables_and_models:
                    # use prexisting prefix
                    table: Any = _select_tables_and_models[_select_prefix][0]
                else:
                    table = model_class.table_schema(self.active_schema)
                    table = table.alias(hash_tablekey(tablekey=table.key, prefix=prefix))

                # it is guranteed that former_table is either root and has a key or is an unique join node
                # except there would be a hash collision which is very unlikely
                transition_key = (get_table_key_or_name(former_table), table.name, field_name)
                if transition_key in transitions:
                    # can not provide new informations
                    former_table = table
                    former_transition = transition_key
                    continue
                and_clause = clauses_mod.and_sqlalchemy(
                    *self._select_from_relationship_clause_generator(
                        foreign_key, table, reverse, former_table
                    )
                )
                transitions[transition_key] = (
                    and_clause,
                    former_transition,
                    _select_prefix,
                )
                if injected_prefix is False:
                    tables_and_models[prefix] = table, model_class
                elif injected_prefix is not True:
                    # we inject a string
                    tables_and_models[injected_prefix] = table, model_class

                # prefix used for select_related
                _select_tables_and_models[_select_prefix] = table, model_class
                former_table = table
                former_transition = transition_key

        while transitions:
            select_from = self._join_table_helper(
                select_from,
                next(iter(transitions.keys())),
                transitions=transitions,
                tables_and_models=_select_tables_and_models,
            )
        self._cached_select_related_expression = (
            select_from,
            tables_and_models,
        )
    return self._cached_select_related_expression

_select_from_relationship_clause_generator staticmethod 

_select_from_relationship_clause_generator(foreign_key, table, reverse, former_table)
Source code in edgy/core/db/querysets/base.py 
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
@staticmethod
def _select_from_relationship_clause_generator(
    foreign_key: BaseForeignKey,
    table: Any,
    reverse: bool,
    former_table: Any,
) -> Any:
    column_names = foreign_key.get_column_names(foreign_key.name)
    assert column_names, f"foreign key without column names detected: {foreign_key.name}"
    for col in column_names:
        colname = foreign_key.from_fk_field_name(foreign_key.name, col) if reverse else col
        if reverse:
            yield getattr(former_table.c, colname) == getattr(table.c, col)
        else:
            yield getattr(former_table.c, colname) == getattr(
                table.c, foreign_key.from_fk_field_name(foreign_key.name, col)
            )

_validate_only_and_defer 

_validate_only_and_defer()
Source code in edgy/core/db/querysets/base.py 
478
479
480
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.")

_as_select_with_tables async 

_as_select_with_tables()

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

Source code in edgy/core/db/querysets/base.py 
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
async def _as_select_with_tables(
    self,
) -> tuple[Any, tables_and_models_type]:
    """
    Builds the query select based on the given parameters and filters.
    """
    self._validate_only_and_defer()
    joins, tables_and_models = self._build_tables_join_from_relationship()
    columns_and_extra: list[Any] = [*self._extra_select]
    for prefix, (table, model_class) in tables_and_models.items():
        if not prefix:
            for column_key, column in table.columns.items():
                # e.g. reflection has not always a field
                field_name = model_class.meta.columns_to_field.get(column_key, column_key)
                if self._only and field_name not in self._only:
                    continue
                if self._defer and field_name in self._defer:
                    continue
                if (
                    self._exclude_secrets
                    and field_name in model_class.meta.fields
                    and model_class.meta.fields[field_name].secret
                ):
                    continue
                # add without alias
                columns_and_extra.append(column)

        else:
            for column_key, column in table.columns.items():
                # e.g. reflection has not always a field
                field_name = model_class.meta.columns_to_field.get(column_key, column_key)
                if (
                    self._only
                    and prefix not in self._only
                    and f"{prefix}__{field_name}" not in self._only
                ):
                    continue
                if self._defer and (
                    prefix in self._defer or f"{prefix}__{field_name}" in self._defer
                ):
                    continue
                if (
                    self._exclude_secrets
                    and field_name in model_class.meta.fields
                    and model_class.meta.fields[field_name].secret
                ):
                    continue
                # alias has name not a key. The name is fully descriptive
                columns_and_extra.append(column.label(f"{table.name}_{column_key}"))
    assert columns_and_extra, "no columns or extra_select specified"
    # all columns are aliased already
    expression = sqlalchemy.sql.select(*columns_and_extra).set_label_style(
        sqlalchemy.LABEL_STYLE_NONE
    )
    expression = expression.select_from(joins)
    expression = expression.where(await self.build_where_clause(self, tables_and_models))

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

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

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

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

    if self.distinct_on is not None:
        expression = self._build_select_distinct(self.distinct_on, expression=expression)
    return expression, tables_and_models

as_select_with_tables async 

as_select_with_tables()

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

Source code in edgy/core/db/querysets/base.py 
555
556
557
558
559
560
561
562
563
async def as_select_with_tables(
    self,
) -> tuple[Any, tables_and_models_type]:
    """
    Builds the query select based on the given parameters and filters.
    """
    if self._cached_select_with_tables is None:
        self._cached_select_with_tables = await self._as_select_with_tables()
    return self._cached_select_with_tables

as_select async 

as_select()
Source code in edgy/core/db/querysets/base.py 
565
566
567
568
async def as_select(
    self,
) -> Any:
    return (await self.as_select_with_tables())[0]

_kwargs_to_clauses 

_kwargs_to_clauses(kwargs)
Source code in edgy/core/db/querysets/base.py 
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
def _kwargs_to_clauses(
    self,
    kwargs: Any,
) -> tuple[list[Any], set[str]]:
    clauses = []
    select_related: set[str] = set()
    cleaned_kwargs = clauses_mod.clean_query_kwargs(
        self.model_class, kwargs, self.embed_parent_filters, model_database=self.database
    )

    for key, value in cleaned_kwargs.items():
        model_class, field_name, op, related_str, _, cross_db_remainder = crawl_relationship(
            self.model_class, key
        )
        if related_str:
            select_related.add(related_str)
        field = model_class.meta.fields.get(field_name, clauses_mod.generic_field)
        if cross_db_remainder:
            assert field is not clauses_mod.generic_field
            fk_field = cast(BaseForeignKey, field)
            sub_query = (
                fk_field.target.query.filter(**{cross_db_remainder: value})
                .only(*fk_field.related_columns.keys())
                .values_list(fields=fk_field.related_columns.keys())
            )

            # bind local vars
            async def wrapper(
                queryset: QuerySet,
                tables_and_models: tables_and_models_type,
                *,
                _field: BaseFieldType = field,
                _sub_query: QuerySet = sub_query,
                _prefix: str = related_str,
            ) -> Any:
                table = tables_and_models[_prefix][0]
                fk_tuple = sqlalchemy.tuple_(
                    *(getattr(table.columns, colname) for colname in _field.get_column_names())
                )
                return fk_tuple.in_(await _sub_query)

            clauses.append(wrapper)
        else:
            assert not isinstance(
                value, BaseModelType
            ), f"should be parsed in clean: {key}: {value}"

            async def wrapper(
                queryset: QuerySet,
                tables_and_models: tables_and_models_type,
                *,
                _field: BaseFieldType = field,
                _value: Any = value,
                _op: Optional[str] = op,
                _prefix: str = related_str,
                # generic field has no field name
                _field_name: str = field_name,
            ) -> Any:
                _value = await clauses_mod.parse_clause_arg(
                    _value, queryset, tables_and_models
                )
                table = tables_and_models[_prefix][0]
                return _field.operator_to_clause(_field_name, _op, table, _value)

            wrapper._edgy_force_callable_queryset_filter = True
            clauses.append(wrapper)

    return clauses, select_related

_prepare_order_by 

_prepare_order_by(order_by)
Source code in edgy/core/db/querysets/base.py 
639
640
641
642
643
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_fields_for_distinct 

_prepare_fields_for_distinct(distinct_on)
Source code in edgy/core/db/querysets/base.py 
645
646
def _prepare_fields_for_distinct(self, distinct_on: str) -> sqlalchemy.Column:
    return self.table.columns[distinct_on]

_embed_parent_in_result async 

_embed_parent_in_result(result)
Source code in edgy/core/db/querysets/base.py 
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
async def _embed_parent_in_result(
    self, result: Union[EdgyModel, Awaitable[EdgyModel]]
) -> tuple[EdgyModel, Any]:
    if isawaitable(result):
        result = await result
    if not self.embed_parent:
        return (cast(EdgyModel, result), cast(EdgyModel, result))
    token = MODEL_GETATTR_BEHAVIOR.set("coro")
    try:
        new_result: Any = result
        for part in self.embed_parent[0].split("__"):
            new_result = getattr(new_result, part)
            if isawaitable(new_result):
                new_result = await new_result
    finally:
        MODEL_GETATTR_BEHAVIOR.reset(token)
    if self.embed_parent[1]:
        setattr(new_result, self.embed_parent[1], result)
    return cast(EdgyModel, result), new_result

_get_or_cache_row async 

_get_or_cache_row(row, tables_and_models, extra_attr='', raw=False)
Source code in edgy/core/db/querysets/base.py 
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
async def _get_or_cache_row(
    self,
    row: Any,
    tables_and_models: dict[str, tuple[sqlalchemy.Table, type[BaseModelType]]],
    extra_attr: str = "",
    raw: bool = False,
) -> tuple[EdgyModel, EdgyModel]:
    is_defer_fields = bool(self._defer)
    raw_result, result = (
        await self._cache.aget_or_cache_many(
            self.model_class,
            [row],
            cache_fn=lambda _row: self.model_class.from_sqla_row(
                _row,
                tables_and_models=tables_and_models,
                select_related=self._select_related,
                only_fields=self._only,
                is_defer_fields=is_defer_fields,
                prefetch_related=self._prefetch_related,
                exclude_secrets=self._exclude_secrets,
                using_schema=self.active_schema,
                database=self.database,
                reference_select=self._reference_select,
            ),
            transform_fn=self._embed_parent_in_result,
        )
    )[0]
    if extra_attr:
        for attr in extra_attr.split(","):
            setattr(self, attr, result)
    return cast("EdgyModel", raw_result), cast("EdgyModel", result)

get_schema 

get_schema()
Source code in edgy/core/db/querysets/base.py 
700
701
702
703
704
705
706
707
def get_schema(self) -> Optional[str]:
    # Differs from get_schema global
    schema = self.using_schema
    if schema is Undefined:
        schema = get_schema()
    if schema is None:
        schema = self.model_class.get_db_schema()
    return schema

_handle_batch async 

_handle_batch(batch, tables_and_models, queryset)
Source code in edgy/core/db/querysets/base.py 
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
739
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
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
async def _handle_batch(
    self,
    batch: Sequence[sqlalchemy.Row],
    tables_and_models: dict[str, tuple[sqlalchemy.Table, type[BaseModelType]]],
    queryset: BaseQuerySet,
) -> Sequence[tuple[BaseModelType, BaseModelType]]:
    is_defer_fields = bool(queryset._defer)
    del queryset
    _prefetch_related: list[Prefetch] = []

    for prefetch in self._prefetch_related:
        check_prefetch_collision(self.model_class, prefetch)  # type: ignore

        crawl_result = crawl_relationship(
            self.model_class, prefetch.related_name, traverse_last=True
        )
        if crawl_result.cross_db_remainder:
            raise NotImplementedError(
                "Cannot prefetch from other db yet. Maybe in future this feature will be added."
            )
        if crawl_result.reverse_path is False:
            QuerySetError(
                detail=("Creating a reverse path is not possible, unidirectional fields used.")
            )
        prefetch_queryset: Optional[QuerySet] = prefetch.queryset

        clauses = [
            {
                f"{crawl_result.reverse_path}__{pkcol}": row._mapping[pkcol]
                for pkcol in self.model_class.pkcolumns
            }
            for row in batch
        ]
        if prefetch_queryset is None:
            prefetch_queryset = crawl_result.model_class.query.local_or(*clauses)
        else:
            # ensure local or
            prefetch_queryset = prefetch_queryset.local_or(*clauses)

        if prefetch_queryset.model_class is self.model_class:
            # queryset is of this model
            prefetch_queryset = prefetch_queryset.select_related(prefetch.related_name)
            prefetch_queryset.embed_parent = (prefetch.related_name, "")
        else:
            # queryset is of the target model
            prefetch_queryset = prefetch_queryset.select_related(
                cast(str, crawl_result.reverse_path)
            )
        new_prefetch = Prefetch(
            related_name=prefetch.related_name,
            to_attr=prefetch.to_attr,
            queryset=prefetch_queryset,
        )
        new_prefetch._bake_prefix = f"{hash_tablekey(tablekey=tables_and_models[''][0].key, prefix=cast(str, crawl_result.reverse_path))}_"
        new_prefetch._is_finished = True
        _prefetch_related.append(new_prefetch)

    return cast(
        Sequence[tuple[BaseModelType, BaseModelType]],
        await self._cache.aget_or_cache_many(
            self.model_class,
            batch,
            cache_fn=lambda row: self.model_class.from_sqla_row(
                row,
                tables_and_models=tables_and_models,
                select_related=self._select_related,
                only_fields=self._only,
                is_defer_fields=is_defer_fields,
                prefetch_related=_prefetch_related,
                exclude_secrets=self._exclude_secrets,
                using_schema=self.active_schema,
                database=self.database,
                reference_select=self._reference_select,
            ),
            transform_fn=self._embed_parent_in_result,
        ),
    )

_execute_iterate async 

_execute_iterate(fetch_all_at_once=False)

Executes the query, iterate.

Source code in edgy/core/db/querysets/base.py 
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
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
853
854
855
856
857
async def _execute_iterate(
    self, fetch_all_at_once: bool = False
) -> AsyncIterator[BaseModelType]:
    """
    Executes the query, iterate.
    """
    if self._cache_fetch_all:
        for result in cast(
            Sequence[tuple[BaseModelType, BaseModelType]],
            self._cache.get_category(self.model_class).values(),
        ):
            yield result[1]
    queryset = self
    if queryset.embed_parent:
        # activates distinct, not distinct on
        queryset = queryset.distinct()  # type: ignore

    expression, tables_and_models = await queryset.as_select_with_tables()

    if not fetch_all_at_once and bool(queryset.database.force_rollback):
        # force_rollback on db = we have only one connection
        # so every operation must be atomic
        # Note: force_rollback is a bit magic, it evaluates its truthiness to the actual value
        warnings.warn(
            'Using queryset iterations with "Database"-level force_rollback set is risky. '
            "Deadlocks can occur because only one connection is used.",
            UserWarning,
            stacklevel=3,
        )
        if queryset._prefetch_related:
            # prefetching will certainly deadlock, let's mitigate
            fetch_all_at_once = True

    counter = 0
    last_element: Optional[tuple[BaseModelType, BaseModelType]] = None
    check_db_connection(queryset.database, stacklevel=4)
    if fetch_all_at_once:
        async with queryset.database as database:
            batch = cast(Sequence[sqlalchemy.Row], await database.fetch_all(expression))
        for row_num, result in enumerate(
            await self._handle_batch(batch, tables_and_models, queryset)
        ):
            if counter == 0:
                self._cache_first = result
            last_element = result
            counter += 1
            self._cache_current_row = batch[row_num]
            yield result[1]
        self._cache_current_row = None
        self._cache_fetch_all = True
    else:
        async with queryset.database as database:
            async for batch in database.batched_iterate(
                expression, batch_size=self._batch_size
            ):
                # clear only result cache
                self._cache.clear()
                self._cache_fetch_all = False
                for row_num, result in enumerate(
                    await self._handle_batch(batch, tables_and_models, queryset)
                ):
                    if counter == 0:
                        self._cache_first = result
                    last_element = result
                    counter += 1
                    self._cache_current_row = batch[row_num]  # type: ignore
                    yield result[1]
            self._cache_current_row = None
    # better update them once
    self._cache_count = counter
    self._cache_last = last_element

_execute_all async 

_execute_all()
Source code in edgy/core/db/querysets/base.py 
859
860
async def _execute_all(self) -> list[EdgyModel]:
    return [result async for result in self._execute_iterate(fetch_all_at_once=True)]

_filter_or_exclude 

_filter_or_exclude(kwargs, clauses, exclude=False, or_=False, allow_global_or=True)

Filters or excludes a given clause for a specific QuerySet.

Source code in edgy/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
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
def _filter_or_exclude(
    self,
    kwargs: Any,
    clauses: Sequence[
        Union[
            sqlalchemy.sql.expression.BinaryExpression,
            Callable[
                [QuerySetType],
                Union[
                    sqlalchemy.sql.expression.BinaryExpression,
                    Awaitable[sqlalchemy.sql.expression.BinaryExpression],
                ],
            ],
            dict[str, Any],
            QuerySet,
        ]
    ],
    exclude: bool = False,
    or_: bool = False,
    allow_global_or: bool = True,
) -> QuerySet:
    """
    Filters or excludes a given clause for a specific QuerySet.
    """
    queryset: QuerySet = self._clone()
    if kwargs:
        clauses = [*clauses, kwargs]
    converted_clauses: Sequence[
        Union[
            sqlalchemy.sql.expression.BinaryExpression,
            Callable[
                [QuerySetType],
                Union[
                    sqlalchemy.sql.expression.BinaryExpression,
                    Awaitable[sqlalchemy.sql.expression.BinaryExpression],
                ],
            ],
        ]
    ] = []
    for raw_clause in clauses:
        if isinstance(raw_clause, dict):
            extracted_clauses, related = queryset._kwargs_to_clauses(kwargs=raw_clause)
            if not queryset._select_related.issuperset(related):
                queryset._select_related.update(related)
                queryset._cached_select_related_expression = None
            if or_ and extracted_clauses:
                wrapper_and = clauses_mod.and_(*extracted_clauses, no_select_related=True)
                if allow_global_or and len(clauses) == 1:
                    # add to global or
                    assert not exclude
                    queryset.or_clauses.append(wrapper_and)
                    return queryset
                converted_clauses.append(wrapper_and)
            else:
                converted_clauses.extend(extracted_clauses)
        elif isinstance(raw_clause, QuerySet):
            assert (
                raw_clause.model_class is queryset.model_class
            ), f"QuerySet arg has wrong model_class {raw_clause.model_class}"
            converted_clauses.append(raw_clause.build_where_clause)
            if not queryset._select_related.issuperset(raw_clause._select_related):
                queryset._select_related.update(raw_clause._select_related)
                queryset._cached_select_related_expression = None
        else:
            converted_clauses.append(raw_clause)
            if hasattr(raw_clause, "_edgy_calculate_select_related"):
                select_related_calculated = raw_clause._edgy_calculate_select_related(queryset)
                if not queryset._select_related.issuperset(select_related_calculated):
                    queryset._select_related.update(select_related_calculated)
                    queryset._cached_select_related_expression = None
    if not converted_clauses:
        return queryset

    if exclude:
        op = clauses_mod.and_ if not or_ else clauses_mod.or_

        queryset.filter_clauses.append(
            clauses_mod.not_(
                op(*converted_clauses, no_select_related=True), no_select_related=True
            )
        )
    elif or_:
        queryset.filter_clauses.append(
            clauses_mod.or_(*converted_clauses, no_select_related=True)
        )
    else:
        # default to and
        queryset.filter_clauses.extend(converted_clauses)
    return queryset

_model_based_delete async 

_model_based_delete()
Source code in edgy/core/db/querysets/base.py 
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
async def _model_based_delete(self) -> int:
    queryset = self.limit(self._batch_size)
    # we set embed_parent on the copy to None to get raw instances
    # embed_parent_filters is not affected
    queryset.embed_parent = None
    counter = 0
    models = await queryset
    while models:
        for model in models:
            counter += 1
            # delete issues already signals
            await model.delete()
        # clear cache and fetch new batch
        models = await queryset.all(True)
    self._clear_cache()
    return counter

_get_raw async 

_get_raw(**kwargs)

Returns a single record based on the given kwargs.

Source code in edgy/core/db/querysets/base.py 
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
async def _get_raw(self, **kwargs: Any) -> tuple[BaseModelType, Any]:
    """
    Returns a single record based on the given kwargs.
    """

    if kwargs:
        cached = cast(
            Optional[tuple[BaseModelType, Any]], self._cache.get(self.model_class, kwargs)
        )
        if cached is not None:
            return cached
        filter_query = cast("BaseQuerySet", self.filter(**kwargs))
        # connect parent query cache
        filter_query._cache = self._cache
        return await filter_query._get_raw()

    expression, tables_and_models = await self.as_select_with_tables()
    check_db_connection(self.database, stacklevel=4)
    async with self.database as database:
        # we want no queryset copy, so use sqlalchemy limit(2)
        rows = await database.fetch_all(expression.limit(2))

    if not rows:
        self._cache_count = 0
        raise ObjectNotFound()
    if len(rows) > 1:
        raise MultipleObjectsReturned()
    self._cache_count = 1

    return await self._get_or_cache_row(rows[0], tables_and_models, "_cache_first,_cache_last")

__repr__ 

__repr__()
Source code in edgy/core/db/querysets/base.py 
1000
1001
def __repr__(self) -> str:
    return f"QuerySet<{self.sql}>"

_sql_helper async 

_sql_helper()
Source code in edgy/core/db/querysets/base.py 
1009
1010
1011
1012
1013
async def _sql_helper(self) -> Any:
    async with self.database:
        return (await self.as_select()).compile(
            self.database.engine, compile_kwargs={"literal_binds": True}
        )

filter 

filter(*clauses, **kwargs)

Filters the QuerySet by the given kwargs and clauses.

Source code in edgy/core/db/querysets/base.py 
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
def filter(
    self,
    *clauses: Union[
        sqlalchemy.sql.expression.BinaryExpression,
        Callable[
            [QuerySetType],
            Union[
                sqlalchemy.sql.expression.BinaryExpression,
                Awaitable[sqlalchemy.sql.expression.BinaryExpression],
            ],
        ],
        dict[str, Any],
        QuerySet,
    ],
    **kwargs: Any,
) -> QuerySet:
    """
    Filters the QuerySet by the given kwargs and clauses.
    """
    return self._filter_or_exclude(clauses=clauses, kwargs=kwargs)

all 

all(clear_cache=False)

Returns a cloned query with empty cache. Optionally just clear the cache and return the same query.

Source code in edgy/core/db/querysets/base.py 
1041
1042
1043
1044
1045
1046
1047
1048
def all(self, clear_cache: bool = False) -> QuerySet:
    """
    Returns a cloned query with empty cache. Optionally just clear the cache and return the same query.
    """
    if clear_cache:
        self._clear_cache()
        return self
    return self._clone()

or_ 

or_(*clauses, **kwargs)

Filters the QuerySet by the OR operand.

Source code in edgy/core/db/querysets/base.py 
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
def or_(
    self,
    *clauses: Union[
        sqlalchemy.sql.expression.BinaryExpression,
        Callable[
            [QuerySetType],
            Union[
                sqlalchemy.sql.expression.BinaryExpression,
                Awaitable[sqlalchemy.sql.expression.BinaryExpression],
            ],
        ],
        dict[str, Any],
        QuerySet,
    ],
    **kwargs: Any,
) -> QuerySet:
    """
    Filters the QuerySet by the OR operand.
    """
    return self._filter_or_exclude(clauses=clauses, or_=True, kwargs=kwargs)

local_or 

local_or(*clauses, **kwargs)

Filters the QuerySet by the OR operand.

Source code in edgy/core/db/querysets/base.py 
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
def local_or(
    self,
    *clauses: Union[
        sqlalchemy.sql.expression.BinaryExpression,
        Callable[
            [QuerySetType],
            Union[
                sqlalchemy.sql.expression.BinaryExpression,
                Awaitable[sqlalchemy.sql.expression.BinaryExpression],
            ],
        ],
        dict[str, Any],
        QuerySet,
    ],
    **kwargs: Any,
) -> QuerySet:
    """
    Filters the QuerySet by the OR operand.
    """
    return self._filter_or_exclude(
        clauses=clauses, or_=True, kwargs=kwargs, allow_global_or=False
    )

and_ 

and_(*clauses, **kwargs)

Filters the QuerySet by the AND operand. Alias of filter.

Source code in edgy/core/db/querysets/base.py 
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
def and_(
    self,
    *clauses: Union[
        sqlalchemy.sql.expression.BinaryExpression,
        Callable[
            [QuerySetType],
            Union[
                sqlalchemy.sql.expression.BinaryExpression,
                Awaitable[sqlalchemy.sql.expression.BinaryExpression],
            ],
        ],
        dict[str, Any],
    ],
    **kwargs: Any,
) -> QuerySet:
    """
    Filters the QuerySet by the AND operand. Alias of filter.
    """
    return self._filter_or_exclude(clauses=clauses, kwargs=kwargs)

not_ 

not_(*clauses, **kwargs)

Filters the QuerySet by the NOT operand. Alias of exclude.

Source code in edgy/core/db/querysets/base.py 
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
def not_(
    self,
    *clauses: Union[
        sqlalchemy.sql.expression.BinaryExpression,
        Callable[
            [QuerySetType],
            Union[
                sqlalchemy.sql.expression.BinaryExpression,
                Awaitable[sqlalchemy.sql.expression.BinaryExpression],
            ],
        ],
        dict[str, Any],
        QuerySet,
    ],
    **kwargs: Any,
) -> QuerySet:
    """
    Filters the QuerySet by the NOT operand. Alias of exclude.
    """
    return self.exclude(*clauses, **kwargs)

exclude 

exclude(*clauses, **kwargs)

Exactly the same as the filter but for the exclude.

Source code in edgy/core/db/querysets/base.py 
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
def exclude(
    self,
    *clauses: Union[
        sqlalchemy.sql.expression.BinaryExpression,
        Callable[
            [QuerySetType],
            Union[
                sqlalchemy.sql.expression.BinaryExpression,
                Awaitable[sqlalchemy.sql.expression.BinaryExpression],
            ],
        ],
        dict[str, Any],
        QuerySet,
    ],
    **kwargs: Any,
) -> QuerySet:
    """
    Exactly the same as the filter but for the exclude.
    """
    return self._filter_or_exclude(clauses=clauses, exclude=True, kwargs=kwargs)

exclude_secrets 

exclude_secrets(exclude_secrets=True)

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

Source code in edgy/core/db/querysets/base.py 
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
def exclude_secrets(
    self,
    exclude_secrets: bool = True,
) -> QuerySet:
    """
    Excludes any field that contains the `secret=True` declared from being leaked.
    """
    queryset = self._clone()
    queryset._exclude_secrets = exclude_secrets
    return queryset

extra_select 

extra_select(*extra)
Source code in edgy/core/db/querysets/base.py 
1167
1168
1169
1170
1171
1172
1173
def extra_select(
    self,
    *extra: sqlalchemy.expression.ColumnClause,
) -> QuerySetType:
    queryset = self._clone()
    queryset._extra_select.extend(extra)
    return queryset

reference_select 

reference_select(references)
Source code in edgy/core/db/querysets/base.py 
1175
1176
1177
1178
def reference_select(self, references: reference_select_type) -> QuerySetType:
    queryset = self._clone()
    queryset._reference_select.update(references)
    return queryset

batch_size 

batch_size(batch_size=None)

Set batch/chunk size. Used for iterate

Source code in edgy/core/db/querysets/base.py 
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
def batch_size(
    self,
    batch_size: Optional[int] = None,
) -> QuerySet:
    """
    Set batch/chunk size. Used for iterate
    """
    queryset = self._clone()
    queryset._batch_size = batch_size
    return queryset

lookup 

lookup(term)

Broader way of searching for a given term

Source code in edgy/core/db/querysets/base.py 
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
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.meta.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.

Source code in edgy/core/db/querysets/base.py 
1216
1217
1218
1219
1220
1221
1222
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

reverse 

reverse()
Source code in edgy/core/db/querysets/base.py 
1224
1225
1226
1227
1228
1229
def reverse(self) -> QuerySet:
    queryset: QuerySet = self._clone()
    queryset._order_by = tuple(
        el[1:] if el.startswith("-") else f"-{el}" for el in queryset._order_by
    )
    return queryset

limit 

limit(limit_count)

Returns a QuerySet limited by.

Source code in edgy/core/db/querysets/base.py 
1231
1232
1233
1234
1235
1236
1237
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.

Source code in edgy/core/db/querysets/base.py 
1239
1240
1241
1242
1243
1244
1245
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.

Source code in edgy/core/db/querysets/base.py 
1247
1248
1249
1250
1251
1252
1253
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(first=True, *distinct_on)

Returns a queryset with distinct results.

Source code in edgy/core/db/querysets/base.py 
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
def distinct(self, first: Union[bool, str] = True, *distinct_on: str) -> QuerySet:
    """
    Returns a queryset with distinct results.
    """
    queryset: QuerySet = self._clone()
    if first is False:
        queryset.distinct_on = None
    elif first is True:
        queryset.distinct_on = []
    else:
        queryset.distinct_on = [first, *distinct_on]
    return queryset

only 

only(*fields)

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

Source code in edgy/core/db/querysets/base.py 
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
def only(self, *fields: str) -> QuerySet:
    """
    Returns a list of models with the selected only fields and always the primary
    key.
    """
    queryset: QuerySet = self._clone()
    only_fields = set(fields)
    if self.model_class.pknames:
        for pkname in self.model_class.pknames:
            if pkname not in fields:
                for pkcolumn in self.model_class.meta.get_columns_for_name(pkname):
                    only_fields.add(pkcolumn.key)
    else:
        for pkcolumn in self.model_class.pkcolumns:
            only_fields.add(pkcolumn.key)
    queryset._only = only_fields
    return queryset

defer 

defer(*fields)

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

Source code in edgy/core/db/querysets/base.py 
1286
1287
1288
1289
1290
1291
1292
1293
1294
def defer(self, *fields: str) -> QuerySet:
    """
    Returns a list of models with the selected only fields and always the primary
    key.
    """
    queryset: QuerySet = self._clone()

    queryset._defer = set(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.

Source code in edgy/core/db/querysets/base.py 
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
1315
1316
def select_related(self, *related: str) -> 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 len(related) >= 1 and not isinstance(cast(Any, related[0]), str):
        warnings.warn(
            "use `select_related` with variadic str arguments instead of a Sequence",
            DeprecationWarning,
            stacklevel=2,
        )
        related = cast(tuple[str, ...], related[0])
    if not self._select_related.issuperset(related):
        queryset._cached_select_related_expression = None
        queryset._select_related.update(related)
    return queryset

values async 

values(fields=None, exclude=None, exclude_none=False)

Returns the results in a python dictionary format.

Source code in edgy/core/db/querysets/base.py 
1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
1334
1335
1336
1337
1338
1339
1340
1341
1342
1343
async def values(
    self,
    fields: Union[Sequence[str], str, None] = None,
    exclude: Union[Sequence[str], set[str]] = None,
    exclude_none: bool = False,
) -> list[Any]:
    """
    Returns the results in a python dictionary format.
    """

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

    rows: list[BaseModelType] = await self

    if fields is not None and not isinstance(fields, Iterable):
        raise QuerySetError(detail="Fields must be a suitable sequence of strings or unset.")

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

values_list async 

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

Returns the results in a python dictionary format.

Source code in edgy/core/db/querysets/base.py 
1345
1346
1347
1348
1349
1350
1351
1352
1353
1354
1355
1356
1357
1358
1359
1360
1361
1362
1363
1364
1365
1366
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.
    """
    rows = await self.values(
        fields=fields,
        exclude=exclude,
        exclude_none=exclude_none,
    )
    if not flat:
        return [tuple(row.values()) for row in rows]
    else:
        try:
            return [row[fields[0]] for row in rows]
        except KeyError:
            raise QuerySetError(detail=f"{fields[0]} does not exist in the results.") from None

exists async 

exists(**kwargs)

Returns a boolean indicating if a record exists or not.

Source code in edgy/core/db/querysets/base.py 
1368
1369
1370
1371
1372
1373
1374
1375
1376
1377
1378
1379
1380
1381
1382
1383
1384
1385
async def exists(self, **kwargs: Any) -> bool:
    """
    Returns a boolean indicating if a record exists or not.
    """
    if kwargs:
        # check cache for existance
        cached = self._cache.get(self.model_class, kwargs)
        if cached is not None:
            return True
        filter_query = self.filter(**kwargs)
        filter_query._cache = self._cache
        return await filter_query.exists()
    queryset: QuerySet = self
    expression = (await queryset.as_select()).exists().select()
    check_db_connection(queryset.database)
    async with queryset.database as database:
        _exists = await database.fetch_val(expression)
    return cast(bool, _exists)

count async 

count()

Returns an indicating the total records.

Source code in edgy/core/db/querysets/base.py 
1387
1388
1389
1390
1391
1392
1393
1394
1395
1396
1397
1398
1399
async def count(self) -> int:
    """
    Returns an indicating the total records.
    """
    if self._cache_count is not None:
        return self._cache_count
    queryset: QuerySet = self
    expression = (await queryset.as_select()).alias("subquery_for_count")
    expression = sqlalchemy.func.count().select().select_from(expression)
    check_db_connection(queryset.database)
    async with queryset.database as database:
        self._cache_count = count = cast("int", await database.fetch_val(expression))
    return count

get_or_none async 

get_or_none(**kwargs)

Fetch one object matching the parameters or returns None.

Source code in edgy/core/db/querysets/base.py 
1401
1402
1403
1404
1405
1406
1407
1408
async def get_or_none(self, **kwargs: Any) -> Union[EdgyEmbedTarget, None]:
    """
    Fetch one object matching the parameters or returns None.
    """
    try:
        return await self.get(**kwargs)
    except ObjectNotFound:
        return None

get async 

get(**kwargs)

Returns a single record based on the given kwargs.

Source code in edgy/core/db/querysets/base.py 
1410
1411
1412
1413
1414
async def get(self, **kwargs: Any) -> EdgyEmbedTarget:
    """
    Returns a single record based on the given kwargs.
    """
    return cast(EdgyEmbedTarget, (await self._get_raw(**kwargs))[1])

first async 

first()

Returns the first record of a given queryset.

Source code in edgy/core/db/querysets/base.py 
1416
1417
1418
1419
1420
1421
1422
1423
1424
1425
1426
1427
1428
1429
1430
1431
1432
1433
1434
1435
1436
async def first(self) -> Union[EdgyEmbedTarget, None]:
    """
    Returns the first record of a given queryset.
    """
    if self._cache_count is not None and self._cache_count == 0:
        return None
    if self._cache_first is not None:
        return cast(EdgyEmbedTarget, self._cache_first[1])
    queryset = self
    if not queryset._order_by:
        queryset = queryset.order_by(*self.model_class.pkcolumns)
    expression, tables_and_models = await queryset.as_select_with_tables()
    self._cached_select_related_expression = queryset._cached_select_related_expression
    check_db_connection(queryset.database)
    async with queryset.database as database:
        row = await database.fetch_one(expression, pos=0)
    if row:
        return (
            await self._get_or_cache_row(row, tables_and_models, extra_attr="_cache_first")
        )[1]
    return None

last async 

last()

Returns the last record of a given queryset.

Source code in edgy/core/db/querysets/base.py 
1438
1439
1440
1441
1442
1443
1444
1445
1446
1447
1448
1449
1450
1451
1452
1453
1454
1455
1456
1457
1458
1459
async def last(self) -> Union[EdgyEmbedTarget, None]:
    """
    Returns the last record of a given queryset.
    """
    if self._cache_count is not None and self._cache_count == 0:
        return None
    if self._cache_last is not None:
        return cast(EdgyEmbedTarget, self._cache_last[1])
    queryset = self
    if not queryset._order_by:
        queryset = queryset.order_by(*self.model_class.pkcolumns)
    queryset = queryset.reverse()
    expression, tables_and_models = await queryset.as_select_with_tables()
    self._cached_select_related_expression = queryset._cached_select_related_expression
    check_db_connection(queryset.database)
    async with queryset.database as database:
        row = await database.fetch_one(expression, pos=0)
    if row:
        return (
            await self._get_or_cache_row(row, tables_and_models, extra_attr="_cache_last")
        )[1]
    return None

create async 

create(*args, **kwargs)

Creates a record in a specific table.

Source code in edgy/core/db/querysets/base.py 
1461
1462
1463
1464
1465
1466
1467
1468
1469
1470
1471
1472
1473
1474
1475
1476
1477
1478
1479
1480
1481
1482
1483
1484
1485
async def create(self, *args: Any, **kwargs: Any) -> EdgyEmbedTarget:
    """
    Creates a record in a specific table.
    """
    # for tenancy
    queryset: QuerySet = self._clone()
    check_db_connection(queryset.database)
    token = CHECK_DB_CONNECTION_SILENCED.set(True)
    try:
        instance = queryset.model_class(*args, **kwargs)
        apply_instance_extras(
            instance,
            self.model_class,
            schema=self.using_schema,
            table=queryset.table,
            database=queryset.database,
        )
        # values=kwargs is required for ensuring all kwargs are seen as explicit kwargs
        instance = await instance.save(force_insert=True, values=set(kwargs.keys()))
        result = await self._embed_parent_in_result(instance)
        self._clear_cache(True)
        self._cache.update([result])
        return cast(EdgyEmbedTarget, result[1])
    finally:
        CHECK_DB_CONNECTION_SILENCED.reset(token)

bulk_create async 

bulk_create(objs)

Bulk creates records in a table

Source code in edgy/core/db/querysets/base.py 
1487
1488
1489
1490
1491
1492
1493
1494
1495
1496
1497
1498
1499
1500
1501
1502
1503
1504
1505
1506
1507
1508
1509
1510
1511
1512
1513
1514
1515
1516
1517
1518
1519
1520
1521
1522
1523
1524
1525
1526
1527
1528
1529
async def bulk_create(self, objs: Iterable[Union[dict[str, Any], EdgyModel]]) -> None:
    """
    Bulk creates records in a table
    """
    queryset: QuerySet = self._clone()

    new_objs: list[EdgyModel] = []

    async def _iterate(obj_or_dict: Union[EdgyModel, dict[str, Any]]) -> dict[str, Any]:
        if isinstance(obj_or_dict, dict):
            obj: EdgyModel = queryset.model_class(**obj_or_dict)
            if (
                self.model_class.meta.post_save_fields
                and not self.model_class.meta.post_save_fields.isdisjoint(obj_or_dict.keys())
            ):
                new_objs.append(obj)
        else:
            obj = obj_or_dict
            if self.model_class.meta.post_save_fields:
                new_objs.append(obj)
        original = obj.extract_db_fields()
        col_values: dict[str, Any] = obj.extract_column_values(
            original, phase="prepare_insert", instance=self
        )
        col_values.update(
            await obj.execute_pre_save_hooks(col_values, original, force_insert=True)
        )
        return col_values

    check_db_connection(queryset.database)
    token = CURRENT_INSTANCE.set(self)
    try:
        async with queryset.database as database, database.transaction():
            expression = queryset.table.insert().values([await _iterate(obj) for obj in objs])
            await database.execute_many(expression)
        self._clear_cache(True)
        if new_objs:
            for obj in new_objs:
                await obj.execute_post_save_hooks(
                    self.model_class.meta.fields.keys(), force_insert=True
                )
    finally:
        CURRENT_INSTANCE.reset(token)

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 Edgy.

Source code in edgy/core/db/querysets/base.py 
1531
1532
1533
1534
1535
1536
1537
1538
1539
1540
1541
1542
1543
1544
1545
1546
1547
1548
1549
1550
1551
1552
1553
1554
1555
1556
1557
1558
1559
1560
1561
1562
1563
1564
1565
1566
1567
1568
1569
1570
1571
1572
1573
1574
1575
1576
1577
1578
1579
1580
1581
1582
1583
1584
1585
1586
1587
1588
1589
1590
async def bulk_update(self, objs: list[EdgyModel], 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 Edgy.
    """
    queryset: QuerySet = self._clone()
    fields = list(fields)

    pk_query_placeholder = (
        getattr(queryset.table.c, pkcol)
        == sqlalchemy.bindparam(
            "__id" if pkcol == "id" else pkcol, type_=getattr(queryset.table.c, pkcol).type
        )
        for pkcol in queryset.pkcolumns
    )
    expression = queryset.table.update().where(*pk_query_placeholder)

    update_list = []
    fields_plus_pk = {*fields, *queryset.model_class.pkcolumns}
    check_db_connection(queryset.database)
    token = CURRENT_INSTANCE.set(self)
    try:
        async with queryset.database as database, database.transaction():
            for obj in objs:
                extracted = obj.extract_db_fields(fields_plus_pk)
                update = queryset.model_class.extract_column_values(
                    extracted,
                    is_update=True,
                    is_partial=True,
                    phase="prepare_update",
                    instance=self,
                    model_instance=obj,
                )
                update.update(
                    await obj.execute_pre_save_hooks(update, extracted, force_insert=False)
                )
                if "id" in update:
                    update["__id"] = update.pop("id")
                update_list.append(update)

            values_placeholder: Any = {
                pkcol: sqlalchemy.bindparam(pkcol, type_=getattr(queryset.table.c, pkcol).type)
                for field in fields
                for pkcol in queryset.model_class.meta.field_to_column_names[field]
            }
            expression = expression.values(values_placeholder)
            await database.execute_many(expression, update_list)
        self._clear_cache()
        if (
            self.model_class.meta.post_save_fields
            and not self.model_class.meta.post_save_fields.isdisjoint(fields)
        ):
            for obj in objs:
                await obj.execute_post_save_hooks(fields, force_insert=False)
    finally:
        CURRENT_INSTANCE.reset(token)

bulk_get_or_create async 

bulk_get_or_create(objs, unique_fields=None)

Bulk gets or creates records in a table.

If records exist based on unique fields, they are retrieved. Otherwise, new records are created.

PARAMETER DESCRIPTION
objs

A list of objects or dictionaries.

TYPE: list[Union[dict[str, Any], EdgyModel]]

unique_fields

Fields that determine uniqueness. If None, all records are treated as new.

TYPE: list[str] | None DEFAULT: None

RETURNS DESCRIPTION
list[EdgyModel]

list[EdgyModel]: A list of retrieved or newly created objects.
Source code in edgy/core/db/querysets/base.py 
1594
1595
1596
1597
1598
1599
1600
1601
1602
1603
1604
1605
1606
1607
1608
1609
1610
1611
1612
1613
1614
1615
1616
1617
1618
1619
1620
1621
1622
1623
1624
1625
1626
1627
1628
1629
1630
1631
1632
1633
1634
1635
1636
1637
1638
1639
1640
1641
1642
1643
1644
1645
1646
1647
1648
1649
1650
1651
1652
1653
1654
1655
1656
1657
1658
1659
1660
1661
1662
1663
1664
1665
1666
1667
1668
1669
1670
1671
1672
1673
1674
1675
1676
1677
1678
1679
1680
1681
1682
1683
1684
1685
1686
1687
1688
1689
1690
1691
1692
1693
1694
1695
1696
1697
1698
1699
1700
1701
1702
1703
1704
1705
1706
1707
1708
1709
1710
1711
1712
1713
1714
1715
1716
async def bulk_get_or_create(
    self,
    objs: list[Union[dict[str, Any], EdgyModel]],
    unique_fields: Union[list[str], None] = None,
) -> list[EdgyModel]:
    """
    Bulk gets or creates records in a table.

    If records exist based on unique fields, they are retrieved.
    Otherwise, new records are created.

    Args:
        objs (list[Union[dict[str, Any], EdgyModel]]): A list of objects or dictionaries.
        unique_fields (list[str] | None): Fields that determine uniqueness. If None, all records are treated as new.

    Returns:
        list[EdgyModel]: A list of retrieved or newly created objects.
    """
    queryset: QuerySet = self._clone()
    new_objs: list[EdgyModel] = []
    retrieved_objs: list[EdgyModel] = []

    if unique_fields:
        existing_records = {}
        for obj in objs:
            if isinstance(obj, dict):
                filter_kwargs = {}
                dict_fields = {}
                for field, value in obj.items():
                    if field in unique_fields:
                        if isinstance(value, dict):
                            dict_fields[field] = value
                        else:
                            filter_kwargs[field] = value
                db_records = [record async for record in queryset.filter(**filter_kwargs)]
                found = False
                for record in db_records:
                    record_dict_fields = {k: getattr(record, k) for k in dict_fields}
                    if dict_fields == record_dict_fields:
                        lookup_key = tuple(
                            json.dumps(getattr(record, field))
                            if isinstance(getattr(record, field), dict)
                            else getattr(record, field)
                            for field in unique_fields
                        )
                        existing_records[lookup_key] = record
                        retrieved_objs.append(record)
                        found = True
                        break
                if found is False:
                    new_objs.append(queryset.model_class(**obj))
            else:
                filter_kwargs = {}
                dict_fields = {}
                for field in unique_fields:
                    value = getattr(obj, field)
                    if isinstance(value, dict):
                        dict_fields[field] = value
                    else:
                        filter_kwargs[field] = value
                db_records = [record async for record in queryset.filter(**filter_kwargs)]
                found = False
                for record in db_records:
                    record_dict_fields = {
                        k: getattr(record, k) for k, _ in dict_fields.items()
                    }
                    if dict_fields == record_dict_fields:
                        lookup_key = tuple(
                            json.dumps(getattr(record, field))
                            if isinstance(getattr(record, field), dict)
                            else getattr(record, field)
                            for field in unique_fields
                        )
                        existing_records[lookup_key] = record
                        retrieved_objs.append(record)
                        found = True
                        break
                if found is False:
                    new_objs.append(obj)

    else:
        new_objs.extend(
            [queryset.model_class(**obj) if isinstance(obj, dict) else obj for obj in objs]
        )
        existing_records = {}

    async def _prepare_obj(obj_or_dict: Union[EdgyModel, dict[str, Any]]) -> EdgyModel:
        if isinstance(obj_or_dict, dict):
            obj: EdgyModel = queryset.model_class(**obj_or_dict)
        else:
            obj = obj_or_dict
        return obj

    async def _iterate(obj: EdgyModel) -> dict[str, Any]:
        original = obj.extract_db_fields()
        col_values: dict[str, Any] = obj.extract_column_values(
            original, phase="prepare_insert", instance=self
        )
        col_values.update(
            await obj.execute_pre_save_hooks(col_values, original, force_insert=True)
        )
        return col_values

    check_db_connection(queryset.database)
    token = CURRENT_INSTANCE.set(self)

    try:
        async with queryset.database as database, database.transaction():
            if new_objs:
                new_obj_values = [await _iterate(obj) for obj in new_objs]
                expression = queryset.table.insert().values(new_obj_values)
                await database.execute_many(expression)
                retrieved_objs.extend(new_objs)

            self._clear_cache()
            for obj in new_objs:
                await obj.execute_post_save_hooks(
                    self.model_class.meta.fields.keys(), force_insert=True
                )
    finally:
        CURRENT_INSTANCE.reset(token)

    return retrieved_objs

delete async 

delete(use_models=False)
Source code in edgy/core/db/querysets/base.py 
1718
1719
1720
1721
1722
1723
1724
1725
1726
1727
1728
1729
1730
1731
1732
1733
1734
1735
1736
1737
1738
1739
1740
1741
async def delete(self, use_models: bool = False) -> int:
    if (
        self.model_class.__require_model_based_deletion__
        or self.model_class.meta.post_delete_fields
    ):
        use_models = True
    if use_models:
        return await self._model_based_delete()

    # delete of model issues already signals, so don't integrate them
    await self.model_class.meta.signals.pre_delete.send_async(self.__class__, instance=self)

    expression = self.table.delete()
    expression = expression.where(await self.build_where_clause())

    check_db_connection(self.database)
    async with self.database as database:
        row_count = cast(int, await database.execute(expression))

    # clear cache before executing post_delete. Fresh results can be retrieved in signals
    self._clear_cache()

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

update async 

update(**kwargs)

Updates records in a specific table with the given kwargs.

Warning: does not execute pre_save_callback/post_save_callback hooks and passes values directly to clean.

Source code in edgy/core/db/querysets/base.py 
1743
1744
1745
1746
1747
1748
1749
1750
1751
1752
1753
1754
1755
1756
1757
1758
1759
1760
1761
1762
1763
1764
1765
1766
1767
async def update(self, **kwargs: Any) -> None:
    """
    Updates records in a specific table with the given kwargs.

    Warning: does not execute pre_save_callback/post_save_callback hooks and passes values directly to clean.
    """

    column_values = self.model_class.extract_column_values(
        kwargs, is_update=True, is_partial=True, phase="prepare_update", instance=self
    )

    # Broadcast the initial update details
    await self.model_class.meta.signals.pre_update.send_async(
        self.__class__, instance=self, kwargs=column_values
    )

    expression = self.table.update().values(**column_values)
    expression = expression.where(await self.build_where_clause())
    check_db_connection(self.database)
    async with self.database as database:
        await database.execute(expression)

    # Broadcast the update executed
    await self.model_class.meta.signals.post_update.send_async(self.__class__, instance=self)
    self._clear_cache()

get_or_create async 

get_or_create(defaults=None, *args, **kwargs)

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

Source code in edgy/core/db/querysets/base.py 
1769
1770
1771
1772
1773
1774
1775
1776
1777
1778
1779
1780
1781
1782
1783
1784
1785
1786
1787
1788
1789
1790
1791
1792
1793
1794
1795
1796
1797
1798
1799
1800
1801
1802
1803
1804
1805
async def get_or_create(
    self, defaults: Union[dict[str, Any], Any, None] = None, *args: Any, **kwargs: Any
) -> tuple[EdgyEmbedTarget, bool]:
    """
    Creates a record in a specific table or updates if already exists.
    """
    if not isinstance(defaults, dict):
        # can be a ModelRef so pass it
        args = (defaults, *args)
        defaults = {}

    try:
        raw_instance, get_instance = await self._get_raw(**kwargs)
    except ObjectNotFound:
        kwargs.update(defaults)
        instance: EdgyEmbedTarget = await self.create(*args, **kwargs)
        return instance, True
    for arg in args:
        if isinstance(arg, ModelRef):
            relation_field = self.model_class.meta.fields[arg.__related_name__]
            extra_params = {}
            try:
                # m2m or foreign key
                target_model_class = relation_field.target
            except AttributeError:
                # reverse m2m or foreign key
                target_model_class = relation_field.related_from
            if not relation_field.is_m2m:
                # sometimes the foreign key is required, so set it already
                extra_params[relation_field.foreign_key.name] = raw_instance
            model = target_model_class(
                **arg.model_dump(exclude={"__related_name__"}),
                **extra_params,
            )
            relation = getattr(raw_instance, arg.__related_name__)
            await relation.add(model)
    return cast(EdgyEmbedTarget, get_instance), False

update_or_create async 

update_or_create(defaults=None, *args, **kwargs)

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

Source code in edgy/core/db/querysets/base.py 
1807
1808
1809
1810
1811
1812
1813
1814
1815
1816
1817
1818
1819
1820
1821
1822
1823
1824
1825
1826
1827
1828
1829
1830
1831
1832
1833
1834
1835
1836
1837
1838
1839
1840
1841
1842
1843
1844
async def update_or_create(
    self, defaults: Union[dict[str, Any], Any, None] = None, *args: Any, **kwargs: Any
) -> tuple[EdgyEmbedTarget, bool]:
    """
    Updates a record in a specific table or creates a new one.
    """
    if not isinstance(defaults, dict):
        # can be a ModelRef so pass it
        args = (defaults, *args)
        defaults = {}
    try:
        raw_instance, get_instance = await self._get_raw(**kwargs)
    except ObjectNotFound:
        kwargs.update(defaults)
        instance: EdgyEmbedTarget = await self.create(*args, **kwargs)
        return instance, True
    await get_instance.update(**defaults)
    for arg in args:
        if isinstance(arg, ModelRef):
            relation_field = self.model_class.meta.fields[arg.__related_name__]
            extra_params = {}
            try:
                # m2m or foreign key
                target_model_class = relation_field.target
            except AttributeError:
                # reverse m2m or foreign key
                target_model_class = relation_field.related_from
            if not relation_field.is_m2m:
                # sometimes the foreign key is required, so set it already
                extra_params[relation_field.foreign_key.name] = raw_instance
            model = target_model_class(
                **arg.model_dump(exclude={"__related_name__"}),
                **extra_params,
            )
            relation = getattr(raw_instance, arg.__related_name__)
            await relation.add(model)
    self._clear_cache()
    return cast(EdgyEmbedTarget, get_instance), False

contains async 

contains(instance)

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

Source code in edgy/core/db/querysets/base.py 
1846
1847
1848
1849
1850
1851
1852
1853
1854
1855
1856
1857
1858
1859
async def contains(self, instance: BaseModelType) -> bool:
    """Returns true if the QuerySet contains the provided object.
    False if otherwise.
    """
    query: Any = {}
    try:
        pkcolumns = instance.pkcolumns
        for pkcolumn in pkcolumns:
            query[pkcolumn] = getattr(instance, pkcolumn)
            if query[pkcolumn] is None:
                raise ValueError("'obj' has incomplete/missing primary key.") from None
    except AttributeError:
        raise ValueError("'obj' must be a model or reflect model instance.") from None
    return await self.exists(**query)

transaction 

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

Return database transaction for the assigned database.

Source code in edgy/core/db/querysets/base.py 
1861
1862
1863
def transaction(self, *, force_rollback: bool = False, **kwargs: Any) -> Transaction:
    """Return database transaction for the assigned database."""
    return self.database.transaction(force_rollback=force_rollback, **kwargs)

__aiter__ async 

__aiter__()
Source code in edgy/core/db/querysets/base.py 
1870
1871
1872
async def __aiter__(self) -> AsyncIterator[Any]:
    async for value in self._execute_iterate():
        yield value