Skip to content

model_row

ModelRow

Bases: NewBaseModel

Source code in ormar\models\model_row.py
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
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
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
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
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
class ModelRow(NewBaseModel):
    @classmethod
    def from_row(  # noqa: CFQ002
        cls,
        row: ResultProxy,
        source_model: Type["Model"],
        select_related: Optional[List] = None,
        related_models: Any = None,
        related_field: Optional["ForeignKeyField"] = None,
        excludable: Optional[ExcludableItems] = None,
        current_relation_str: str = "",
        proxy_source_model: Optional[Type["Model"]] = None,
        used_prefixes: Optional[List[str]] = None,
    ) -> Optional["Model"]:
        """
        Model method to convert raw sql row from database into ormar.Model instance.
        Traverses nested models if they were specified in select_related for query.

        Called recurrently and returns model instance if it's present in the row.
        Note that it's processing one row at a time, so if there are duplicates of
        parent row that needs to be joined/combined
        (like parent row in sql join with 2+ child rows)
        instances populated in this method are later combined in the QuerySet.
        Other method working directly on raw database results is in prefetch_query,
        where rows are populated in a different way as they do not have
        nested models in result.

        :param used_prefixes: list of already extracted prefixes
        :type used_prefixes: List[str]
        :param proxy_source_model: source model from which querysetproxy is constructed
        :type proxy_source_model: Optional[Type["ModelRow"]]
        :param excludable: structure of fields to include and exclude
        :type excludable: ExcludableItems
        :param current_relation_str: name of the relation field
        :type current_relation_str: str
        :param source_model: model on which relation was defined
        :type source_model: Type[Model]
        :param row: raw result row from the database
        :type row: ResultProxy
        :param select_related: list of names of related models fetched from database
        :type select_related: List
        :param related_models: list or dict of related models
        :type related_models: Union[List, Dict]
        :param related_field: field with relation declaration
        :type related_field: ForeignKeyField
        :return: returns model if model is populated from database
        :rtype: Optional[Model]
        """
        item: Dict[str, Any] = {}
        select_related = select_related or []
        related_models = related_models or []
        table_prefix = ""
        used_prefixes = used_prefixes if used_prefixes is not None else []
        excludable = excludable or ExcludableItems()

        if select_related:
            related_models = group_related_list(select_related)

        if related_field:
            table_prefix = cls._process_table_prefix(
                source_model=source_model,
                current_relation_str=current_relation_str,
                related_field=related_field,
                used_prefixes=used_prefixes,
            )

        item = cls._populate_nested_models_from_row(
            item=item,
            row=row,
            related_models=related_models,
            excludable=excludable,
            current_relation_str=current_relation_str,
            source_model=source_model,  # type: ignore
            proxy_source_model=proxy_source_model,  # type: ignore
            table_prefix=table_prefix,
            used_prefixes=used_prefixes,
        )
        item = cls.extract_prefixed_table_columns(
            item=item, row=row, table_prefix=table_prefix, excludable=excludable
        )

        instance: Optional["Model"] = None
        if item.get(cls.ormar_config.pkname, None) is not None:
            item["__excluded__"] = cls.get_names_to_exclude(
                excludable=excludable, alias=table_prefix
            )
            instance = cast("Model", cls(**item))
            instance.set_save_status(True)
        return instance

    @classmethod
    def _process_table_prefix(
        cls,
        source_model: Type["Model"],
        current_relation_str: str,
        related_field: "ForeignKeyField",
        used_prefixes: List[str],
    ) -> str:
        """

        :param source_model: model on which relation was defined
        :type source_model: Type[Model]
        :param current_relation_str: current relation string
        :type current_relation_str: str
        :param related_field: field with relation declaration
        :type related_field: "ForeignKeyField"
        :param used_prefixes: list of already extracted prefixes
        :type used_prefixes: List[str]
        :return: table_prefix to use
        :rtype: str
        """
        if related_field.is_multi:
            previous_model = related_field.through
        else:
            previous_model = related_field.owner
        table_prefix = cls.ormar_config.alias_manager.resolve_relation_alias(
            from_model=previous_model, relation_name=related_field.name
        )
        if not table_prefix or table_prefix in used_prefixes:
            manager = cls.ormar_config.alias_manager
            table_prefix = manager.resolve_relation_alias_after_complex(
                source_model=source_model,
                relation_str=current_relation_str,
                relation_field=related_field,
            )
        used_prefixes.append(table_prefix)
        return table_prefix

    @classmethod
    def _populate_nested_models_from_row(  # noqa: CFQ002
        cls,
        item: dict,
        row: ResultProxy,
        source_model: Type["Model"],
        related_models: Any,
        excludable: ExcludableItems,
        table_prefix: str,
        used_prefixes: List[str],
        current_relation_str: Optional[str] = None,
        proxy_source_model: Optional[Type["Model"]] = None,
    ) -> dict:
        """
        Traverses structure of related models and populates the nested models
        from the database row.
        Related models can be a list if only directly related models are to be
        populated, converted to dict if related models also have their own related
        models to be populated.

        Recurrently calls from_row method on nested instances and create nested
        instances. In the end those instances are added to the final model dictionary.

        :param proxy_source_model: source model from which querysetproxy is constructed
        :type proxy_source_model: Optional[Type["ModelRow"]]
        :param excludable: structure of fields to include and exclude
        :type excludable: ExcludableItems
        :param source_model: source model from which relation started
        :type source_model: Type[Model]
        :param current_relation_str: joined related parts into one string
        :type current_relation_str: str
        :param item: dictionary of already populated nested models, otherwise empty dict
        :type item: Dict
        :param row: raw result row from the database
        :type row: ResultProxy
        :param related_models: list or dict of related models
        :type related_models: Union[Dict, List]
        :return: dictionary with keys corresponding to model fields names
        and values are database values
        :rtype: Dict
        """

        for related in related_models:
            field = cls.ormar_config.model_fields[related]
            field = cast("ForeignKeyField", field)
            model_cls = field.to
            model_excludable = excludable.get(
                model_cls=cast(Type["Model"], cls), alias=table_prefix
            )
            if model_excludable.is_excluded(related):
                continue

            relation_str, remainder = cls._process_remainder_and_relation_string(
                related_models=related_models,
                current_relation_str=current_relation_str,
                related=related,
            )
            child = model_cls.from_row(
                row,
                related_models=remainder,
                related_field=field,
                excludable=excludable,
                current_relation_str=relation_str,
                source_model=source_model,
                proxy_source_model=proxy_source_model,
                used_prefixes=used_prefixes,
            )
            item[model_cls.get_column_name_from_alias(related)] = child
            if (
                field.is_multi
                and child
                and not model_excludable.is_excluded(field.through.get_name())
            ):
                cls._populate_through_instance(
                    row=row,
                    item=item,
                    related=related,
                    excludable=excludable,
                    child=child,
                    proxy_source_model=proxy_source_model,
                )

        return item

    @staticmethod
    def _process_remainder_and_relation_string(
        related_models: Union[Dict, List],
        current_relation_str: Optional[str],
        related: str,
    ) -> Tuple[str, Optional[Union[Dict, List]]]:
        """
        Process remainder models and relation string

        :param related_models: list or dict of related models
        :type related_models: Union[Dict, List]
        :param current_relation_str: current relation string
        :type current_relation_str: Optional[str]
        :param related: name of the relation
        :type related: str
        """
        relation_str = (
            "__".join([current_relation_str, related])
            if current_relation_str
            else related
        )

        remainder = None
        if isinstance(related_models, dict) and related_models[related]:
            remainder = related_models[related]
        return relation_str, remainder

    @classmethod
    def _populate_through_instance(  # noqa: CFQ002
        cls,
        row: ResultProxy,
        item: Dict,
        related: str,
        excludable: ExcludableItems,
        child: "Model",
        proxy_source_model: Optional[Type["Model"]],
    ) -> None:
        """
        Populates the through model on reverse side of current query.
        Normally it's child class, unless the query is from queryset.

        :param row: row from db result
        :type row: ResultProxy
        :param item: parent item dict
        :type item: Dict
        :param related: current relation name
        :type related: str
        :param excludable: structure of fields to include and exclude
        :type excludable: ExcludableItems
        :param child: child item of parent
        :type child: "Model"
        :param proxy_source_model: source model from which querysetproxy is constructed
        :type proxy_source_model: Type["Model"]
        """
        through_name = cls.ormar_config.model_fields[related].through.get_name()
        through_child = cls._create_through_instance(
            row=row, related=related, through_name=through_name, excludable=excludable
        )

        if child.__class__ != proxy_source_model:
            setattr(child, through_name, through_child)
        else:
            item[through_name] = through_child
        child.set_save_status(True)

    @classmethod
    def _create_through_instance(
        cls,
        row: ResultProxy,
        through_name: str,
        related: str,
        excludable: ExcludableItems,
    ) -> "ModelRow":
        """
        Initialize the through model from db row.
        Excluded all relation fields and other exclude/include set in excludable.

        :param row: loaded row from database
        :type row: sqlalchemy.engine.ResultProxy
        :param through_name: name of the through field
        :type through_name: str
        :param related: name of the relation
        :type related: str
        :param excludable: structure of fields to include and exclude
        :type excludable: ExcludableItems
        :return: initialized through model without relation
        :rtype: "ModelRow"
        """
        model_cls = cls.ormar_config.model_fields[through_name].to
        table_prefix = cls.ormar_config.alias_manager.resolve_relation_alias(
            from_model=cls, relation_name=related
        )
        # remove relations on through field
        model_excludable = excludable.get(model_cls=model_cls, alias=table_prefix)
        model_excludable.set_values(
            value=model_cls.extract_related_names(), is_exclude=True
        )
        child_dict = model_cls.extract_prefixed_table_columns(
            item={}, row=row, excludable=excludable, table_prefix=table_prefix
        )
        child_dict["__excluded__"] = model_cls.get_names_to_exclude(
            excludable=excludable, alias=table_prefix
        )
        child = model_cls(**child_dict)  # type: ignore
        return child

    @classmethod
    def extract_prefixed_table_columns(
        cls,
        item: dict,
        row: ResultProxy,
        table_prefix: str,
        excludable: ExcludableItems,
    ) -> Dict:
        """
        Extracts own fields from raw sql result, using a given prefix.
        Prefix changes depending on the table's position in a join.

        If the table is a main table, there is no prefix.
        All joined tables have prefixes to allow duplicate column names,
        as well as duplicated joins to the same table from multiple different tables.

        Extracted fields populates the related dict later used to construct a Model.

        Used in Model.from_row and PrefetchQuery._populate_rows methods.

        :param excludable: structure of fields to include and exclude
        :type excludable: ExcludableItems
        :param item: dictionary of already populated nested models, otherwise empty dict
        :type item: Dict
        :param row: raw result row from the database
        :type row: sqlalchemy.engine.result.ResultProxy
        :param table_prefix: prefix of the table from AliasManager
        each pair of tables have own prefix (two of them depending on direction) -
        used in joins to allow multiple joins to the same table.
        :type table_prefix: str
        :return: dictionary with keys corresponding to model fields names
        and values are database values
        :rtype: Dict
        """
        selected_columns = cls.own_table_columns(
            model=cls, excludable=excludable, alias=table_prefix, use_alias=False
        )

        column_prefix = table_prefix + "_" if table_prefix else ""
        for column in cls.ormar_config.table.columns:
            alias = cls.get_column_name_from_alias(column.name)
            if alias not in item and alias in selected_columns:
                prefixed_name = f"{column_prefix}{column.name}"
                item[alias] = row[prefixed_name]

        return item

extract_prefixed_table_columns(item, row, table_prefix, excludable) classmethod

Extracts own fields from raw sql result, using a given prefix. Prefix changes depending on the table's position in a join.

If the table is a main table, there is no prefix. All joined tables have prefixes to allow duplicate column names, as well as duplicated joins to the same table from multiple different tables.

Extracted fields populates the related dict later used to construct a Model.

Used in Model.from_row and PrefetchQuery._populate_rows methods.

Parameters:

Name Type Description Default
excludable ExcludableItems

structure of fields to include and exclude

required
item dict

dictionary of already populated nested models, otherwise empty dict

required
row Row

raw result row from the database

required
table_prefix str

prefix of the table from AliasManager each pair of tables have own prefix (two of them depending on direction) - used in joins to allow multiple joins to the same table.

required

Returns:

Type Description
Dict

dictionary with keys corresponding to model fields names and values are database values

Source code in ormar\models\model_row.py
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
@classmethod
def extract_prefixed_table_columns(
    cls,
    item: dict,
    row: ResultProxy,
    table_prefix: str,
    excludable: ExcludableItems,
) -> Dict:
    """
    Extracts own fields from raw sql result, using a given prefix.
    Prefix changes depending on the table's position in a join.

    If the table is a main table, there is no prefix.
    All joined tables have prefixes to allow duplicate column names,
    as well as duplicated joins to the same table from multiple different tables.

    Extracted fields populates the related dict later used to construct a Model.

    Used in Model.from_row and PrefetchQuery._populate_rows methods.

    :param excludable: structure of fields to include and exclude
    :type excludable: ExcludableItems
    :param item: dictionary of already populated nested models, otherwise empty dict
    :type item: Dict
    :param row: raw result row from the database
    :type row: sqlalchemy.engine.result.ResultProxy
    :param table_prefix: prefix of the table from AliasManager
    each pair of tables have own prefix (two of them depending on direction) -
    used in joins to allow multiple joins to the same table.
    :type table_prefix: str
    :return: dictionary with keys corresponding to model fields names
    and values are database values
    :rtype: Dict
    """
    selected_columns = cls.own_table_columns(
        model=cls, excludable=excludable, alias=table_prefix, use_alias=False
    )

    column_prefix = table_prefix + "_" if table_prefix else ""
    for column in cls.ormar_config.table.columns:
        alias = cls.get_column_name_from_alias(column.name)
        if alias not in item and alias in selected_columns:
            prefixed_name = f"{column_prefix}{column.name}"
            item[alias] = row[prefixed_name]

    return item

from_row(row, source_model, select_related=None, related_models=None, related_field=None, excludable=None, current_relation_str='', proxy_source_model=None, used_prefixes=None) classmethod

Model method to convert raw sql row from database into ormar.Model instance. Traverses nested models if they were specified in select_related for query.

Called recurrently and returns model instance if it's present in the row. Note that it's processing one row at a time, so if there are duplicates of parent row that needs to be joined/combined (like parent row in sql join with 2+ child rows) instances populated in this method are later combined in the QuerySet. Other method working directly on raw database results is in prefetch_query, where rows are populated in a different way as they do not have nested models in result.

Parameters:

Name Type Description Default
used_prefixes Optional[List[str]]

list of already extracted prefixes

None
proxy_source_model Optional[Type[Model]]

source model from which querysetproxy is constructed

None
excludable Optional[ExcludableItems]

structure of fields to include and exclude

None
current_relation_str str

name of the relation field

''
source_model Type[Model]

model on which relation was defined

required
row Row

raw result row from the database

required
select_related Optional[List]

list of names of related models fetched from database

None
related_models Any

list or dict of related models

None
related_field Optional[ForeignKeyField]

field with relation declaration

None

Returns:

Type Description
Optional[Model]

returns model if model is populated from database

Source code in ormar\models\model_row.py
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
@classmethod
def from_row(  # noqa: CFQ002
    cls,
    row: ResultProxy,
    source_model: Type["Model"],
    select_related: Optional[List] = None,
    related_models: Any = None,
    related_field: Optional["ForeignKeyField"] = None,
    excludable: Optional[ExcludableItems] = None,
    current_relation_str: str = "",
    proxy_source_model: Optional[Type["Model"]] = None,
    used_prefixes: Optional[List[str]] = None,
) -> Optional["Model"]:
    """
    Model method to convert raw sql row from database into ormar.Model instance.
    Traverses nested models if they were specified in select_related for query.

    Called recurrently and returns model instance if it's present in the row.
    Note that it's processing one row at a time, so if there are duplicates of
    parent row that needs to be joined/combined
    (like parent row in sql join with 2+ child rows)
    instances populated in this method are later combined in the QuerySet.
    Other method working directly on raw database results is in prefetch_query,
    where rows are populated in a different way as they do not have
    nested models in result.

    :param used_prefixes: list of already extracted prefixes
    :type used_prefixes: List[str]
    :param proxy_source_model: source model from which querysetproxy is constructed
    :type proxy_source_model: Optional[Type["ModelRow"]]
    :param excludable: structure of fields to include and exclude
    :type excludable: ExcludableItems
    :param current_relation_str: name of the relation field
    :type current_relation_str: str
    :param source_model: model on which relation was defined
    :type source_model: Type[Model]
    :param row: raw result row from the database
    :type row: ResultProxy
    :param select_related: list of names of related models fetched from database
    :type select_related: List
    :param related_models: list or dict of related models
    :type related_models: Union[List, Dict]
    :param related_field: field with relation declaration
    :type related_field: ForeignKeyField
    :return: returns model if model is populated from database
    :rtype: Optional[Model]
    """
    item: Dict[str, Any] = {}
    select_related = select_related or []
    related_models = related_models or []
    table_prefix = ""
    used_prefixes = used_prefixes if used_prefixes is not None else []
    excludable = excludable or ExcludableItems()

    if select_related:
        related_models = group_related_list(select_related)

    if related_field:
        table_prefix = cls._process_table_prefix(
            source_model=source_model,
            current_relation_str=current_relation_str,
            related_field=related_field,
            used_prefixes=used_prefixes,
        )

    item = cls._populate_nested_models_from_row(
        item=item,
        row=row,
        related_models=related_models,
        excludable=excludable,
        current_relation_str=current_relation_str,
        source_model=source_model,  # type: ignore
        proxy_source_model=proxy_source_model,  # type: ignore
        table_prefix=table_prefix,
        used_prefixes=used_prefixes,
    )
    item = cls.extract_prefixed_table_columns(
        item=item, row=row, table_prefix=table_prefix, excludable=excludable
    )

    instance: Optional["Model"] = None
    if item.get(cls.ormar_config.pkname, None) is not None:
        item["__excluded__"] = cls.get_names_to_exclude(
            excludable=excludable, alias=table_prefix
        )
        instance = cast("Model", cls(**item))
        instance.set_save_status(True)
    return instance