Prodream/ormar
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

wip - through models fields -> attached in queries, accesible from instances, creates in add and queryset create

Browse Source
This commit is contained in:
collerek
2021-02-15 17:30:14 +01:00
parent 868243686d
commit 3fd231cf3c
19 changed files with 677 additions and 374 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
ormar/models/__init__.py
View File

@ -5,6 +5,7 @@ ass well as vast number of helper functions for pydantic, sqlalchemy and relatio
"""
from ormar.models.newbasemodel import NewBaseModel # noqa I100
from ormar.models.model_row import ModelRow # noqa I100
from ormar.models.model import Model # noqa I100
__all__ = ["NewBaseModel", "Model"]
__all__ = ["NewBaseModel", "Model", "ModelRow"]

36
ormar/models/helpers/relations.py
View File

@ -2,7 +2,7 @@ from typing import TYPE_CHECKING, Type
import ormar
from ormar import ForeignKey, ManyToMany
from ormar.fields import ManyToManyField
from ormar.fields import ManyToManyField, Through, ThroughField
from ormar.fields.foreign_key import ForeignKeyField
from ormar.models.helpers.sqlalchemy import adjust_through_many_to_many_model
from ormar.relations import AliasManager
@ -81,7 +81,8 @@ def expand_reverse_relationships(model: Type["Model"]) -> None:
:param model: model on which relation should be checked and registered
:type model: Model class
"""
for model_field in model.Meta.model_fields.values():
model_fields = list(model.Meta.model_fields.values())
for model_field in model_fields:
if (
issubclass(model_field, ForeignKeyField)
and not model_field.has_unresolved_forward_refs()
@ -113,6 +114,7 @@ def register_reverse_model_fields(model_field: Type["ForeignKeyField"]) -> None:
self_reference_primary=model_field.self_reference_primary,
)
# register foreign keys on through model
register_through_shortcut_fields(model_field=model_field)
adjust_through_many_to_many_model(model_field=model_field)
else:
model_field.to.Meta.model_fields[related_name] = ForeignKey(
@ -125,6 +127,34 @@ def register_reverse_model_fields(model_field: Type["ForeignKeyField"]) -> None:
)
def register_through_shortcut_fields(model_field: Type["ManyToManyField"]) -> None:
"""
Registers m2m relation through shortcut on both ends of the relation.
:param model_field: relation field defined in parent model
:type model_field: ManyToManyField
"""
through_model = model_field.through
through_name = through_model.get_name(lower=True)
related_name = model_field.get_related_name()
model_field.owner.Meta.model_fields[through_name] = Through(
through_model,
real_name=through_name,
virtual=True,
related_name=model_field.name,
owner=model_field.owner,
)
model_field.to.Meta.model_fields[through_name] = Through(
through_model,
real_name=through_name,
virtual=True,
related_name=related_name,
owner=model_field.to,
)
def register_relation_in_alias_manager(field: Type[ForeignKeyField]) -> None:
"""
Registers the relation (and reverse relation) in alias manager.
@ -142,7 +172,7 @@ def register_relation_in_alias_manager(field: Type[ForeignKeyField]) -> None:
if field.has_unresolved_forward_refs():
return
register_many_to_many_relation_on_build(field=field)
elif issubclass(field, ForeignKeyField):
elif issubclass(field, ForeignKeyField) and not issubclass(field, ThroughField):
if field.has_unresolved_forward_refs():
return
register_relation_on_build(field=field)

154
ormar/models/metaclass.py
View File

@ -46,6 +46,7 @@ if TYPE_CHECKING: # pragma no cover
from ormar import Model
CONFIG_KEY = "Config"
PARSED_FIELDS_KEY = "__parsed_fields__"
class ModelMeta:
@ -141,83 +142,6 @@ def register_signals(new_model: Type["Model"]) -> None: # noqa: CCR001
new_model.Meta.signals = signals
class ModelMetaclass(pydantic.main.ModelMetaclass):
def __new__( # type: ignore # noqa: CCR001
mcs: "ModelMetaclass", name: str, bases: Any, attrs: dict
) -> "ModelMetaclass":
"""
Metaclass used by ormar Models that performs configuration
and build of ormar Models.
Sets pydantic configuration.
Extract model_fields and convert them to pydantic FieldInfo,
updates class namespace.
Extracts settings and fields from parent classes.
Fetches methods decorated with @property_field decorator
to expose them later in dict().
Construct parent pydantic Metaclass/ Model.
If class has Meta class declared (so actual ormar Models) it also:
* populate sqlalchemy columns, pkname and tables from model_fields
* register reverse relationships on related models
* registers all relations in alias manager that populates table_prefixes
* exposes alias manager on each Model
* creates QuerySet for each model and exposes it on a class
:param name: name of current class
:type name: str
:param bases: base classes
:type bases: Tuple
:param attrs: class namespace
:type attrs: Dict
"""
attrs["Config"] = get_pydantic_base_orm_config()
attrs["__name__"] = name
attrs, model_fields = extract_annotations_and_default_vals(attrs)
for base in reversed(bases):
mod = base.__module__
if mod.startswith("ormar.models.") or mod.startswith("pydantic."):
continue
attrs, model_fields = extract_from_parents_definition(
base_class=base, curr_class=mcs, attrs=attrs, model_fields=model_fields
)
new_model = super().__new__( # type: ignore
mcs, name, bases, attrs
)
add_cached_properties(new_model)
if hasattr(new_model, "Meta"):
populate_default_options_values(new_model, model_fields)
check_required_meta_parameters(new_model)
add_property_fields(new_model, attrs)
register_signals(new_model=new_model)
populate_choices_validators(new_model)
if not new_model.Meta.abstract:
new_model = populate_meta_tablename_columns_and_pk(name, new_model)
populate_meta_sqlalchemy_table_if_required(new_model.Meta)
expand_reverse_relationships(new_model)
for field in new_model.Meta.model_fields.values():
register_relation_in_alias_manager(field=field)
if new_model.Meta.pkname not in attrs["__annotations__"]:
field_name = new_model.Meta.pkname
attrs["__annotations__"][field_name] = Optional[int] # type: ignore
attrs[field_name] = None
new_model.__fields__[field_name] = get_pydantic_field(
field_name=field_name, model=new_model
)
new_model.Meta.alias_manager = alias_manager
new_model.objects = QuerySet(new_model)
return new_model
def verify_constraint_names(
base_class: "Model", model_fields: Dict, parent_value: List
) -> None:
@ -539,4 +463,78 @@ def update_attrs_and_fields(
return updated_model_fields
PARSED_FIELDS_KEY = "__parsed_fields__"
class ModelMetaclass(pydantic.main.ModelMetaclass):
def __new__( # type: ignore # noqa: CCR001
mcs: "ModelMetaclass", name: str, bases: Any, attrs: dict
) -> "ModelMetaclass":
"""
Metaclass used by ormar Models that performs configuration
and build of ormar Models.
Sets pydantic configuration.
Extract model_fields and convert them to pydantic FieldInfo,
updates class namespace.
Extracts settings and fields from parent classes.
Fetches methods decorated with @property_field decorator
to expose them later in dict().
Construct parent pydantic Metaclass/ Model.
If class has Meta class declared (so actual ormar Models) it also:
* populate sqlalchemy columns, pkname and tables from model_fields
* register reverse relationships on related models
* registers all relations in alias manager that populates table_prefixes
* exposes alias manager on each Model
* creates QuerySet for each model and exposes it on a class
:param name: name of current class
:type name: str
:param bases: base classes
:type bases: Tuple
:param attrs: class namespace
:type attrs: Dict
"""
attrs["Config"] = get_pydantic_base_orm_config()
attrs["__name__"] = name
attrs, model_fields = extract_annotations_and_default_vals(attrs)
for base in reversed(bases):
mod = base.__module__
if mod.startswith("ormar.models.") or mod.startswith("pydantic."):
continue
attrs, model_fields = extract_from_parents_definition(
base_class=base, curr_class=mcs, attrs=attrs, model_fields=model_fields
)
new_model = super().__new__( # type: ignore
mcs, name, bases, attrs
)
add_cached_properties(new_model)
if hasattr(new_model, "Meta"):
populate_default_options_values(new_model, model_fields)
check_required_meta_parameters(new_model)
add_property_fields(new_model, attrs)
register_signals(new_model=new_model)
populate_choices_validators(new_model)
if not new_model.Meta.abstract:
new_model = populate_meta_tablename_columns_and_pk(name, new_model)
populate_meta_sqlalchemy_table_if_required(new_model.Meta)
expand_reverse_relationships(new_model)
for field in new_model.Meta.model_fields.values():
register_relation_in_alias_manager(field=field)
if new_model.Meta.pkname not in attrs["__annotations__"]:
field_name = new_model.Meta.pkname
attrs["__annotations__"][field_name] = Optional[int] # type: ignore
attrs[field_name] = None
new_model.__fields__[field_name] = get_pydantic_field(
field_name=field_name, model=new_model
)
new_model.Meta.alias_manager = alias_manager
new_model.objects = QuerySet(new_model)
return new_model

3
ormar/models/mixins/excludable_mixin.py
View File

@ -31,6 +31,7 @@ class ExcludableMixin(RelationMixin):
if TYPE_CHECKING: # pragma: no cover
from ormar import Model
from ormar.models import ModelRow
@staticmethod
def get_child(
@ -157,7 +158,7 @@ class ExcludableMixin(RelationMixin):
@classmethod
def own_table_columns(
cls,
model: Type["Model"],
model: Union[Type["Model"], Type["ModelRow"]],
fields: Optional[Union[Set, Dict]],
exclude_fields: Optional[Union[Set, Dict]],
use_alias: bool = False,

29
ormar/models/mixins/relation_mixin.py
View File

@ -1,6 +1,8 @@
import inspect
from typing import List, Optional, Set, TYPE_CHECKING
from ormar import ManyToManyField
from ormar.fields import ThroughField
from ormar.fields.foreign_key import ForeignKeyField
@ -43,27 +45,46 @@ class RelationMixin:
return cls._related_fields
related_fields = []
for name in cls.extract_related_names():
for name in cls.extract_related_names().union(cls.extract_through_names()):
related_fields.append(cls.Meta.model_fields[name])
cls._related_fields = related_fields
return related_fields
@classmethod
def extract_through_names(cls) -> Set:
"""
Extracts related fields through names which are shortcuts to through models.
:return: set of related through fields names
:rtype: Set
"""
related_fields = set()
for name in cls.extract_related_names():
field = cls.Meta.model_fields[name]
if issubclass(field, ManyToManyField):
related_fields.add(field.through.get_name(lower=True))
return related_fields
@classmethod
def extract_related_names(cls) -> Set:
"""
Returns List of fields names for all relations declared on a model.
List is cached in cls._related_names for quicker access.
:return: list of related fields names
:rtype: List
:return: set of related fields names
:rtype: Set
"""
if isinstance(cls._related_names, Set):
return cls._related_names
related_names = set()
for name, field in cls.Meta.model_fields.items():
if inspect.isclass(field) and issubclass(field, ForeignKeyField):
if (
inspect.isclass(field)
and issubclass(field, ForeignKeyField)
and not issubclass(field, ThroughField)
):
related_names.add(name)
cls._related_names = related_names

252
ormar/models/model.py
View File

@ -1,24 +1,17 @@
from typing import (
Any,
Dict,
List,
Optional,
Set,
TYPE_CHECKING,
Tuple,
Type,
TypeVar,
Union,
)
import sqlalchemy
import ormar.queryset # noqa I100
from ormar.exceptions import ModelPersistenceError, NoMatch
from ormar.fields.many_to_many import ManyToManyField
from ormar.models import NewBaseModel # noqa I100
from ormar.models.helpers.models import group_related_list
from ormar.models.metaclass import ModelMeta
from ormar.models.model_row import ModelRow
if TYPE_CHECKING: # pragma nocover
from ormar import QuerySet
@ -26,7 +19,7 @@ if TYPE_CHECKING: # pragma nocover
T = TypeVar("T", bound="Model")
class Model(NewBaseModel):
class Model(ModelRow):
__abstract__ = False
if TYPE_CHECKING: # pragma nocover
Meta: ModelMeta
@ -36,247 +29,6 @@ class Model(NewBaseModel):
_repr = {k: getattr(self, k) for k, v in self.Meta.model_fields.items()}
return f"{self.__class__.__name__}({str(_repr)})"
@classmethod
def from_row( # noqa CCR001
cls: Type[T],
row: sqlalchemy.engine.ResultProxy,
select_related: List = None,
related_models: Any = None,
previous_model: Type[T] = None,
source_model: Type[T] = None,
related_name: str = None,
fields: Optional[Union[Dict, Set]] = None,
exclude_fields: Optional[Union[Dict, Set]] = None,
current_relation_str: str = None,
) -> Optional[T]:
"""
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 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: sqlalchemy.engine.result.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 previous_model: internal param for nested models to specify table_prefix
:type previous_model: Model class
:param related_name: internal parameter - name of current nested model
:type related_name: str
:param fields: fields and related model fields to include
if provided only those are included
:type fields: Optional[Union[Dict, Set]]
:param exclude_fields: fields and related model fields to exclude
excludes the fields even if they are provided in fields
:type exclude_fields: Optional[Union[Dict, Set]]
: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 = ""
if select_related:
source_model = cls
related_models = group_related_list(select_related)
rel_name2 = related_name
if (
previous_model
and related_name
and issubclass(
previous_model.Meta.model_fields[related_name], ManyToManyField
)
):
through_field = previous_model.Meta.model_fields[related_name]
if (
through_field.self_reference
and related_name == through_field.self_reference_primary
):
rel_name2 = through_field.default_source_field_name() # type: ignore
else:
rel_name2 = through_field.default_target_field_name() # type: ignore
previous_model = through_field.through # type: ignore
if previous_model and rel_name2:
if current_relation_str and "__" in current_relation_str and source_model:
table_prefix = cls.Meta.alias_manager.resolve_relation_alias(
from_model=source_model, relation_name=current_relation_str
)
if not table_prefix:
table_prefix = cls.Meta.alias_manager.resolve_relation_alias(
from_model=previous_model, relation_name=rel_name2
)
item = cls.populate_nested_models_from_row(
item=item,
row=row,
related_models=related_models,
fields=fields,
exclude_fields=exclude_fields,
current_relation_str=current_relation_str,
source_model=source_model,
)
item = cls.extract_prefixed_table_columns(
item=item,
row=row,
table_prefix=table_prefix,
fields=fields,
exclude_fields=exclude_fields,
)
instance: Optional[T] = None
if item.get(cls.Meta.pkname, None) is not None:
item["__excluded__"] = cls.get_names_to_exclude(
fields=fields, exclude_fields=exclude_fields
)
instance = cls(**item)
instance.set_save_status(True)
return instance
@classmethod
def populate_nested_models_from_row( # noqa: CFQ002
cls,
item: dict,
row: sqlalchemy.engine.ResultProxy,
related_models: Any,
fields: Optional[Union[Dict, Set]] = None,
exclude_fields: Optional[Union[Dict, Set]] = None,
current_relation_str: str = None,
source_model: Type[T] = 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 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: sqlalchemy.engine.result.ResultProxy
:param related_models: list or dict of related models
:type related_models: Union[Dict, List]
:param fields: fields and related model fields to include -
if provided only those are included
:type fields: Optional[Union[Dict, Set]]
:param exclude_fields: fields and related model fields to exclude
excludes the fields even if they are provided in fields
:type exclude_fields: Optional[Union[Dict, Set]]
:return: dictionary with keys corresponding to model fields names
and values are database values
:rtype: Dict
"""
for related in related_models:
relation_str = (
"__".join([current_relation_str, related])
if current_relation_str
else related
)
fields = cls.get_included(fields, related)
exclude_fields = cls.get_excluded(exclude_fields, related)
model_cls = cls.Meta.model_fields[related].to
remainder = None
if isinstance(related_models, dict) and related_models[related]:
remainder = related_models[related]
child = model_cls.from_row(
row,
related_models=remainder,
previous_model=cls,
related_name=related,
fields=fields,
exclude_fields=exclude_fields,
current_relation_str=relation_str,
source_model=source_model,
)
item[model_cls.get_column_name_from_alias(related)] = child
return item
@classmethod
def extract_prefixed_table_columns( # noqa CCR001
cls,
item: dict,
row: sqlalchemy.engine.result.ResultProxy,
table_prefix: str,
fields: Optional[Union[Dict, Set]] = None,
exclude_fields: Optional[Union[Dict, Set]] = None,
) -> 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 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
:param fields: fields and related model fields to include -
if provided only those are included
:type fields: Optional[Union[Dict, Set]]
:param exclude_fields: fields and related model fields to exclude
excludes the fields even if they are provided in fields
:type exclude_fields: Optional[Union[Dict, Set]]
:return: dictionary with keys corresponding to model fields names
and values are database values
:rtype: Dict
"""
# databases does not keep aliases in Record for postgres, change to raw row
source = row._row if cls.db_backend_name() == "postgresql" else row
selected_columns = cls.own_table_columns(
model=cls,
fields=fields or {},
exclude_fields=exclude_fields or {},
use_alias=False,
)
for column in cls.Meta.table.columns:
alias = cls.get_column_name_from_alias(column.name)
if alias not in item and alias in selected_columns:
prefixed_name = (
f'{table_prefix + "_" if table_prefix else ""}{column.name}'
)
item[alias] = source[prefixed_name]
return item
async def upsert(self: T, **kwargs: Any) -> T:
"""
Performs either a save or an update depending on the presence of the pk.

303
ormar/models/model_row.py Normal file
View File

@ -0,0 +1,303 @@
from typing import (
Any,
Dict,
List,
Optional,
Set,
Type,
TypeVar,
Union,
)
import sqlalchemy
from ormar import ManyToManyField # noqa: I202
from ormar.models import NewBaseModel
from ormar.models.helpers.models import group_related_list
T = TypeVar("T", bound="ModelRow")
class ModelRow(NewBaseModel):
@classmethod
def from_row( # noqa CCR001
cls: Type[T],
row: sqlalchemy.engine.ResultProxy,
select_related: List = None,
related_models: Any = None,
previous_model: Type[T] = None,
source_model: Type[T] = None,
related_name: str = None,
fields: Optional[Union[Dict, Set]] = None,
exclude_fields: Optional[Union[Dict, Set]] = None,
current_relation_str: str = None,
) -> Optional[T]:
"""
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 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: sqlalchemy.engine.result.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 previous_model: internal param for nested models to specify table_prefix
:type previous_model: Model class
:param related_name: internal parameter - name of current nested model
:type related_name: str
:param fields: fields and related model fields to include
if provided only those are included
:type fields: Optional[Union[Dict, Set]]
:param exclude_fields: fields and related model fields to exclude
excludes the fields even if they are provided in fields
:type exclude_fields: Optional[Union[Dict, Set]]
: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 = ""
if select_related:
source_model = cls
related_models = group_related_list(select_related)
rel_name2 = related_name
# TODO: refactor this into field classes?
if (
previous_model
and related_name
and issubclass(
previous_model.Meta.model_fields[related_name], ManyToManyField
)
):
through_field = previous_model.Meta.model_fields[related_name]
if (
through_field.self_reference
and related_name == through_field.self_reference_primary
):
rel_name2 = through_field.default_source_field_name() # type: ignore
else:
rel_name2 = through_field.default_target_field_name() # type: ignore
previous_model = through_field.through # type: ignore
if previous_model and rel_name2:
if current_relation_str and "__" in current_relation_str and source_model:
table_prefix = cls.Meta.alias_manager.resolve_relation_alias(
from_model=source_model, relation_name=current_relation_str
)
if not table_prefix:
table_prefix = cls.Meta.alias_manager.resolve_relation_alias(
from_model=previous_model, relation_name=rel_name2
)
item = cls.populate_nested_models_from_row(
item=item,
row=row,
related_models=related_models,
fields=fields,
exclude_fields=exclude_fields,
current_relation_str=current_relation_str,
source_model=source_model,
)
item = cls.extract_prefixed_table_columns(
item=item,
row=row,
table_prefix=table_prefix,
fields=fields,
exclude_fields=exclude_fields,
)
instance: Optional[T] = None
if item.get(cls.Meta.pkname, None) is not None:
item["__excluded__"] = cls.get_names_to_exclude(
fields=fields, exclude_fields=exclude_fields
)
instance = cls(**item)
instance.set_save_status(True)
return instance
@classmethod
def populate_nested_models_from_row( # noqa: CFQ002
cls,
item: dict,
row: sqlalchemy.engine.ResultProxy,
related_models: Any,
fields: Optional[Union[Dict, Set]] = None,
exclude_fields: Optional[Union[Dict, Set]] = None,
current_relation_str: str = None,
source_model: Type[T] = 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 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: sqlalchemy.engine.result.ResultProxy
:param related_models: list or dict of related models
:type related_models: Union[Dict, List]
:param fields: fields and related model fields to include -
if provided only those are included
:type fields: Optional[Union[Dict, Set]]
:param exclude_fields: fields and related model fields to exclude
excludes the fields even if they are provided in fields
:type exclude_fields: Optional[Union[Dict, Set]]
:return: dictionary with keys corresponding to model fields names
and values are database values
:rtype: Dict
"""
for related in related_models:
relation_str = (
"__".join([current_relation_str, related])
if current_relation_str
else related
)
field = cls.Meta.model_fields[related]
fields = cls.get_included(fields, related)
exclude_fields = cls.get_excluded(exclude_fields, related)
model_cls = field.to
remainder = None
if isinstance(related_models, dict) and related_models[related]:
remainder = related_models[related]
child = model_cls.from_row(
row,
related_models=remainder,
previous_model=cls,
related_name=related,
fields=fields,
exclude_fields=exclude_fields,
current_relation_str=relation_str,
source_model=source_model,
)
item[model_cls.get_column_name_from_alias(related)] = child
if issubclass(field, ManyToManyField) and child:
# TODO: way to figure out which side should be populated?
through_name = cls.Meta.model_fields[related].through.get_name()
# for now it's nested dict, should be instance?
through_child = cls.populate_through_instance(
row=row,
related=related,
through_name=through_name,
fields=fields,
exclude_fields=exclude_fields,
)
item[through_name] = through_child
setattr(child, through_name, through_child)
child.set_save_status(True)
return item
@classmethod
def populate_through_instance(
cls,
row: sqlalchemy.engine.ResultProxy,
through_name: str,
related: str,
fields: Optional[Union[Dict, Set]] = None,
exclude_fields: Optional[Union[Dict, Set]] = None,
) -> Dict:
# TODO: fix excludes and includes
fields = cls.get_included(fields, through_name)
# exclude_fields = cls.get_excluded(exclude_fields, through_name)
model_cls = cls.Meta.model_fields[through_name].to
exclude_fields = model_cls.extract_related_names()
table_prefix = cls.Meta.alias_manager.resolve_relation_alias(
from_model=cls, relation_name=related
)
child = model_cls.extract_prefixed_table_columns(
item={},
row=row,
table_prefix=table_prefix,
fields=fields,
exclude_fields=exclude_fields,
)
return child
@classmethod
def extract_prefixed_table_columns( # noqa CCR001
cls,
item: dict,
row: sqlalchemy.engine.result.ResultProxy,
table_prefix: str,
fields: Optional[Union[Dict, Set]] = None,
exclude_fields: Optional[Union[Dict, Set]] = None,
) -> 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 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
:param fields: fields and related model fields to include -
if provided only those are included
:type fields: Optional[Union[Dict, Set]]
:param exclude_fields: fields and related model fields to exclude
excludes the fields even if they are provided in fields
:type exclude_fields: Optional[Union[Dict, Set]]
:return: dictionary with keys corresponding to model fields names
and values are database values
:rtype: Dict
"""
# databases does not keep aliases in Record for postgres, change to raw row
source = row._row if cls.db_backend_name() == "postgresql" else row
selected_columns = cls.own_table_columns(
model=cls,
fields=fields or {},
exclude_fields=exclude_fields or {},
use_alias=False,
)
for column in cls.Meta.table.columns:
alias = cls.get_column_name_from_alias(column.name)
if alias not in item and alias in selected_columns:
prefixed_name = (
f'{table_prefix + "_" if table_prefix else ""}{column.name}'
)
item[alias] = source[prefixed_name]
return item

6
ormar/models/newbasemodel.py
View File

@ -172,7 +172,7 @@ class NewBaseModel(pydantic.BaseModel, ModelTableProxy, metaclass=ModelMetaclass
object.__setattr__(self, "__fields_set__", fields_set)
# register the columns models after initialization
for related in self.extract_related_names():
for related in self.extract_related_names().union(self.extract_through_names()):
self.Meta.model_fields[related].expand_relationship(
new_kwargs.get(related), self, to_register=True,
)
@ -267,6 +267,10 @@ class NewBaseModel(pydantic.BaseModel, ModelTableProxy, metaclass=ModelMetaclass
return object.__getattribute__(
self, "_extract_related_model_instead_of_field"
)(item)
if item in object.__getattribute__(self, "extract_through_names")():
return object.__getattribute__(
self, "_extract_related_model_instead_of_field"
)(item)
if item in object.__getattribute__(self, "Meta").property_fields:
value = object.__getattribute__(self, item)
return value() if callable(value) else value

2
ormar/models/quick_access_views.py
View File

@ -34,10 +34,12 @@ quick_access_set = {
"_skip_ellipsis",
"_update_and_follow",
"_update_excluded_with_related_not_required",
"_verify_model_can_be_initialized",
"copy",
"delete",
"dict",
"extract_related_names",
"extract_through_names",
"update_from_dict",
"get_column_alias",
"get_column_name_from_alias",