Skip to content

QuerySet class

edgy.QuerySet

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

Bases: BaseQuerySet, QuerySetProtocol

QuerySet object used for query retrieving.

PARAMETER DESCRIPTION
model_class

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

database

TYPE: Union[Database, None] DEFAULT: None

filter_clauses

TYPE: Any DEFAULT: None

or_clauses

TYPE: Any DEFAULT: None

select_related

TYPE: Any DEFAULT: None

prefetch_related

TYPE: Any DEFAULT: None

limit_count

TYPE: Any DEFAULT: None

limit_offset

TYPE: Any DEFAULT: None

order_by

TYPE: Any DEFAULT: None

group_by

TYPE: Any DEFAULT: None

distinct_on

TYPE: Any DEFAULT: None

only_fields

TYPE: Any DEFAULT: None

defer_fields

TYPE: Any DEFAULT: None

m2m_related

TYPE: Any DEFAULT: None

using_schema

TYPE: Any DEFAULT: None

table

TYPE: Any DEFAULT: None

exclude_secrets

TYPE: Any DEFAULT: False

Source code in edgy/core/db/querysets/base.py
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
def __init__(
    self,
    model_class: Union[Type["Model"], None] = None,
    database: Union["Database", None] = None,
    filter_clauses: Any = None,
    or_clauses: Any = None,
    select_related: Any = None,
    prefetch_related: Any = None,
    limit_count: Any = None,
    limit_offset: Any = None,
    order_by: Any = None,
    group_by: Any = None,
    distinct_on: Any = None,
    only_fields: Any = None,
    defer_fields: Any = None,
    m2m_related: Any = None,
    using_schema: Any = None,
    table: Any = None,
    exclude_secrets: Any = False,
) -> None:
    super().__init__(model_class=model_class)
    self.model_class = cast("Type[Model]", model_class)
    self.filter_clauses = [] if filter_clauses is None else filter_clauses
    self.or_clauses = [] if or_clauses is None else or_clauses
    self.limit_count = limit_count
    self._select_related = [] if select_related is None else select_related
    self._prefetch_related = [] if prefetch_related is None else prefetch_related
    self._offset = limit_offset
    self._order_by = [] if order_by is None else order_by
    self._group_by = [] if group_by is None else group_by
    self.distinct_on = [] if distinct_on is None else distinct_on
    self._only = [] if only_fields is None else only_fields
    self._defer = [] if defer_fields is None else defer_fields
    self._expression = None
    self._cache = None
    self._m2m_related = m2m_related  # type: ignore
    self.using_schema = using_schema
    self._exclude_secrets = exclude_secrets or False
    self.extra: Dict[str, Any] = {}

    # Making sure the queryset always starts without any schema associated unless specified
    if self.is_m2m and not self._m2m_related:
        self._m2m_related = self.model_class.meta.multi_related[0]

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

model_class instance-attribute

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

database instance-attribute

database = database

table instance-attribute

table = table

is_m2m property

is_m2m

filter_clauses instance-attribute

filter_clauses = [] if filter_clauses is None else filter_clauses

or_clauses instance-attribute

or_clauses = [] if or_clauses is None else or_clauses

limit_count instance-attribute

limit_count = limit_count
_select_related = [] if select_related is None else select_related
_prefetch_related = [] if prefetch_related is None else prefetch_related

_offset instance-attribute

_offset = limit_offset

_order_by instance-attribute

_order_by = [] if order_by is None else order_by

_group_by instance-attribute

_group_by = [] if group_by is None else group_by

distinct_on instance-attribute

distinct_on = [] if distinct_on is None else distinct_on

_only instance-attribute

_only = [] if only_fields is None else only_fields

_defer instance-attribute

_defer = [] if defer_fields is None else defer_fields

_expression instance-attribute

_expression = None

_cache instance-attribute

_cache = None

using_schema instance-attribute

using_schema = using_schema

_exclude_secrets instance-attribute

_exclude_secrets = exclude_secrets or False

extra instance-attribute

extra = {}

sql property writable

sql

execute async

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

_extract_model_references

_extract_model_references(extracted_values, model_class)

Exracts any possible model references from the EdgyModel and returns a dictionary.

PARAMETER DESCRIPTION
extracted_values

TYPE: Any

model_class

TYPE: Optional[Type[Model]]

Source code in edgy/core/utils/models.py
62
63
64
65
66
67
68
69
70
71
72
73
def _extract_model_references(
    self, extracted_values: Any, model_class: Optional[Type["Model"]]
) -> Any:
    """
    Exracts any possible model references from the EdgyModel and returns a dictionary.
    """
    model_references = {
        name: extracted_values.get(name, None)
        for name in model_class.meta.model_references.keys()  # type: ignore
        if extracted_values.get(name)
    }
    return model_references

_extract_values_from_field

_extract_values_from_field(extracted_values, model_class=None, is_update=False)

Extracts all the deffault values from the given fields and returns the raw value corresponding to each field.

PARAMETER DESCRIPTION
extracted_values

TYPE: Any

model_class

TYPE: Optional[Type[Model]] DEFAULT: None

is_update

TYPE: bool DEFAULT: False

Source code in edgy/core/utils/models.py
 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
def _extract_values_from_field(
    self,
    extracted_values: Any,
    model_class: Optional[Type["Model"]] = None,
    is_update: bool = False,
) -> Any:
    """
    Extracts all the deffault values from the given fields and returns the raw
    value corresponding to each field.
    """
    model_cls = model_class or self
    validated = {}
    for name, field in model_cls.fields.items():  # type: ignore
        if field.read_only:
            if field.has_default():
                if not is_update:
                    validated[name] = field.get_default_value()
                else:
                    # For datetimes with `auto_now` and `auto_now_add`
                    if not _has_auto_now_add(field):
                        validated[name] = field.get_default_value()
            continue

        if name not in extracted_values:
            if field.has_default():
                validated[name] = field.get_default_value()
            continue

        item = extracted_values[name]
        value = field.check(item) if hasattr(field, "check") else None
        validated[name] = value

    # Update with any ModelRef
    validated.update(self._extract_model_references(extracted_values, model_cls))
    return validated

_extract_db_fields_from_model

_extract_db_fields_from_model(model_class)

Extacts all the db fields and excludes the related_names since those are simply relations.

PARAMETER DESCRIPTION
model_class

TYPE: Type[Model]

Source code in edgy/core/utils/models.py
111
112
113
114
115
116
117
def _extract_db_fields_from_model(self, model_class: Type["Model"]) -> Dict[Any, Any]:
    """
    Extacts all the db fields and excludes the related_names since those
    are simply relations.
    """
    related_names = model_class.meta.related_names
    return {k: v for k, v in model_class.model_fields.items() if k not in related_names}

_update_auto_now_fields

_update_auto_now_fields(values, fields)

Updates the auto_now fields

PARAMETER DESCRIPTION
values

TYPE: Any

fields

TYPE: Any

Source code in edgy/core/utils/models.py
41
42
43
44
45
46
47
48
def _update_auto_now_fields(self, values: Any, fields: Any) -> Any:
    """
    Updates the `auto_now` fields
    """
    for name, field in fields.items():
        if isinstance(field, Field) and _has_auto_now(field) and _is_datetime(field):
            values[name] = field.get_default_value()  # type: ignore
    return values

_resolve_value

_resolve_value(value)
PARAMETER DESCRIPTION
value

TYPE: Any

Source code in edgy/core/utils/models.py
50
51
52
53
54
55
56
57
58
def _resolve_value(self, value: typing.Any) -> typing.Any:
    if isinstance(value, dict):
        return dumps(
            value,
            option=OPT_SERIALIZE_NUMPY | OPT_OMIT_MICROSECONDS,
        ).decode("utf-8")
    elif isinstance(value, Enum):
        return value.name
    return value
prefetch_related(*prefetch)

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

PARAMETER DESCRIPTION
*prefetch

TYPE: Prefetch DEFAULT: ()

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

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

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

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

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

using

using(schema)

Enables and switches the db schema.

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

PARAMETER DESCRIPTION
schema

TYPE: str

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

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

using_with_db

using_with_db(connection_name, schema=None)

Enables and switches the database connection.

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

PARAMETER DESCRIPTION
connection_name

TYPE: str

schema

TYPE: Optional[str] DEFAULT: None

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

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

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

_build_order_by_expression

_build_order_by_expression(order_by, expression)

Builds the order by expression

PARAMETER DESCRIPTION
order_by

TYPE: Any

expression

TYPE: Any

Source code in edgy/core/db/querysets/base.py
 97
 98
 99
100
101
def _build_order_by_expression(self, order_by: Any, expression: Any) -> Any:
    """Builds the order by expression"""
    order_by = list(map(self._prepare_order_by, order_by))
    expression = expression.order_by(*order_by)
    return expression

_build_group_by_expression

_build_group_by_expression(group_by, expression)

Builds the group by expression

PARAMETER DESCRIPTION
group_by

TYPE: Any

expression

TYPE: Any

Source code in edgy/core/db/querysets/base.py
103
104
105
106
107
def _build_group_by_expression(self, group_by: Any, expression: Any) -> Any:
    """Builds the group by expression"""
    group_by = list(map(self._prepare_group_by, group_by))
    expression = expression.group_by(*group_by)
    return expression

_build_filter_clauses_expression

_build_filter_clauses_expression(filter_clauses, expression)

Builds the filter clauses expression

PARAMETER DESCRIPTION
filter_clauses

TYPE: Any

expression

TYPE: Any

Source code in edgy/core/db/querysets/base.py
109
110
111
112
113
114
115
116
def _build_filter_clauses_expression(self, filter_clauses: Any, expression: Any) -> Any:
    """Builds the filter clauses expression"""
    if len(filter_clauses) == 1:
        clause = filter_clauses[0]
    else:
        clause = sqlalchemy.sql.and_(*filter_clauses)
    expression = expression.where(clause)
    return expression

_build_or_clauses_expression

_build_or_clauses_expression(or_clauses, expression)

Builds the filter clauses expression

PARAMETER DESCRIPTION
or_clauses

TYPE: Any

expression

TYPE: Any

Source code in edgy/core/db/querysets/base.py
118
119
120
121
122
123
124
125
def _build_or_clauses_expression(self, or_clauses: Any, expression: Any) -> Any:
    """Builds the filter clauses expression"""
    if len(or_clauses) == 1:
        clause = or_clauses[0]
    else:
        clause = sqlalchemy.sql.or_(*or_clauses)
    expression = expression.where(clause)
    return expression

_build_select_distinct

_build_select_distinct(distinct_on, expression)

Filters selects only specific fields

PARAMETER DESCRIPTION
distinct_on

TYPE: Any

expression

TYPE: Any

Source code in edgy/core/db/querysets/base.py
127
128
129
130
131
def _build_select_distinct(self, distinct_on: Any, expression: Any) -> Any:
    """Filters selects only specific fields"""
    distinct_on = list(map(self._prepare_fields_for_distinct, distinct_on))
    expression = expression.distinct(*distinct_on)
    return expression

_is_multiple_foreign_key

_is_multiple_foreign_key(model_class)

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

Returns a tuple of bool, list of tuples

PARAMETER DESCRIPTION
model_class

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

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

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

    counter = 0

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

        if counter > 1:
            has_many = True

        if isinstance(value, (BaseForeignKeyField, BaseOneToOneKeyField)):
            tablename = value.to if isinstance(value.to, str) else value.to.__name__  # type: ignore

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

    return has_many, foreign_keys

_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
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
def _build_tables_select_from_relationship(self) -> Any:
    """
    Builds the tables relationships and joins.
    When a table contains more than one foreign key pointing to the same
    destination table, a lookup for the related field is made to understand
    from which foreign key the table is looked up from.
    """
    queryset: "QuerySet" = self._clone()

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

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

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

            table = model_class.table

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

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

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

            tables.append(table)

    return tables, select_from

_validate_only_and_defer

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

_secret_recursive_names

_secret_recursive_names(model_class, columns=None)

Recursively gets the names of the fields excluding the secrets.

PARAMETER DESCRIPTION
model_class

TYPE: Any

columns

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

Source code in edgy/core/db/querysets/base.py
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
def _secret_recursive_names(
    self, model_class: Any, columns: Union[List[str], None] = None
) -> List[str]:
    """
    Recursively gets the names of the fields excluding the secrets.
    """
    if columns is None:
        columns = []

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

    columns = list(set(columns))
    return columns

_build_select

_build_select()

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

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

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

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

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

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

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

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

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

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

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

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

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

    queryset._expression = expression  # type: ignore
    return expression

_filter_query

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

TYPE: bool DEFAULT: False

or_

TYPE: bool DEFAULT: False

**kwargs

TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py
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
def _filter_query(
    self,
    exclude: bool = False,
    or_: bool = False,
    **kwargs: Any,
) -> "QuerySet":
    from edgy.core.db.models import Model

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

    # 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

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

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

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

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

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

            column = model_class.table.columns[field_name]

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

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

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

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

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

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

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

_validate_kwargs

_validate_kwargs(**kwargs)
PARAMETER DESCRIPTION
**kwargs

TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py
427
428
def _validate_kwargs(self, **kwargs: Any) -> Any:
    return self._extract_values_from_field(kwargs, model_class=self.model_class)

_prepare_order_by

_prepare_order_by(order_by)
PARAMETER DESCRIPTION
order_by

TYPE: str

Source code in edgy/core/db/querysets/base.py
430
431
432
433
434
def _prepare_order_by(self, order_by: str) -> Any:
    reverse = order_by.startswith("-")
    order_by = order_by.lstrip("-")
    order_col = self.table.columns[order_by]
    return order_col.desc() if reverse else order_col

_prepare_group_by

_prepare_group_by(group_by)
PARAMETER DESCRIPTION
group_by

TYPE: str

Source code in edgy/core/db/querysets/base.py
436
437
438
439
def _prepare_group_by(self, group_by: str) -> Any:
    group_by = group_by.lstrip("-")
    group_col = self.table.columns[group_by]
    return group_col

_prepare_fields_for_distinct

_prepare_fields_for_distinct(distinct_on)
PARAMETER DESCRIPTION
distinct_on

TYPE: str

Source code in edgy/core/db/querysets/base.py
441
442
443
def _prepare_fields_for_distinct(self, distinct_on: str) -> Any:
    _distinct_on: sqlalchemy.Column = self.table.columns[distinct_on]
    return _distinct_on

_clone

_clone()

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

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

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

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

    return queryset

__aiter__ async

__aiter__()
Source code in edgy/core/db/querysets/base.py
499
500
501
async def __aiter__(self) -> AsyncIterator[EdgyModel]:
    for value in await self:
        yield value

_set_query_expression

_set_query_expression(expression)

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

PARAMETER DESCRIPTION
expression

TYPE: Any

Source code in edgy/core/db/querysets/base.py
503
504
505
506
507
508
def _set_query_expression(self, expression: Any) -> None:
    """
    Sets the value of the sql property to the expression used.
    """
    self.sql = expression
    self.model_class.raw_query = self.sql

_filter_or_exclude

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

Filters or excludes a given clause for a specific QuerySet.

PARAMETER DESCRIPTION
clause

TYPE: Optional[BinaryExpression] DEFAULT: None

exclude

TYPE: bool DEFAULT: False

**kwargs

TYPE: Any DEFAULT: {}

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

    queryset.filter_clauses.append(clause)
    return queryset

filter

filter(clause=None, **kwargs)

Filters the QuerySet by the given kwargs and clause.

PARAMETER DESCRIPTION
clause

TYPE: Optional[BinaryExpression] DEFAULT: None

**kwargs

TYPE: Any DEFAULT: {}

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

or_

or_(clause=None, **kwargs)

Filters the QuerySet by the OR operand.

PARAMETER DESCRIPTION
clause

TYPE: Optional[BinaryExpression] DEFAULT: None

**kwargs

TYPE: Any DEFAULT: {}

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

and_

and_(clause=None, **kwargs)

Filters the QuerySet by the AND operand.

PARAMETER DESCRIPTION
clause

TYPE: Optional[BinaryExpression] DEFAULT: None

**kwargs

TYPE: Any DEFAULT: {}

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

not_

not_(clause=None, **kwargs)

Filters the QuerySet by the NOT operand.

PARAMETER DESCRIPTION
clause

TYPE: Optional[BinaryExpression] DEFAULT: None

**kwargs

TYPE: Any DEFAULT: {}

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

exclude

exclude(clause=None, **kwargs)

Exactly the same as the filter but for the exclude.

PARAMETER DESCRIPTION
clause

TYPE: Optional[BinaryExpression] DEFAULT: None

**kwargs

TYPE: Any DEFAULT: {}

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

exclude_secrets

exclude_secrets(clause=None, **kwargs)

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

PARAMETER DESCRIPTION
clause

TYPE: Optional[BinaryExpression] DEFAULT: None

**kwargs

TYPE: Any DEFAULT: {}

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

lookup

lookup(term)

Broader way of searching for a given term

PARAMETER DESCRIPTION
term

TYPE: Any

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

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

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

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

    return queryset

order_by

order_by(*order_by)

Returns a QuerySet ordered by the given fields.

PARAMETER DESCRIPTION
*order_by

TYPE: str DEFAULT: ()

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

limit

limit(limit_count)

Returns a QuerySet limited by.

PARAMETER DESCRIPTION
limit_count

TYPE: int

Source code in edgy/core/db/querysets/base.py
632
633
634
635
636
637
638
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
640
641
642
643
644
645
646
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: Sequence[str] DEFAULT: ()

Source code in edgy/core/db/querysets/base.py
648
649
650
651
652
653
654
def group_by(self, *group_by: Sequence[str]) -> "QuerySet":
    """
    Returns the values grouped by the given fields.
    """
    queryset: "QuerySet" = self._clone()
    queryset._group_by = group_by
    return queryset

distinct

distinct(*distinct_on)

Returns a queryset with distinct results.

PARAMETER DESCRIPTION
*distinct_on

TYPE: Sequence[str] DEFAULT: ()

Source code in edgy/core/db/querysets/base.py
656
657
658
659
660
661
662
def distinct(self, *distinct_on: Sequence[str]) -> "QuerySet":
    """
    Returns a queryset with distinct results.
    """
    queryset: "QuerySet" = self._clone()
    queryset.distinct_on = distinct_on
    return queryset

only

only(*fields)

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

PARAMETER DESCRIPTION
*fields

TYPE: Sequence[str] DEFAULT: ()

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

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

defer

defer(*fields)

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

PARAMETER DESCRIPTION
*fields

TYPE: Sequence[str] DEFAULT: ()

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

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

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

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

PARAMETER DESCRIPTION
related

TYPE: Any

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

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

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

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

values async

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

Returns the results in a python dictionary format.

PARAMETER DESCRIPTION
fields

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

exclude

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

exclude_none

TYPE: bool DEFAULT: False

flatten

TYPE: bool DEFAULT: False

**kwargs

TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py
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
async def values(
    self,
    fields: Union[Sequence[str], str, None] = None,
    exclude: Union[Sequence[str], Set[str]] = None,
    exclude_none: bool = False,
    flatten: bool = False,
    **kwargs: Any,
) -> List[Any]:
    """
    Returns the results in a python dictionary format.
    """
    fields = fields or []
    queryset: "QuerySet" = self._clone()
    rows: List[Type["Model"]] = await queryset.all()

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

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

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

    if not as_tuple:
        return rows

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

values_list async

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

Returns the results in a python dictionary format.

PARAMETER DESCRIPTION
fields

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

exclude

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

exclude_none

TYPE: bool DEFAULT: False

flat

TYPE: bool DEFAULT: False

Source code in edgy/core/db/querysets/base.py
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
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.
    """
    queryset: "QuerySet" = self._clone()
    fields = fields or []
    if flat and len(fields) > 1:
        raise QuerySetError(
            detail=f"Maximum of 1 in fields when `flat` is enables, got {len(fields)} instead."
        ) from None

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

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

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

exists async

exists(**kwargs)

Returns a boolean indicating if a record exists or not.

PARAMETER DESCRIPTION
**kwargs

TYPE: Any DEFAULT: {}

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

count async

count(**kwargs)

Returns an indicating the total records.

PARAMETER DESCRIPTION
**kwargs

TYPE: Any DEFAULT: {}

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

get_or_none async

get_or_none(**kwargs)

Fetch one object matching the parameters or returns None.

PARAMETER DESCRIPTION
**kwargs

TYPE: Any DEFAULT: {}

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

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

_all async

_all(**kwargs)

Executes the query.

PARAMETER DESCRIPTION
**kwargs

TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py
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
async def _all(self, **kwargs: Any) -> List[EdgyModel]:
    """
    Executes the query.
    """
    queryset: "QuerySet" = self._clone()
    if queryset.is_m2m:
        queryset.distinct_on = [queryset.m2m_related]

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

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

    rows = await queryset.database.fetch_all(expression)

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

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

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

    if not queryset.is_m2m:
        return results

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

all

all(**kwargs)

Returns the queryset records based on specific filters

PARAMETER DESCRIPTION
**kwargs

TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py
858
859
860
861
862
863
864
def all(self, **kwargs: Any) -> "QuerySet":
    """
    Returns the queryset records based on specific filters
    """
    queryset: "QuerySet" = self._clone()
    queryset.extra = kwargs
    return queryset

get async

get(**kwargs)

Returns a single record based on the given kwargs.

PARAMETER DESCRIPTION
**kwargs

TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py
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
async def get(self, **kwargs: Any) -> EdgyModel:
    """
    Returns a single record based on the given kwargs.
    """
    queryset: "QuerySet" = self._clone()

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

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

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

    if not rows:
        raise ObjectNotFound()
    if len(rows) > 1:
        raise MultipleObjectsReturned()

    return queryset.model_class.from_sqla_row(
        rows[0],
        select_related=queryset._select_related,
        is_only_fields=is_only_fields,
        only_fields=queryset._only,
        is_defer_fields=is_defer_fields,
        prefetch_related=queryset._prefetch_related,
        exclude_secrets=queryset._exclude_secrets,
        using_schema=queryset.using_schema,
    )

first async

first(**kwargs)

Returns the first record of a given queryset.

PARAMETER DESCRIPTION
**kwargs

TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py
898
899
900
901
902
903
904
905
906
907
908
909
async def first(self, **kwargs: Any) -> Union[EdgyModel, None]:
    """
    Returns the first record of a given queryset.
    """
    queryset: "QuerySet" = self._clone()
    if kwargs:
        return await queryset.filter(**kwargs).order_by("id").get()

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

last async

last(**kwargs)

Returns the last record of a given queryset.

PARAMETER DESCRIPTION
**kwargs

TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py
911
912
913
914
915
916
917
918
919
920
921
922
async def last(self, **kwargs: Any) -> Union[EdgyModel, None]:
    """
    Returns the last record of a given queryset.
    """
    queryset: "QuerySet" = self._clone()
    if kwargs:
        return await queryset.filter(**kwargs).order_by("-id").get()

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

create async

create(**kwargs)

Creates a record in a specific table.

PARAMETER DESCRIPTION
**kwargs

TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py
924
925
926
927
928
929
930
931
932
933
async def create(self, **kwargs: Any) -> EdgyModel:
    """
    Creates a record in a specific table.
    """
    queryset: "QuerySet" = self._clone()
    kwargs = queryset._validate_kwargs(**kwargs)
    instance = queryset.model_class(**kwargs)
    instance.table = queryset.table
    instance = await instance.save(force_save=True, values=kwargs)
    return instance

bulk_create async

bulk_create(objs)

Bulk creates records in a table

PARAMETER DESCRIPTION
objs

TYPE: List[Dict]

Source code in edgy/core/db/querysets/base.py
935
936
937
938
939
940
941
942
943
944
async def bulk_create(self, objs: List[Dict]) -> None:
    """
    Bulk creates records in a table
    """
    queryset: "QuerySet" = self._clone()
    new_objs = [queryset._validate_kwargs(**obj) for obj in objs]

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

bulk_update async

bulk_update(objs, fields)

Bulk updates records in a table.

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

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

PARAMETER DESCRIPTION
objs

TYPE: List[EdgyModel]

fields

TYPE: List[str]

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

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

    new_objs = [
        queryset._extract_values_from_field(obj, queryset.model_class) for obj in new_objs
    ]

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

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

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

delete async

delete()
Source code in edgy/core/db/querysets/base.py
984
985
986
987
988
989
990
991
992
993
994
995
996
async def delete(self) -> None:
    queryset: "QuerySet" = self._clone()

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

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

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

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

update async

update(**kwargs)

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

PARAMETER DESCRIPTION
**kwargs

TYPE: Any DEFAULT: {}

Source code in 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
1018
1019
1020
1021
1022
1023
async def update(self, **kwargs: Any) -> None:
    """
    Updates a record in a specific table with the given kwargs.
    """
    queryset: "QuerySet" = self._clone()

    extracted_fields = queryset._extract_values_from_field(
        kwargs, model_class=queryset.model_class
    )
    kwargs = queryset._update_auto_now_fields(extracted_fields, queryset.model_class.fields)

    # Broadcast the initial update details
    await self.model_class.signals.pre_update.send(
        sender=self.__class__, instance=self, kwargs=kwargs
    )

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

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

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

    # Broadcast the update executed
    await self.model_class.signals.post_update.send(sender=self.__class__, instance=self)

get_or_create async

get_or_create(defaults, **kwargs)

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

PARAMETER DESCRIPTION
defaults

TYPE: Dict[str, Any]

**kwargs

TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
async def get_or_create(
    self, defaults: Dict[str, Any], **kwargs: Any
) -> Tuple[EdgyModel, bool]:
    """
    Creates a record in a specific table or updates if already exists.
    """
    queryset: "QuerySet" = self._clone()

    try:
        instance = await queryset.get(**kwargs)
        return instance, False
    except ObjectNotFound:
        kwargs.update(defaults)
        instance = await queryset.create(**kwargs)
        return instance, True

update_or_create async

update_or_create(defaults, **kwargs)

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

PARAMETER DESCRIPTION
defaults

TYPE: Dict[str, Any]

**kwargs

TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
async def update_or_create(
    self, defaults: Dict[str, Any], **kwargs: Any
) -> Tuple[EdgyModel, bool]:
    """
    Updates a record in a specific table or creates a new one.
    """
    queryset: "QuerySet" = self._clone()
    try:
        instance = await queryset.get(**kwargs)
        await instance.update(**defaults)
        return instance, False
    except ObjectNotFound:
        kwargs.update(defaults)
        instance = await queryset.create(**kwargs)
        return instance, True

contains async

contains(instance)

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

PARAMETER DESCRIPTION
instance

TYPE: EdgyModel

Source code in edgy/core/db/querysets/base.py
1057
1058
1059
1060
1061
1062
1063
1064
1065
async def contains(self, instance: EdgyModel) -> bool:
    """Returns true if the QuerySet contains the provided object.
    False if otherwise.
    """
    queryset: "QuerySet" = self._clone()

    if getattr(instance, "pk", None) is None:
        raise ValueError("'obj' must be a model or reflect model instance.")
    return await queryset.filter(pk=instance.pk).exists()

_execute async

_execute()
Source code in edgy/core/db/querysets/base.py
1067
1068
1069
1070
async def _execute(self) -> Any:
    queryset: "QuerySet" = self._clone()
    records = await queryset._all(**queryset.extra)
    return records