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

further refactor into mixins

Browse Source
This commit is contained in:
collerek
2020-12-31 09:23:21 +01:00
parent e695db712f
commit 101ea57879
12 changed files with 390 additions and 305 deletions
Download Patch File Download Diff File Expand all files Collapse all files

49
ormar/models/excludable.py
View File

@ -1,49 +0,0 @@
from typing import Dict, Set, Union
class Excludable:
@staticmethod
def get_child(
items: Union[Set, Dict, None], key: str = None
) -> Union[Set, Dict, None]:
if isinstance(items, dict):
return items.get(key, {})
return items
@staticmethod
def get_excluded(
exclude: Union[Set, Dict, None], key: str = None
) -> Union[Set, Dict, None]:
return Excludable.get_child(items=exclude, key=key)
@staticmethod
def get_included(
include: Union[Set, Dict, None], key: str = None
) -> Union[Set, Dict, None]:
return Excludable.get_child(items=include, key=key)
@staticmethod
def is_excluded(exclude: Union[Set, Dict, None], key: str = None) -> bool:
if exclude is None:
return False
if exclude is Ellipsis: # pragma: nocover
return True
to_exclude = Excludable.get_excluded(exclude=exclude, key=key)
if isinstance(to_exclude, Set):
return key in to_exclude
if to_exclude is ...:
return True
return False
@staticmethod
def is_included(include: Union[Set, Dict, None], key: str = None) -> bool:
if include is None:
return True
if include is Ellipsis:
return True
to_include = Excludable.get_included(include=include, key=key)
if isinstance(to_include, Set):
return key in to_include
if to_include is ...:
return True
return False

31
ormar/models/helpers/__init__.py
View File

@ -0,0 +1,31 @@
from ormar.models.helpers.models import (
extract_annotations_and_default_vals,
populate_default_options_values,
)
from ormar.models.helpers.pydantic import (
get_potential_fields,
get_pydantic_base_orm_config,
get_pydantic_field,
)
from ormar.models.helpers.relations import (
alias_manager,
register_relation_in_alias_manager,
)
from ormar.models.helpers.relations import expand_reverse_relationships
from ormar.models.helpers.sqlalchemy import (
populate_meta_sqlalchemy_table_if_required,
populate_meta_tablename_columns_and_pk,
)
__all__ = [
"expand_reverse_relationships",
"extract_annotations_and_default_vals",
"populate_meta_tablename_columns_and_pk",
"populate_meta_sqlalchemy_table_if_required",
"populate_default_options_values",
"alias_manager",
"register_relation_in_alias_manager",
"get_pydantic_field",
"get_potential_fields",
"get_pydantic_base_orm_config",
]

17
ormar/models/metaclass.py
View File

@ -21,23 +21,17 @@ from ormar import ForeignKey, Integer, ModelDefinitionError # noqa I100
from ormar.fields import BaseField
from ormar.fields.foreign_key import ForeignKeyField
from ormar.fields.many_to_many import ManyToManyField
from ormar.models.helpers.models import (
from ormar.models.helpers import (
alias_manager,
expand_reverse_relationships,
extract_annotations_and_default_vals,
populate_default_options_values,
)
from ormar.models.helpers.pydantic import (
get_potential_fields,
get_pydantic_base_orm_config,
get_pydantic_field,
)
from ormar.models.helpers.relations import (
alias_manager,
register_relation_in_alias_manager,
)
from ormar.models.helpers.relations import expand_reverse_relationships
from ormar.models.helpers.sqlalchemy import (
populate_default_options_values,
populate_meta_sqlalchemy_table_if_required,
populate_meta_tablename_columns_and_pk,
register_relation_in_alias_manager,
)
from ormar.models.quick_access_views import quick_access_set
from ormar.queryset import QuerySet
@ -387,7 +381,6 @@ def copy_data_from_parent_model( # noqa: CCR001
}
populate_meta_sqlalchemy_table_if_required(new_meta)
copy_name = through_class.__name__ + attrs.get("__name__", "")
# TODO: when adding additional fields they need to be copied here
copy_through = type(copy_name, (ormar.Model,), {"Meta": new_meta})
copy_field.through = copy_through

16
ormar/models/mixins/__init__.py
View File

@ -1,5 +1,19 @@
"""
Package contains functionalities divided by features.
All mixins are combined into ModelTableProxy which is one of the parents of Model.
The split into mixins was done to ease the maintainability of the proxy class, as
it became quite complicated over time.
"""
from ormar.models.mixins.alias_mixin import AliasMixin
from ormar.models.mixins.excludable_mixin import ExcludableMixin
from ormar.models.mixins.merge_mixin import MergeModelMixin
from ormar.models.mixins.prefetch_mixin import PrefetchQueryMixin
from ormar.models.mixins.save_mixin import SavePrepareMixin
__all__ = ["MergeModelMixin", "AliasMixin", "PrefetchQueryMixin"]
__all__ = [
"MergeModelMixin",
"AliasMixin",
"PrefetchQueryMixin",
"SavePrepareMixin",
"ExcludableMixin",
]

87
ormar/models/mixins/alias_mixin.py
View File

@ -1,19 +1,35 @@
from typing import Dict, List, Optional, Set, TYPE_CHECKING, Type, Union
from typing import Dict, TYPE_CHECKING
class AliasMixin:
if TYPE_CHECKING: # pragma: no cover
from ormar import Model, ModelMeta
from ormar import ModelMeta
Meta: ModelMeta
@classmethod
def get_column_alias(cls, field_name: str) -> str:
"""
Returns db alias (column name in db) for given ormar field.
For fields without alias field name is returned.
:param field_name: name of the field to get alias from
:type field_name: str
:return: alias (db name) if set, otherwise passed name
:rtype: str
"""
field = cls.Meta.model_fields.get(field_name)
return field.get_alias() if field is not None else field_name
@classmethod
def get_column_name_from_alias(cls, alias: str) -> str:
"""
Returns ormar field name for given db alias (column name in db).
If field do not have alias it's returned as is.
:param alias:
:type alias: str
:return: field name if set, otherwise passed alias (db name)
:rtype: str
"""
for field_name, field in cls.Meta.model_fields.items():
if field.get_alias() == alias:
return field_name
@ -21,6 +37,15 @@ class AliasMixin:
@classmethod
def translate_columns_to_aliases(cls, new_kwargs: Dict) -> Dict:
"""
Translates dictionary of model fields changing field names into aliases.
If field has no alias the field name remains intact.
Only fields present in the dictionary are translated.
:param new_kwargs: dict with fields names and their values
:type new_kwargs: Dict
:return: dict with aliases and their values
:rtype: Dict
"""
for field_name, field in cls.Meta.model_fields.items():
if field_name in new_kwargs:
new_kwargs[field.get_alias()] = new_kwargs.pop(field_name)
@ -28,56 +53,16 @@ class AliasMixin:
@classmethod
def translate_aliases_to_columns(cls, new_kwargs: Dict) -> Dict:
"""
Translates dictionary of model fields changing aliases into field names.
If field has no alias the alias is already a field name.
Only fields present in the dictionary are translated.
:param new_kwargs: dict with aliases and their values
:type new_kwargs: Dict
:return: dict with fields names and their values
:rtype: Dict
"""
for field_name, field in cls.Meta.model_fields.items():
if field.alias and field.alias in new_kwargs:
new_kwargs[field_name] = new_kwargs.pop(field.alias)
return new_kwargs
@staticmethod
def _populate_pk_column(
model: Type["Model"], columns: List[str], use_alias: bool = False,
) -> List[str]:
pk_alias = (
model.get_column_alias(model.Meta.pkname)
if use_alias
else model.Meta.pkname
)
if pk_alias not in columns:
columns.append(pk_alias)
return columns
@classmethod
def own_table_columns(
cls,
model: Type["Model"],
fields: Optional[Union[Set, Dict]],
exclude_fields: Optional[Union[Set, Dict]],
use_alias: bool = False,
) -> List[str]:
columns = [
model.get_column_name_from_alias(col.name) if not use_alias else col.name
for col in model.Meta.table.columns
]
field_names = [
model.get_column_name_from_alias(col.name)
for col in model.Meta.table.columns
]
if fields:
columns = [
col
for col, name in zip(columns, field_names)
if model.is_included(fields, name)
]
if exclude_fields:
columns = [
col
for col, name in zip(columns, field_names)
if not model.is_excluded(exclude_fields, name)
]
# always has to return pk column for ormar to work
columns = cls._populate_pk_column(
model=model, columns=columns, use_alias=use_alias
)
return columns

169
ormar/models/mixins/excludable_mixin.py Normal file
View File

@ -0,0 +1,169 @@
from typing import (
AbstractSet,
Any,
Dict,
List,
Mapping,
Optional,
Set,
TYPE_CHECKING,
Type,
TypeVar,
Union,
)
from ormar.models.mixins.relation_mixin import RelationMixin
from ormar.queryset.utils import translate_list_to_dict, update
if TYPE_CHECKING: # pragma no cover
from ormar import Model
T = TypeVar("T", bound=Model)
IntStr = Union[int, str]
AbstractSetIntStr = AbstractSet[IntStr]
MappingIntStrAny = Mapping[IntStr, Any]
class ExcludableMixin(RelationMixin):
if TYPE_CHECKING: # pragma: no cover
from ormar import Model
@staticmethod
def get_child(
items: Union[Set, Dict, None], key: str = None
) -> Union[Set, Dict, None]:
if isinstance(items, dict):
return items.get(key, {})
return items
@staticmethod
def get_excluded(
exclude: Union[Set, Dict, None], key: str = None
) -> Union[Set, Dict, None]:
return ExcludableMixin.get_child(items=exclude, key=key)
@staticmethod
def get_included(
include: Union[Set, Dict, None], key: str = None
) -> Union[Set, Dict, None]:
return ExcludableMixin.get_child(items=include, key=key)
@staticmethod
def is_excluded(exclude: Union[Set, Dict, None], key: str = None) -> bool:
if exclude is None:
return False
if exclude is Ellipsis: # pragma: nocover
return True
to_exclude = ExcludableMixin.get_excluded(exclude=exclude, key=key)
if isinstance(to_exclude, Set):
return key in to_exclude
if to_exclude is ...:
return True
return False
@staticmethod
def is_included(include: Union[Set, Dict, None], key: str = None) -> bool:
if include is None:
return True
if include is Ellipsis:
return True
to_include = ExcludableMixin.get_included(include=include, key=key)
if isinstance(to_include, Set):
return key in to_include
if to_include is ...:
return True
return False
@staticmethod
def _populate_pk_column(
model: Type["Model"], columns: List[str], use_alias: bool = False,
) -> List[str]:
pk_alias = (
model.get_column_alias(model.Meta.pkname)
if use_alias
else model.Meta.pkname
)
if pk_alias not in columns:
columns.append(pk_alias)
return columns
@classmethod
def own_table_columns(
cls,
model: Type["Model"],
fields: Optional[Union[Set, Dict]],
exclude_fields: Optional[Union[Set, Dict]],
use_alias: bool = False,
) -> List[str]:
columns = [
model.get_column_name_from_alias(col.name) if not use_alias else col.name
for col in model.Meta.table.columns
]
field_names = [
model.get_column_name_from_alias(col.name)
for col in model.Meta.table.columns
]
if fields:
columns = [
col
for col, name in zip(columns, field_names)
if model.is_included(fields, name)
]
if exclude_fields:
columns = [
col
for col, name in zip(columns, field_names)
if not model.is_excluded(exclude_fields, name)
]
# always has to return pk column for ormar to work
columns = cls._populate_pk_column(
model=model, columns=columns, use_alias=use_alias
)
return columns
@classmethod
def _update_excluded_with_related_not_required(
cls,
exclude: Union["AbstractSetIntStr", "MappingIntStrAny", None],
nested: bool = False,
) -> Union[Set, Dict]:
exclude = exclude or {}
related_set = cls._exclude_related_names_not_required(nested=nested)
if isinstance(exclude, set):
exclude.union(related_set)
else:
related_dict = translate_list_to_dict(related_set)
exclude = update(related_dict, exclude)
return exclude
@classmethod
def get_names_to_exclude(
cls,
fields: Optional[Union[Dict, Set]] = None,
exclude_fields: Optional[Union[Dict, Set]] = None,
) -> Set:
fields_names = cls.extract_db_own_fields()
if fields and fields is not Ellipsis:
fields_to_keep = {name for name in fields if name in fields_names}
else:
fields_to_keep = fields_names
fields_to_exclude = fields_names - fields_to_keep
if isinstance(exclude_fields, Set):
fields_to_exclude = fields_to_exclude.union(
{name for name in exclude_fields if name in fields_names}
)
elif isinstance(exclude_fields, Dict):
new_to_exclude = {
name
for name in exclude_fields
if name in fields_names and exclude_fields[name] is Ellipsis
}
fields_to_exclude = fields_to_exclude.union(new_to_exclude)
fields_to_exclude = fields_to_exclude - {cls.Meta.pkname}
return fields_to_exclude

4
ormar/models/mixins/prefetch_mixin.py
View File

@ -2,14 +2,14 @@ from typing import Callable, Dict, List, TYPE_CHECKING, Tuple, Type
import ormar
from ormar.fields import BaseField
from ormar.models.mixins.relation_mixin import RelationMixin
class PrefetchQueryMixin:
class PrefetchQueryMixin(RelationMixin):
if TYPE_CHECKING: # pragma no cover
from ormar import Model
get_name: Callable # defined in NewBaseModel
extract_related_names: Callable # defined in ModelTableProxy
@staticmethod
def get_clause_target_and_filter_column_name(

68
ormar/models/mixins/relation_mixin.py Normal file
View File

@ -0,0 +1,68 @@
import inspect
from typing import List, Optional, Set, TYPE_CHECKING
from ormar.fields.foreign_key import ForeignKeyField
class RelationMixin:
if TYPE_CHECKING: # pragma no cover
from ormar import ModelMeta
Meta: ModelMeta
_related_names: Optional[Set]
_related_fields: Optional[List]
@classmethod
def extract_db_own_fields(cls) -> Set:
related_names = cls.extract_related_names()
self_fields = {
name for name in cls.Meta.model_fields.keys() if name not in related_names
}
return self_fields
@classmethod
def extract_related_fields(cls) -> List:
if isinstance(cls._related_fields, List):
return cls._related_fields
related_fields = []
for name in cls.extract_related_names():
related_fields.append(cls.Meta.model_fields[name])
cls._related_fields = related_fields
return related_fields
@classmethod
def extract_related_names(cls) -> 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):
related_names.add(name)
cls._related_names = related_names
return related_names
@classmethod
def _extract_db_related_names(cls) -> Set:
related_names = cls.extract_related_names()
related_names = {
name
for name in related_names
if cls.Meta.model_fields[name].is_valid_uni_relation()
}
return related_names
@classmethod
def _exclude_related_names_not_required(cls, nested: bool = False) -> Set:
if nested:
return cls.extract_related_names()
related_names = cls.extract_related_names()
related_names = {
name for name in related_names if cls.Meta.model_fields[name].nullable
}
return related_names

47
ormar/models/mixins/save_mixin.py Normal file
View File

@ -0,0 +1,47 @@
from typing import Dict
import ormar
from ormar.exceptions import ModelPersistenceError
from ormar.models.mixins.relation_mixin import RelationMixin
class SavePrepareMixin(RelationMixin):
@classmethod
def substitute_models_with_pks(cls, model_dict: Dict) -> Dict: # noqa CCR001
for field in cls.extract_related_names():
field_value = model_dict.get(field, None)
if field_value is not None:
target_field = cls.Meta.model_fields[field]
target_pkname = target_field.to.Meta.pkname
if isinstance(field_value, ormar.Model):
pk_value = getattr(field_value, target_pkname)
if not pk_value:
raise ModelPersistenceError(
f"You cannot save {field_value.get_name()} "
f"model without pk set!"
)
model_dict[field] = pk_value
elif field_value: # nested dict
if isinstance(field_value, list):
model_dict[field] = [
target.get(target_pkname) for target in field_value
]
else:
model_dict[field] = field_value.get(target_pkname)
else:
model_dict.pop(field, None)
return model_dict
@classmethod
def populate_default_values(cls, new_kwargs: Dict) -> Dict:
for field_name, field in cls.Meta.model_fields.items():
if (
field_name not in new_kwargs
and field.has_default(use_server=False)
and not field.pydantic_only
):
new_kwargs[field_name] = field.get_default()
# clear fields with server_default set as None
if field.server_default is not None and not new_kwargs.get(field_name):
new_kwargs.pop(field_name, None)
return new_kwargs

189
ormar/models/modelproxy.py
View File

@ -1,183 +1,14 @@
import inspect
from typing import (
AbstractSet,
Any,
Callable,
Dict,
List,
Mapping,
Optional,
Set,
TYPE_CHECKING,
TypeVar,
Union,
)
import ormar # noqa: I100
from ormar.exceptions import ModelPersistenceError
from ormar.fields import BaseField
from ormar.fields.foreign_key import ForeignKeyField
from ormar.models.metaclass import ModelMeta
from ormar.models.mixins import AliasMixin, MergeModelMixin, PrefetchQueryMixin
from ormar.queryset.utils import translate_list_to_dict, update
if TYPE_CHECKING: # pragma no cover
from ormar import Model
T = TypeVar("T", bound=Model)
IntStr = Union[int, str]
AbstractSetIntStr = AbstractSet[IntStr]
MappingIntStrAny = Mapping[IntStr, Any]
Field = TypeVar("Field", bound=BaseField)
class ModelTableProxy(PrefetchQueryMixin, MergeModelMixin, AliasMixin):
if TYPE_CHECKING: # pragma no cover
Meta: ModelMeta
_related_names: Optional[Set]
_related_fields: Optional[List]
pk: Any
get_name: Callable
_props: Set
dict: Callable # noqa: A001, VNE003
@classmethod
def extract_db_own_fields(cls) -> Set:
related_names = cls.extract_related_names()
self_fields = {
name for name in cls.Meta.model_fields.keys() if name not in related_names
}
return self_fields
@classmethod
def substitute_models_with_pks(cls, model_dict: Dict) -> Dict: # noqa CCR001
for field in cls.extract_related_names():
field_value = model_dict.get(field, None)
if field_value is not None:
target_field = cls.Meta.model_fields[field]
target_pkname = target_field.to.Meta.pkname
if isinstance(field_value, ormar.Model):
pk_value = getattr(field_value, target_pkname)
if not pk_value:
raise ModelPersistenceError(
f"You cannot save {field_value.get_name()} "
f"model without pk set!"
from ormar.models.mixins import (
AliasMixin,
ExcludableMixin,
MergeModelMixin,
PrefetchQueryMixin,
SavePrepareMixin,
)
model_dict[field] = pk_value
elif field_value: # nested dict
if isinstance(field_value, list):
model_dict[field] = [
target.get(target_pkname) for target in field_value
]
else:
model_dict[field] = field_value.get(target_pkname)
else:
model_dict.pop(field, None)
return model_dict
@classmethod
def populate_default_values(cls, new_kwargs: Dict) -> Dict:
for field_name, field in cls.Meta.model_fields.items():
if (
field_name not in new_kwargs
and field.has_default(use_server=False)
and not field.pydantic_only
class ModelTableProxy(
PrefetchQueryMixin, MergeModelMixin, AliasMixin, SavePrepareMixin, ExcludableMixin
):
new_kwargs[field_name] = field.get_default()
# clear fields with server_default set as None
if field.server_default is not None and not new_kwargs.get(field_name):
new_kwargs.pop(field_name, None)
return new_kwargs
@classmethod
def extract_related_fields(cls) -> List:
if isinstance(cls._related_fields, List):
return cls._related_fields
related_fields = []
for name in cls.extract_related_names():
related_fields.append(cls.Meta.model_fields[name])
cls._related_fields = related_fields
return related_fields
@classmethod
def extract_related_names(cls) -> 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):
related_names.add(name)
cls._related_names = related_names
return related_names
@classmethod
def _extract_db_related_names(cls) -> Set:
related_names = cls.extract_related_names()
related_names = {
name
for name in related_names
if cls.Meta.model_fields[name].is_valid_uni_relation()
}
return related_names
@classmethod
def _exclude_related_names_not_required(cls, nested: bool = False) -> Set:
if nested:
return cls.extract_related_names()
related_names = cls.extract_related_names()
related_names = {
name for name in related_names if cls.Meta.model_fields[name].nullable
}
return related_names
@classmethod
def _update_excluded_with_related_not_required(
cls,
exclude: Union["AbstractSetIntStr", "MappingIntStrAny", None],
nested: bool = False,
) -> Union[Set, Dict]:
exclude = exclude or {}
related_set = cls._exclude_related_names_not_required(nested=nested)
if isinstance(exclude, set):
exclude.union(related_set)
else:
related_dict = translate_list_to_dict(related_set)
exclude = update(related_dict, exclude)
return exclude
@classmethod
def get_names_to_exclude(
cls,
fields: Optional[Union[Dict, Set]] = None,
exclude_fields: Optional[Union[Dict, Set]] = None,
) -> Set:
fields_names = cls.extract_db_own_fields()
if fields and fields is not Ellipsis:
fields_to_keep = {name for name in fields if name in fields_names}
else:
fields_to_keep = fields_names
fields_to_exclude = fields_names - fields_to_keep
if isinstance(exclude_fields, Set):
fields_to_exclude = fields_to_exclude.union(
{name for name in exclude_fields if name in fields_names}
)
elif isinstance(exclude_fields, Dict):
new_to_exclude = {
name
for name in exclude_fields
if name in fields_names and exclude_fields[name] is Ellipsis
}
fields_to_exclude = fields_to_exclude.union(new_to_exclude)
fields_to_exclude = fields_to_exclude - {cls.Meta.pkname}
return fields_to_exclude
pass

10
ormar/models/newbasemodel.py
View File

@ -28,7 +28,6 @@ from pydantic import BaseModel
import ormar # noqa I100
from ormar.exceptions import ModelError
from ormar.fields import BaseField
from ormar.models.excludable import Excludable
from ormar.models.metaclass import ModelMeta, ModelMetaclass
from ormar.models.modelproxy import ModelTableProxy
from ormar.queryset.utils import translate_list_to_dict
@ -47,9 +46,7 @@ if TYPE_CHECKING: # pragma no cover
MappingIntStrAny = Mapping[IntStr, Any]
class NewBaseModel(
pydantic.BaseModel, ModelTableProxy, Excludable, metaclass=ModelMetaclass
):
class NewBaseModel(pydantic.BaseModel, ModelTableProxy, metaclass=ModelMetaclass):
__slots__ = ("_orm_id", "_orm_saved", "_orm", "_pk_column")
if TYPE_CHECKING: # pragma no cover
@ -272,11 +269,10 @@ class NewBaseModel(
continue
return result
@staticmethod
def _skip_ellipsis(
items: Union[Set, Dict, None], key: str
self, items: Union[Set, Dict, None], key: str
) -> Union[Set, Dict, None]:
result = Excludable.get_child(items, key)
result = self.get_child(items, key)
return result if result is not Ellipsis else None
def _extract_nested_models( # noqa: CCR001

6
tests/test_queryset_utils.py
View File

@ -2,15 +2,15 @@ import databases
import sqlalchemy
import ormar
from ormar.models.excludable import Excludable
from ormar.models.mixins import ExcludableMixin
from ormar.queryset.prefetch_query import sort_models
from ormar.queryset.utils import translate_list_to_dict, update_dict_from_list, update
from tests.settings import DATABASE_URL
def test_empty_excludable():
assert Excludable.is_included(None, "key") # all fields included if empty
assert not Excludable.is_excluded(None, "key") # none field excluded if empty
assert ExcludableMixin.is_included(None, "key") # all fields included if empty
assert not ExcludableMixin.is_excluded(None, "key") # none field excluded if empty
def test_list_to_dict_translation():