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 
 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
175
176
177
178
179
180
181
182
183
184
185
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:
    super().__init__(model_class=model_class)
    self.filter_clauses: list[Any] = list(filter_clauses)
    self.or_clauses: list[Any] = []
    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[
            str,
            dict[str, tuple[sqlalchemy.Table, type[BaseModelType]]],
            dict[str, set[str]],
            Any,
        ]
    ] = 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 = []

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.

build_where_clause async 

build_where_clause(_=None)

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

PARAMETER DESCRIPTION
_

TYPE: Any DEFAULT: None

Source code in edgy/core/db/querysets/base.py 
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
async def build_where_clause(self, _: Any = None) -> Any:
    """Build a where clause from the filters which can be passed in a where function."""
    # ignored args for passing build_where_clause in filter_clauses
    where_clause: list[Any] = []
    if self.or_clauses:
        or_clauses = await self._resolve_clause_args(self.or_clauses)
        where_clause.append(
            or_clauses[0] if len(or_clauses) == 1 else clauses_mod.or_(*or_clauses)
        )

    if self.filter_clauses:
        # we AND by default
        where_clause.extend(await self._resolve_clause_args(self.filter_clauses))
    # for nicer unpacking
    return clauses_mod.and_(*where_clause)

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
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
        queryset.active_schema = queryset.get_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 
61
62
63
64
65
66
67
68
69
70
71
72
73
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 
106
107
108
109
110
111
112
113
114
115
116
117
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 
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
def _clone(self) -> "QuerySet":
    """
    Return a copy of the current QuerySet that's ready for another
    operation.
    """
    queryset = self.__class__(
        self.model_class,
        database=self.database,
        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 = list(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 
216
217
218
219
220
221
222
223
224
225
226
227
228
229
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 
231
232
233
234
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 
236
237
238
239
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

_resolve_clause_args async 

_resolve_clause_args(args)
PARAMETER DESCRIPTION
args

TYPE: Any

Source code in edgy/core/db/querysets/base.py 
241
242
243
244
245
246
247
248
async def _resolve_clause_args(self, args: Any) -> Any:
    result: list[Any] = []
    for arg in args:
        result.append(_parse_clause_arg(arg, self))
    if self.database.force_rollback:
        return [await el for el in result]
    else:
        return await asyncio.gather(*result)

_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 
266
267
268
269
270
271
272
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, transitions_is_full_outer)
PARAMETER DESCRIPTION
join_clause

TYPE: Any

current_transition

TYPE: tuple[str, str]

transitions

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

tables_and_models

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

transitions_is_full_outer

TYPE: dict[tuple[str, str], bool]

Source code in edgy/core/db/querysets/base.py 
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
@classmethod
def _join_table_helper(
    cls,
    join_clause: Any,
    current_transition: tuple[str, str],
    *,
    transitions: dict[tuple[str, str], tuple[Any, set[tuple[str, str]]]],
    tables_and_models: dict[str, tuple["sqlalchemy.Table", type["BaseModelType"]]],
    transitions_is_full_outer: dict[tuple[str, str], bool],
) -> Any:
    if current_transition not in transitions:
        return join_clause
    transition_value = transitions.pop(current_transition)

    for dep in transition_value[1]:
        join_clause = cls._join_table_helper(
            join_clause,
            dep,
            transitions=transitions,
            tables_and_models=tables_and_models,
            transitions_is_full_outer=transitions_is_full_outer,
        )

    return sqlalchemy.sql.join(
        join_clause,
        tables_and_models[current_transition[1]][0],
        transition_value[0],
        isouter=True,
        full=transitions_is_full_outer.get(current_transition, False),
    )

_build_tables_select_from_relationship 

_build_tables_select_from_relationship()

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

Source code in edgy/core/db/querysets/base.py 
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
443
444
445
446
447
448
449
450
451
452
453
def _build_tables_select_from_relationship(
    self,
) -> tuple[
    str, dict[str, tuple["sqlalchemy.Table", type["BaseModelType"]]], dict[str, set[str]], Any
]:
    """
    Builds the tables relationships and joins.
    When a table contains more than one foreign key pointing to the same
    destination table, a lookup for the related field is made to understand
    from which foreign key the table is looked up from.
    """

    # How does this work?
    # First we build a transitions table with dependencies in case multiple pathes to the same table exist
    # Secondly we check if a select_related path is joining a table from the set in an opposite direction
    # If yes, we mark the transition for a full outer join (dangerous, there could be side-effects)
    # 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
        maintablekey = maintable.key
        tables_and_models: dict[str, tuple[sqlalchemy.Table, type[BaseModelType]]] = {
            select_from.key: (select_from, self.model_class)
        }
        prefixes: dict[str, set[str]] = defaultdict(set)
        transitions: dict[tuple[str, str], tuple[Any, set[tuple[str, str]]]] = {}
        transitions_is_full_outer: dict[tuple[str, str], bool] = {}

        # 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 = ""
            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
                table = model_class.table_schema(self.active_schema)
                # use table from tables_and_models
                if table.key in tables_and_models:
                    table = tables_and_models[table.key][0]

                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
                    table = model_class.table_schema(self.active_schema)
                    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
                prefix = f"{prefix}__{field_name}" if prefix else f"{prefix}"
                prefixes[table.key].add(prefix)
                transition_key = (former_table.key, table.key)
                if transition_key in transitions:
                    # can not provide new informations
                    former_table = table
                    former_transition = transition_key
                    continue
                and_clause = clauses_mod.and_(
                    *self._select_from_relationship_clause_generator(
                        foreign_key, table, reverse, former_table
                    )
                )
                if (table.key, former_table.key) in transitions:
                    _transition_key = (table.key, former_table.key)
                    # inverted
                    # only make full outer when not the main query
                    if former_table.key != maintablekey:
                        transitions_is_full_outer[_transition_key] = True
                    transitions[_transition_key] = (
                        clauses_mod.or_(transitions[_transition_key][0], and_clause),
                        {*transitions[_transition_key][1], former_transition}
                        if former_transition
                        else transitions[_transition_key][1],
                    )
                elif table.key in tables_and_models:
                    for _transition_key in transitions:
                        if _transition_key[1] == table.key:
                            break
                    else:
                        # this should never happen
                        raise Exception("transition not found despite in tables_and_models")
                    transitions[_transition_key] = (
                        clauses_mod.or_(and_clause, transitions[_transition_key][0]),
                        {*transitions[_transition_key][1], former_transition}
                        if former_transition
                        else transitions[_transition_key][0],
                    )
                else:
                    transitions[(former_table.key, table.key)] = (
                        and_clause,
                        {former_transition} if former_transition else set(),
                    )
                tables_and_models[table.key] = 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=tables_and_models,
                transitions_is_full_outer=transitions_is_full_outer,
            )
        self._cached_select_related_expression = (
            maintablekey,
            tables_and_models,
            prefixes,
            select_from,
        )
    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 
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
@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 
473
474
475
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 
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
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
async def _as_select_with_tables(
    self,
) -> tuple[Any, dict[str, tuple["sqlalchemy.Table", type["BaseModelType"]]]]:
    """
    Builds the query select based on the given parameters and filters.
    """
    queryset: BaseQuerySet = self

    queryset._validate_only_and_defer()
    maintable, tables_and_models, prefixes, select_from = (
        queryset._build_tables_select_from_relationship()
    )
    columns = []
    for tablekey, (table, model_class) in tables_and_models.items():
        if tablekey == maintable:
            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 queryset._only and field_name not in queryset._only:
                    continue
                if queryset._defer and field_name in queryset._defer:
                    continue
                if (
                    queryset._exclude_secrets
                    and field_name in model_class.meta.fields
                    and model_class.meta.fields[field_name].secret
                ):
                    continue

                # columns.append(column.label(f"{table.key.replace(".", "_")}_{column.key}"))
                columns.append(column)

        else:
            prefixes_for_table = prefixes[table.key]
            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 queryset._only and all(
                    f"{prefix}" not in queryset._only
                    and f"{prefix}__{field_name}" not in queryset._only
                    for prefix in prefixes_for_table
                ):
                    continue
                if queryset._defer and any(
                    f"{prefix}" in queryset._defer
                    or f"{prefix}__{field_name}" in queryset._defer
                    for prefix in prefixes_for_table
                ):
                    continue
                if (
                    queryset._exclude_secrets
                    and field_name in model_class.meta.fields
                    and model_class.meta.fields[field_name].secret
                ):
                    continue
                # columns.append(column.label(f"{table.key.replace(".", "_")}_{column.key}"))
                columns.append(column)
    assert columns, "no columns specified"
    # all columns are aliased now
    expression = sqlalchemy.sql.select(*columns).set_label_style(
        sqlalchemy.LABEL_STYLE_TABLENAME_PLUS_COL
    )
    expression = expression.select_from(select_from)
    expression = expression.where(await queryset.build_where_clause())

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

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

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

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

    if queryset.distinct_on is not None:
        expression = queryset._build_select_distinct(
            queryset.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 
564
565
566
567
568
569
570
571
572
async def as_select_with_tables(
    self,
) -> tuple[Any, dict[str, tuple["sqlalchemy.Table", type["BaseModelType"]]]]:
    """
    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 
574
575
576
577
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 
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
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
def _kwargs_to_clauses(
    self,
    kwargs: Any,
) -> tuple[list[Any], set[str]]:
    clauses = []
    select_related: set[str] = set()

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

    cleaned_kwargs = 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, generic_field)
        if cross_db_remainder:
            assert field is not 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",
                _field: "BaseFieldType" = field,
                _sub_query: "QuerySet" = sub_query,
            ) -> Any:
                fk_tuple = sqlalchemy.tuple_(
                    *(
                        getattr(queryset.table.columns, colname)
                        for colname in _field.get_column_names()
                    )
                )
                return fk_tuple.in_(await _sub_query)

            clauses.append(wrapper)
        elif callable(value):
            # bind local vars
            async def wrapper(
                queryset: "QuerySet",
                _field: "BaseFieldType" = field,
                _value: Any = value,
                _op: Optional[str] = op,
            ) -> Any:
                _value = _value(queryset)
                if isawaitable(_value):
                    _value = await _value
                return _field.operator_to_clause(
                    _field.name,
                    _op,
                    queryset.model_class.table_schema(queryset.active_schema),
                    _value,
                )

            clauses.append(wrapper)

        else:
            assert not isinstance(
                value, BaseModelType
            ), f"should be parsed in clean: {key}: {value}"
            clauses.append(
                field.operator_to_clause(
                    field_name, op, model_class.table_schema(self.active_schema), value
                )
            )

    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 
658
659
660
661
662
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 
664
665
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 
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
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 
687
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
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 
718
719
720
721
722
723
724
725
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 
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
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
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] = []

    clauses = []
    for pkcol in self.model_class.pkcolumns:
        clauses.append(
            getattr(self.table.columns, pkcol).in_(
                [
                    row._mapping[f"{self.table.key.replace('.', '_', 1)}_{pkcol}"]
                    for row in batch
                ]
            )
        )
    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."
            )
        prefetch_queryset: Optional[QuerySet] = prefetch.queryset
        if prefetch_queryset is None:
            if crawl_result.reverse_path is False:
                prefetch_queryset = self.model_class.query.filter(*clauses)
            else:
                prefetch_queryset = crawl_result.model_class.query.filter(*clauses)
        else:
            prefetch_queryset = prefetch_queryset.filter(*clauses)

        if prefetch_queryset.model_class == self.model_class:
            # queryset is of this model
            prefetch_queryset = prefetch_queryset.select_related(prefetch.related_name)
            prefetch_queryset.embed_parent = (prefetch.related_name, "")
        elif crawl_result.reverse_path is False:
            QuerySetError(
                detail=(
                    f"Creating a reverse path is not possible, unidirectional fields used."
                    f"You may want to use as queryset a queryset of model class {self.model_class!r}."
                )
            )
        else:
            # queryset is of the target model
            prefetch_queryset = prefetch_queryset.select_related(crawl_result.reverse_path)
        new_prefetch = Prefetch(
            related_name=prefetch.related_name,
            to_attr=prefetch.to_attr,
            queryset=prefetch_queryset,
        )
        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 
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
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
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)
    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 
880
881
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[[QueryType], 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 
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
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
def _filter_or_exclude(
    self,
    kwargs: Any,
    clauses: Sequence[
        Union[
            "sqlalchemy.sql.expression.BinaryExpression",
            Callable[
                ["QueryType"],
                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[
                [QueryType],
                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:

                async def wrapper_and(
                    queryset: "QuerySet",
                    _extracted_clauses: Sequence[
                        Union[
                            "sqlalchemy.sql.expression.BinaryExpression",
                            Callable[
                                ["QueryType"],
                                Union[
                                    "sqlalchemy.sql.expression.BinaryExpression",
                                    Awaitable["sqlalchemy.sql.expression.BinaryExpression"],
                                ],
                            ],
                        ]
                    ] = extracted_clauses,
                ) -> Any:
                    return clauses_mod.and_(
                        *(await self._resolve_clause_args(_extracted_clauses))
                    )

                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 not converted_clauses:
        return queryset

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

        async def wrapper(queryset: "QuerySet") -> Any:
            return clauses_mod.not_(op(*(await self._resolve_clause_args(converted_clauses))))

        queryset.filter_clauses.append(wrapper)
    elif or_:

        async def wrapper(queryset: "QuerySet") -> Any:
            return clauses_mod.or_(*(await self._resolve_clause_args(converted_clauses)))

        queryset.filter_clauses.append(wrapper)
    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 
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
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 
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
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)
    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")

filter 

filter(*clauses, **kwargs)

Filters the QuerySet by the given kwargs and clauses.

PARAMETER DESCRIPTION
*clauses

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

**kwargs

TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py 
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
def filter(
    self,
    *clauses: Union[
        "sqlalchemy.sql.expression.BinaryExpression",
        Callable[
            ["QueryType"],
            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 
1069
1070
1071
1072
1073
1074
1075
1076
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[[QueryType], Union[BinaryExpression, Awaitable[BinaryExpression]]], dict[str, Any], QuerySet] DEFAULT: ()

**kwargs

TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py 
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
def or_(
    self,
    *clauses: Union[
        "sqlalchemy.sql.expression.BinaryExpression",
        Callable[
            ["QueryType"],
            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)

and_ 

and_(*clauses, **kwargs)

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

PARAMETER DESCRIPTION
*clauses

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

**kwargs

TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py 
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
def and_(
    self,
    *clauses: Union[
        "sqlalchemy.sql.expression.BinaryExpression",
        Callable[
            ["QueryType"],
            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[[QueryType], Union[BinaryExpression, Awaitable[BinaryExpression]]], dict[str, Any], QuerySet] DEFAULT: ()

**kwargs

TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py 
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
def not_(
    self,
    *clauses: Union[
        "sqlalchemy.sql.expression.BinaryExpression",
        Callable[
            ["QueryType"],
            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[[QueryType], Union[BinaryExpression, Awaitable[BinaryExpression]]], dict[str, Any], QuerySet] DEFAULT: ()

**kwargs

TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py 
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
def exclude(
    self,
    *clauses: Union[
        "sqlalchemy.sql.expression.BinaryExpression",
        Callable[
            ["QueryType"],
            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 
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
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 
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
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 
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
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 
1208
1209
1210
1211
1212
1213
1214
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 
1216
1217
1218
1219
1220
1221
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 
1223
1224
1225
1226
1227
1228
1229
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 
1231
1232
1233
1234
1235
1236
1237
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 
1239
1240
1241
1242
1243
1244
1245
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 
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
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 
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
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 
1278
1279
1280
1281
1282
1283
1284
1285
1286
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 
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
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 
1310
1311
1312
1313
1314
1315
1316
1317
1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
1334
1335
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 
1337
1338
1339
1340
1341
1342
1343
1344
1345
1346
1347
1348
1349
1350
1351
1352
1353
1354
1355
1356
1357
1358
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 
1360
1361
1362
1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
1373
1374
1375
1376
1377
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 
1379
1380
1381
1382
1383
1384
1385
1386
1387
1388
1389
1390
1391
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 
1393
1394
1395
1396
1397
1398
1399
1400
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 
1402
1403
1404
1405
1406
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 
1408
1409
1410
1411
1412
1413
1414
1415
1416
1417
1418
1419
1420
1421
1422
1423
1424
1425
1426
1427
1428
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 
1430
1431
1432
1433
1434
1435
1436
1437
1438
1439
1440
1441
1442
1443
1444
1445
1446
1447
1448
1449
1450
1451
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 
1453
1454
1455
1456
1457
1458
1459
1460
1461
1462
1463
1464
1465
1466
1467
1468
1469
1470
1471
1472
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 
1474
1475
1476
1477
1478
1479
1480
1481
1482
1483
1484
1485
1486
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
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 
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
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
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 
1579
1580
1581
1582
1583
1584
1585
1586
1587
1588
1589
1590
1591
1592
1593
1594
1595
1596
1597
1598
1599
1600
1601
1602
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 
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
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 
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
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 
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
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 
1707
1708
1709
1710
1711
1712
1713
1714
1715
1716
1717
1718
1719
1720
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 
1722
1723
1724
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 
1731
1732
1733
async def __aiter__(self) -> AsyncIterator[Any]:
    async for value in self._execute_iterate():
        yield value