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)

Bases: BaseQuerySet

QuerySet object used for query retrieving. Public interface

PARAMETER DESCRIPTION
model_class

TYPE: Union[type[BaseModelType], None] DEFAULT: None

database

TYPE: Union[Database, None] DEFAULT: None

filter_clauses

TYPE: Iterable[Any] DEFAULT: _empty_set

select_related

TYPE: Iterable[str] DEFAULT: _empty_set

prefetch_related

TYPE: Iterable[Prefetch] DEFAULT: _empty_set

limit_count

TYPE: Optional[int] DEFAULT: None

limit

TYPE: Optional[int] DEFAULT: None

limit_offset

TYPE: Optional[int] DEFAULT: None

offset

TYPE: Optional[int] DEFAULT: None

batch_size

TYPE: Optional[int] DEFAULT: None

order_by

TYPE: Iterable[str] DEFAULT: _empty_set

group_by

TYPE: Iterable[str] DEFAULT: _empty_set

distinct_on

TYPE: Union[None, Literal[True], Iterable[str]] DEFAULT: None

distinct

TYPE: Union[None, Literal[True], Iterable[str]] DEFAULT: None

only_fields

TYPE: Optional[Iterable[str]] DEFAULT: None

only

TYPE: Iterable[str] DEFAULT: _empty_set

defer_fields

TYPE: Optional[Sequence[str]] DEFAULT: None

defer

TYPE: Iterable[str] DEFAULT: _empty_set

embed_parent

TYPE: Optional[tuple[str, Union[str, str]]] DEFAULT: None

embed_parent_filters

TYPE: Optional[tuple[str, str]] DEFAULT: None

using_schema

TYPE: Union[str, None, Any] DEFAULT: Undefined

table

TYPE: Optional[Table] DEFAULT: None

exclude_secrets

TYPE: bool DEFAULT: False

Source code in edgy/core/db/querysets/base.py 
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
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,
) -> 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._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

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

PARAMETER DESCRIPTION
_

TYPE: Any DEFAULT: None

tables_and_models

TYPE: Optional[tables_and_models_type] DEFAULT: None

Source code in edgy/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
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
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.

PARAMETER DESCRIPTION
_positional

TYPE: Any DEFAULT: _sentinel

database

TYPE: Union[str, Any, None, Database] DEFAULT: Undefined

schema

TYPE: Union[str, Any, None, Literal[False]] DEFAULT: Undefined

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.

PARAMETER DESCRIPTION
*prefetch

TYPE: Prefetch DEFAULT: ()

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

PARAMETER DESCRIPTION
connection_name

TYPE: str

schema

TYPE: Union[str, Any, None, Literal[False]] DEFAULT: Undefined

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 
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
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,
    )
    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)
PARAMETER DESCRIPTION
keep_result_cache

TYPE: bool DEFAULT: False

Source code in edgy/core/db/querysets/base.py 
191
192
193
194
195
196
197
198
199
200
201
202
203
204
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

PARAMETER DESCRIPTION
order_by

TYPE: Any

expression

TYPE: Any

Source code in edgy/core/db/querysets/base.py 
206
207
208
209
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

PARAMETER DESCRIPTION
group_by

TYPE: Any

expression

TYPE: Any

Source code in edgy/core/db/querysets/base.py 
211
212
213
214
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

PARAMETER DESCRIPTION
distinct_on

TYPE: Optional[Sequence[str]]

expression

TYPE: Any

Source code in edgy/core/db/querysets/base.py 
261
262
263
264
265
266
267
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)
PARAMETER DESCRIPTION
join_clause

TYPE: Any

current_transition

TYPE: tuple[str, str, str]

transitions

TYPE: dict[tuple[str, str, str], tuple[Any, Optional[tuple[str, str, str]], str]]

tables_and_models

TYPE: dict[str, tuple[Table, type[BaseModelType]]]

Source code in edgy/core/db/querysets/base.py 
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
@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 
297
298
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
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
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)
PARAMETER DESCRIPTION
foreign_key

TYPE: BaseForeignKey

table

TYPE: Any

reverse

TYPE: bool

former_table

TYPE: Any

Source code in edgy/core/db/querysets/base.py 
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
@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 
462
463
464
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 
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
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
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: list[Any] = []
    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.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.append(column.label(f"{table.name}_{column_key}"))
    assert columns, "no columns specified"
    # all columns are aliased already
    expression = sqlalchemy.sql.select(*columns).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 
537
538
539
540
541
542
543
544
545
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 
547
548
549
550
async def as_select(
    self,
) -> Any:
    return (await self.as_select_with_tables())[0]

_kwargs_to_clauses 

_kwargs_to_clauses(kwargs)
PARAMETER DESCRIPTION
kwargs

TYPE: Any

Source code in edgy/core/db/querysets/base.py 
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
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
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,
            ) -> 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)
PARAMETER DESCRIPTION
order_by

TYPE: str

Source code in edgy/core/db/querysets/base.py 
619
620
621
622
623
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)
PARAMETER DESCRIPTION
distinct_on

TYPE: str

Source code in edgy/core/db/querysets/base.py 
625
626
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)
PARAMETER DESCRIPTION
result

TYPE: Union[EdgyModel, Awaitable[EdgyModel]]

Source code in edgy/core/db/querysets/base.py 
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
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)
PARAMETER DESCRIPTION
row

TYPE: Any

tables_and_models

TYPE: dict[str, tuple[Table, type[BaseModelType]]]

extra_attr

TYPE: str DEFAULT: ''

raw

TYPE: bool DEFAULT: False

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
667
668
669
670
671
672
673
674
675
676
677
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,
            ),
            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 
679
680
681
682
683
684
685
686
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)
PARAMETER DESCRIPTION
batch

TYPE: Sequence[Row]

tables_and_models

TYPE: dict[str, tuple[Table, type[BaseModelType]]]

queryset

TYPE: BaseQuerySet

Source code in edgy/core/db/querysets/base.py 
688
689
690
691
692
693
694
695
696
697
698
699
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
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
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,
            ),
            transform_fn=self._embed_parent_in_result,
        ),
    )

_execute_iterate async 

_execute_iterate(fetch_all_at_once=False)

Executes the query, iterate.

PARAMETER DESCRIPTION
fetch_all_at_once

TYPE: bool DEFAULT: False

Source code in edgy/core/db/querysets/base.py 
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
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
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 
837
838
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.

PARAMETER DESCRIPTION
kwargs

TYPE: Any

clauses

TYPE: Sequence[Union[BinaryExpression, Callable[[QuerySetType], Union[BinaryExpression, Awaitable[BinaryExpression]]], dict[str, Any], QuerySet]]

exclude

TYPE: bool DEFAULT: False

or_

TYPE: bool DEFAULT: False

allow_global_or

TYPE: bool DEFAULT: True

Source code in edgy/core/db/querysets/base.py 
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
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
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 
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
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.

PARAMETER DESCRIPTION
**kwargs

TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py 
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
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 
978
979
def __repr__(self) -> str:
    return f"QuerySet<{self.sql}>"

_sql_helper async 

_sql_helper()
Source code in edgy/core/db/querysets/base.py 
987
988
989
990
991
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.

PARAMETER DESCRIPTION
*clauses

TYPE: Union[BinaryExpression, Callable[[QuerySetType], Union[BinaryExpression, Awaitable[BinaryExpression]]], dict[str, Any], QuerySet] DEFAULT: ()

**kwargs

TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py 
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
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.

PARAMETER DESCRIPTION
clear_cache

TYPE: bool DEFAULT: False

Source code in edgy/core/db/querysets/base.py 
1019
1020
1021
1022
1023
1024
1025
1026
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.

PARAMETER DESCRIPTION
*clauses

TYPE: Union[BinaryExpression, Callable[[QuerySetType], Union[BinaryExpression, Awaitable[BinaryExpression]]], dict[str, Any], QuerySet] DEFAULT: ()

**kwargs

TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py 
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
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.

PARAMETER DESCRIPTION
*clauses

TYPE: Union[BinaryExpression, Callable[[QuerySetType], Union[BinaryExpression, Awaitable[BinaryExpression]]], dict[str, Any], QuerySet] DEFAULT: ()

**kwargs

TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py 
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
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.

PARAMETER DESCRIPTION
*clauses

TYPE: Union[BinaryExpression, Callable[[QuerySetType], Union[BinaryExpression, Awaitable[BinaryExpression]]], dict[str, Any]] DEFAULT: ()

**kwargs

TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py 
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
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.

PARAMETER DESCRIPTION
*clauses

TYPE: Union[BinaryExpression, Callable[[QuerySetType], Union[BinaryExpression, Awaitable[BinaryExpression]]], dict[str, Any], QuerySet] DEFAULT: ()

**kwargs

TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py 
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
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.

PARAMETER DESCRIPTION
*clauses

TYPE: Union[BinaryExpression, Callable[[QuerySetType], Union[BinaryExpression, Awaitable[BinaryExpression]]], dict[str, Any], QuerySet] DEFAULT: ()

**kwargs

TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py 
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
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.

PARAMETER DESCRIPTION
exclude_secrets

TYPE: bool DEFAULT: True

Source code in edgy/core/db/querysets/base.py 
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
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

batch_size 

batch_size(batch_size=None)

Set batch/chunk size. Used for iterate

PARAMETER DESCRIPTION
batch_size

TYPE: Optional[int] DEFAULT: None

Source code in edgy/core/db/querysets/base.py 
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
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

PARAMETER DESCRIPTION
term

TYPE: Any

Source code in edgy/core/db/querysets/base.py 
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
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.

PARAMETER DESCRIPTION
*order_by

TYPE: str DEFAULT: ()

Source code in edgy/core/db/querysets/base.py 
1181
1182
1183
1184
1185
1186
1187
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 
1189
1190
1191
1192
1193
1194
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.

PARAMETER DESCRIPTION
limit_count

TYPE: int

Source code in edgy/core/db/querysets/base.py 
1196
1197
1198
1199
1200
1201
1202
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 edgy/core/db/querysets/base.py 
1204
1205
1206
1207
1208
1209
1210
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 edgy/core/db/querysets/base.py 
1212
1213
1214
1215
1216
1217
1218
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.

PARAMETER DESCRIPTION
first

TYPE: Union[bool, str] DEFAULT: True

*distinct_on

TYPE: str DEFAULT: ()

Source code in edgy/core/db/querysets/base.py 
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
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.

PARAMETER DESCRIPTION
*fields

TYPE: str DEFAULT: ()

Source code in edgy/core/db/querysets/base.py 
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
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.

PARAMETER DESCRIPTION
*fields

TYPE: str DEFAULT: ()

Source code in edgy/core/db/querysets/base.py 
1251
1252
1253
1254
1255
1256
1257
1258
1259
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.

PARAMETER DESCRIPTION
*related

TYPE: str DEFAULT: ()

Source code in edgy/core/db/querysets/base.py 
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
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.

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

Source code in edgy/core/db/querysets/base.py 
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
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, CollectionsIterable):
        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.

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 edgy/core/db/querysets/base.py 
1310
1311
1312
1313
1314
1315
1316
1317
1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
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.

PARAMETER DESCRIPTION
**kwargs

TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py 
1333
1334
1335
1336
1337
1338
1339
1340
1341
1342
1343
1344
1345
1346
1347
1348
1349
1350
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 
1352
1353
1354
1355
1356
1357
1358
1359
1360
1361
1362
1363
1364
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.

PARAMETER DESCRIPTION
**kwargs

TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py 
1366
1367
1368
1369
1370
1371
1372
1373
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.

PARAMETER DESCRIPTION
**kwargs

TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py 
1375
1376
1377
1378
1379
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 
1381
1382
1383
1384
1385
1386
1387
1388
1389
1390
1391
1392
1393
1394
1395
1396
1397
1398
1399
1400
1401
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 
1403
1404
1405
1406
1407
1408
1409
1410
1411
1412
1413
1414
1415
1416
1417
1418
1419
1420
1421
1422
1423
1424
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.

PARAMETER DESCRIPTION
*args

TYPE: Any DEFAULT: ()

**kwargs

TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py 
1426
1427
1428
1429
1430
1431
1432
1433
1434
1435
1436
1437
1438
1439
1440
1441
1442
1443
1444
1445
async def create(self, *args: Any, **kwargs: Any) -> EdgyEmbedTarget:
    """
    Creates a record in a specific table.
    """
    # for tenancy
    queryset: QuerySet = self._clone()
    instance = self.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])

bulk_create async 

bulk_create(objs)

Bulk creates records in a table

PARAMETER DESCRIPTION
objs

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

Source code in edgy/core/db/querysets/base.py 
1447
1448
1449
1450
1451
1452
1453
1454
1455
1456
1457
1458
1459
1460
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
1486
1487
1488
1489
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.

PARAMETER DESCRIPTION
objs

TYPE: list[EdgyModel]

fields

TYPE: list[str]

Source code in edgy/core/db/querysets/base.py 
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
1530
1531
1532
1533
1534
1535
1536
1537
1538
1539
1540
1541
1542
1543
1544
1545
1546
1547
1548
1549
1550
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)

delete async 

delete(use_models=False)
PARAMETER DESCRIPTION
use_models

TYPE: bool DEFAULT: False

Source code in edgy/core/db/querysets/base.py 
1552
1553
1554
1555
1556
1557
1558
1559
1560
1561
1562
1563
1564
1565
1566
1567
1568
1569
1570
1571
1572
1573
1574
1575
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.

PARAMETER DESCRIPTION
**kwargs

TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py 
1577
1578
1579
1580
1581
1582
1583
1584
1585
1586
1587
1588
1589
1590
1591
1592
1593
1594
1595
1596
1597
1598
1599
1600
1601
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.

PARAMETER DESCRIPTION
defaults

TYPE: Union[dict[str, Any], Any, None] DEFAULT: None

*args

TYPE: Any DEFAULT: ()

**kwargs

TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py 
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
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.

PARAMETER DESCRIPTION
defaults

TYPE: Union[dict[str, Any], Any, None] DEFAULT: None

*args

TYPE: Any DEFAULT: ()

**kwargs

TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py 
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
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.

PARAMETER DESCRIPTION
instance

TYPE: BaseModelType

Source code in edgy/core/db/querysets/base.py 
1680
1681
1682
1683
1684
1685
1686
1687
1688
1689
1690
1691
1692
1693
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.

PARAMETER DESCRIPTION
force_rollback

TYPE: bool DEFAULT: False

**kwargs

TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py 
1695
1696
1697
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 
1704
1705
1706
async def __aiter__(self) -> AsyncIterator[Any]:
    async for value in self._execute_iterate():
        yield value