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

remove auto related parsing, switch to relations on instance instead of relationship manager

Browse Source
This commit is contained in:
collerek
2020-08-24 11:15:59 +02:00
parent 9bbf6f93ed
commit 63a24e7d36
12 changed files with 295 additions and 209 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
ormar/models/model.py
View File

@ -1,4 +1,5 @@
from typing import Any, List
import itertools
from typing import Any, List, Tuple, Union
import sqlalchemy
@ -6,6 +7,21 @@ import ormar.queryset # noqa I100
from ormar.models import NewBaseModel # noqa I100
def group_related_list(list_):
test_dict = dict()
grouped = itertools.groupby(list_, key=lambda x: x.split("__")[0])
for key, group in grouped:
group_list = list(group)
new = [
"__".join(x.split("__")[1:]) for x in group_list if len(x.split("__")) > 1
]
if any("__" in x for x in new):
test_dict[key] = group_related_list(new)
else:
test_dict[key] = new
return test_dict
class Model(NewBaseModel):
__abstract__ = False
@ -14,22 +30,27 @@ class Model(NewBaseModel):
cls,
row: sqlalchemy.engine.ResultProxy,
select_related: List = None,
related_models: Any = None,
previous_table: str = None,
) -> "Model":
) -> Union["Model", Tuple["Model", dict]]:
item = {}
select_related = select_related or []
related_models = related_models or []
if select_related:
related_models = group_related_list(select_related)
table_prefix = cls.Meta._orm_relationship_manager.resolve_relation_join(
previous_table, cls.Meta.table.name
)
previous_table = cls.Meta.table.name
for related in select_related:
if "__" in related:
first_part, remainder = related.split("__", 1)
for related in related_models:
if isinstance(related_models, dict) and related_models[related]:
first_part, remainder = related, related_models[related]
model_cls = cls.Meta.model_fields[first_part].to
child = model_cls.from_row(
row, select_related=[remainder], previous_table=previous_table
row, related_models=remainder, previous_table=previous_table
)
item[first_part] = child
else:
@ -43,7 +64,8 @@ class Model(NewBaseModel):
f'{table_prefix + "_" if table_prefix else ""}{column.name}'
]
return cls(**item)
instance = cls(**item) if item.get(cls.Meta.pkname, None) is not None else None
return instance
async def save(self) -> "Model":
self_fields = self._extract_model_db_fields()

18
ormar/models/modelproxy.py
View File

@ -43,6 +43,18 @@ class ModelTableProxy:
related_names.add(name)
return related_names
@classmethod
def _extract_db_related_names(cls) -> Set:
related_names = set()
for name, field in cls.Meta.model_fields.items():
if (
inspect.isclass(field)
and issubclass(field, ForeignKeyField)
and not field.virtual
):
related_names.add(name)
return related_names
@classmethod
def _exclude_related_names_not_required(cls, nested: bool = False) -> Set:
if nested:
@ -62,7 +74,7 @@ class ModelTableProxy:
self_fields = {
k: v for k, v in self_fields.items() if k in self.Meta.table.columns
}
for field in self._extract_related_names():
for field in self._extract_db_related_names():
target_pk_name = self.Meta.model_fields[field].to.Meta.pkname
if getattr(self, field) is not None:
self_fields[field] = getattr(getattr(self, field), target_pk_name)
@ -72,8 +84,8 @@ class ModelTableProxy:
def merge_instances_list(cls, result_rows: List["Model"]) -> List["Model"]:
merged_rows = []
for index, model in enumerate(result_rows):
if index > 0 and model.pk == result_rows[index - 1].pk:
result_rows[-1] = cls.merge_two_instances(model, merged_rows[-1])
if index > 0 and model.pk == merged_rows[-1].pk:
merged_rows[-1] = cls.merge_two_instances(model, merged_rows[-1])
else:
merged_rows.append(model)
return merged_rows

52
ormar/models/newbasemodel.py
View File

@ -20,9 +20,10 @@ from pydantic import BaseModel
import ormar # noqa I100
from ormar.fields import BaseField
from ormar.fields.foreign_key import ForeignKeyField
from ormar.models.metaclass import ModelMeta, ModelMetaclass
from ormar.models.modelproxy import ModelTableProxy
from ormar.relations import AliasManager
from ormar.relations import AliasManager, RelationsManager
if TYPE_CHECKING: # pragma no cover
from ormar.models.model import Model
@ -34,7 +35,7 @@ if TYPE_CHECKING: # pragma no cover
class NewBaseModel(pydantic.BaseModel, ModelTableProxy, metaclass=ModelMetaclass):
__slots__ = ("_orm_id", "_orm_saved")
__slots__ = ("_orm_id", "_orm_saved", "_orm")
if TYPE_CHECKING: # pragma no cover
__model_fields__: Dict[str, TypeVar[BaseField]]
@ -46,6 +47,7 @@ class NewBaseModel(pydantic.BaseModel, ModelTableProxy, metaclass=ModelMetaclass
__metadata__: sqlalchemy.MetaData
__database__: databases.Database
_orm_relationship_manager: AliasManager
_orm: RelationsManager
Meta: ModelMeta
# noinspection PyMissingConstructor
@ -53,6 +55,18 @@ class NewBaseModel(pydantic.BaseModel, ModelTableProxy, metaclass=ModelMetaclass
object.__setattr__(self, "_orm_id", uuid.uuid4().hex)
object.__setattr__(self, "_orm_saved", False)
object.__setattr__(
self,
"_orm",
RelationsManager(
related_fields=[
field
for name, field in self.Meta.model_fields.items()
if issubclass(field, ForeignKeyField)
],
owner=self,
),
)
pk_only = kwargs.pop("__pk_only__", False)
if "pk" in kwargs:
@ -71,16 +85,12 @@ class NewBaseModel(pydantic.BaseModel, ModelTableProxy, metaclass=ModelMetaclass
object.__setattr__(self, "__dict__", values)
object.__setattr__(self, "__fields_set__", fields_set)
def __del__(self) -> None:
self.Meta._orm_relationship_manager.deregister(self)
def __setattr__(self, name: str, value: Any) -> None:
relation_key = self.get_name(title=True) + "_" + name
if name in self.__slots__:
object.__setattr__(self, name, value)
elif name == "pk":
object.__setattr__(self, self.Meta.pkname, value)
elif self.Meta._orm_relationship_manager.contains(relation_key, self):
elif name in self._orm:
self.Meta.model_fields[name].expand_relationship(value, self)
else:
value = (
@ -91,24 +101,27 @@ class NewBaseModel(pydantic.BaseModel, ModelTableProxy, metaclass=ModelMetaclass
super().__setattr__(name, value)
def __getattribute__(self, item: str) -> Any:
if item != "__fields__" and item in self.__fields__:
related = self._extract_related_model_instead_of_field(item)
if related:
return related
value = object.__getattribute__(self, item)
if item in ("_orm_id", "_orm_saved", "_orm", "__fields__"):
return object.__getattribute__(self, item)
elif item != "_extract_related_names" and item in self._extract_related_names():
return self._extract_related_model_instead_of_field(item)
elif item == "pk":
return self.__dict__.get(self.Meta.pkname, None)
elif item != "__fields__" and item in self.__fields__:
value = self.__dict__.get(item, None)
value = self._convert_json(item, value, "loads")
return value
return super().__getattribute__(item)
def __getattr__(self, item: str) -> Optional[Union["Model", List["Model"]]]:
return self._extract_related_model_instead_of_field(item)
# def __getattr__(self, item: str) -> Optional[Union["Model", List["Model"]]]:
# return self._extract_related_model_instead_of_field(item)
def _extract_related_model_instead_of_field(
self, item: str
) -> Optional[Union["Model", List["Model"]]]:
relation_key = self.get_name(title=True) + "_" + item
if self.Meta._orm_relationship_manager.contains(relation_key, self):
return self.Meta._orm_relationship_manager.get(relation_key, self)
# relation_key = self.get_name(title=True) + "_" + item
if item in self._orm:
return self._orm.get(item)
def __same__(self, other: "Model") -> bool:
if self.__class__ != other.__class__: # pragma no cover
@ -128,10 +141,6 @@ class NewBaseModel(pydantic.BaseModel, ModelTableProxy, metaclass=ModelMetaclass
name = name.title()
return name
@property
def pk(self) -> Any:
return getattr(self, self.Meta.pkname)
@property
def pk_column(self) -> sqlalchemy.Column:
return self.Meta.table.primary_key.columns.values()[0]
@ -177,7 +186,6 @@ class NewBaseModel(pydantic.BaseModel, ModelTableProxy, metaclass=ModelMetaclass
setattr(self, key, value)
def _convert_json(self, column_name: str, value: Any, op: str) -> Union[str, dict]:
if not self._is_conversion_to_json_needed(column_name):
return value