finish docstrings in models package

This commit is contained in:
collerek
2021-01-01 12:54:38 +01:00
parent c4ff69b683
commit 7a8d11b1c7
4 changed files with 298 additions and 5 deletions

View File

@ -7,10 +7,9 @@ from sqlalchemy.types import TypeDecorator
class UUID(TypeDecorator): # pragma nocover class UUID(TypeDecorator): # pragma nocover
"""Platform-independent GUID type. """
Platform-independent GUID type.
Uses CHAR(36) if in a string mode, otherwise uses CHAR(32), to store UUID. Uses CHAR(36) if in a string mode, otherwise uses CHAR(32), to store UUID.
""" """
impl = CHAR impl = CHAR

View File

@ -1,4 +1,3 @@
import ormar # noqa: I100
from ormar.models.mixins import ( from ormar.models.mixins import (
AliasMixin, AliasMixin,
ExcludableMixin, ExcludableMixin,
@ -11,4 +10,7 @@ from ormar.models.mixins import (
class ModelTableProxy( class ModelTableProxy(
PrefetchQueryMixin, MergeModelMixin, AliasMixin, SavePrepareMixin, ExcludableMixin PrefetchQueryMixin, MergeModelMixin, AliasMixin, SavePrepareMixin, ExcludableMixin
): ):
pass """
Used to combine all mixins with different set of functionalities.
One of the bases of the ormar Model class.
"""

View File

@ -47,6 +47,15 @@ if TYPE_CHECKING: # pragma no cover
class NewBaseModel(pydantic.BaseModel, ModelTableProxy, metaclass=ModelMetaclass): class NewBaseModel(pydantic.BaseModel, ModelTableProxy, metaclass=ModelMetaclass):
"""
Main base class of ormar Model.
Inherits from pydantic BaseModel and has all mixins combined in ModelTableProxy.
Constructed with ModelMetaclass which in turn also inherits pydantic metaclass.
Abstracts away all internals and helper functions, so final Model class has only
the logic concerned with database connection and data persistance.
"""
__slots__ = ("_orm_id", "_orm_saved", "_orm", "_pk_column") __slots__ = ("_orm_id", "_orm_saved", "_orm", "_pk_column")
if TYPE_CHECKING: # pragma no cover if TYPE_CHECKING: # pragma no cover
@ -69,6 +78,37 @@ class NewBaseModel(pydantic.BaseModel, ModelTableProxy, metaclass=ModelMetaclass
# noinspection PyMissingConstructor # noinspection PyMissingConstructor
def __init__(self, *args: Any, **kwargs: Any) -> None: # type: ignore def __init__(self, *args: Any, **kwargs: Any) -> None: # type: ignore
"""
Initializer that creates a new ormar Model that is also pydantic Model at the
same time.
Passed keyword arguments can be only field names and their corresponding values
as those will be passed to pydantic validation that will complain if extra
params are passed.
If relations are defined each relation is expanded and children models are also
initialized and validated. Relation from both sides is registered so you can
access related models from both sides.
Json fields are automatically loaded/dumped if needed.
Models marked as abstract=True in internal Meta class cannot be initialized.
Accepts also special __pk_only__ flag that indicates that Model is constructed
only with primary key value (so no other fields, it's a child model on other
Model), that causes skipping the validation, that's the only case when the
validation can be skipped.
Accepts also special __excluded__ parameter that contains a set of fields that
should be explicitly set to None, as otherwise pydantic will try to populate
them with their default values if default is set.
:raises: ModelError if abstract model is initialized or unknown field is passed
:param args: ignored args
:type args: Any
:param kwargs: keyword arguments - all fields values and some special params
:type kwargs: Any
"""
if self.Meta.abstract: if self.Meta.abstract:
raise ModelError(f"You cannot initialize abstract model {self.get_name()}") raise ModelError(f"You cannot initialize abstract model {self.get_name()}")
object.__setattr__(self, "_orm_id", uuid.uuid4().hex) object.__setattr__(self, "_orm_id", uuid.uuid4().hex)
@ -126,6 +166,32 @@ class NewBaseModel(pydantic.BaseModel, ModelTableProxy, metaclass=ModelMetaclass
) )
def __setattr__(self, name: str, value: Any) -> None: # noqa CCR001 def __setattr__(self, name: str, value: Any) -> None: # noqa CCR001
"""
Overwrites setattr in object to allow for special behaviour of certain params.
Parameter "pk" is translated into actual primary key field name.
Relations are expanded (child model constructed if needed) and registered on
both ends of the relation. The related models are handled by RelationshipManager
exposed at _orm param.
Json fields converted if needed.
Setting pk, foreign key value or any other field value sets Model save status
to False. Setting a reverse relation or many to many relation does not as it
does not modify the state of the model (but related model or through model).
To short circuit all checks and expansions the set of attribute names present
on each model is gathered into _quick_access_fields that is looked first and
if field is in this set the object setattr is called directly.
:param name: name of the attribute to set
:type name: str
:param value: value of the attribute to set
:type value: Any
:return: None
:rtype: None
"""
if name in object.__getattribute__(self, "_quick_access_fields"): if name in object.__getattribute__(self, "_quick_access_fields"):
object.__setattr__(self, name, value) object.__setattr__(self, name, value)
elif name == "pk": elif name == "pk":
@ -152,6 +218,36 @@ class NewBaseModel(pydantic.BaseModel, ModelTableProxy, metaclass=ModelMetaclass
self.set_save_status(False) self.set_save_status(False)
def __getattribute__(self, item: str) -> Any: def __getattribute__(self, item: str) -> Any:
"""
Because we need to overwrite getting the attribute by ormar instead of pydantic
as well as returning related models and not the value stored on the model the
__getattribute__ needs to be used not __getattr__.
It's used to access all attributes so it can be a big overhead that's why a
number of short circuits is used.
To short circuit all checks and expansions the set of attribute names present
on each model is gathered into _quick_access_fields that is looked first and
if field is in this set the object setattr is called directly.
To avoid recursion object's getattribute is used to actually get the attribute
value from the model after the checks.
Even the function calls are constructed with objects functions.
Parameter "pk" is translated into actual primary key field name.
Relations are returned so the actual related model is returned and not current
model's field. The related models are handled by RelationshipManager exposed
at _orm param.
Json fields are converted if needed.
:param item: name of the attribute to retrieve
:type item: str
:return: value of the attribute
:rtype: Any
"""
if item in object.__getattribute__(self, "_quick_access_fields"): if item in object.__getattribute__(self, "_quick_access_fields"):
return object.__getattribute__(self, item) return object.__getattribute__(self, item)
if item == "pk": if item == "pk":
@ -172,16 +268,42 @@ class NewBaseModel(pydantic.BaseModel, ModelTableProxy, metaclass=ModelMetaclass
def _extract_related_model_instead_of_field( def _extract_related_model_instead_of_field(
self, item: str self, item: str
) -> Optional[Union["T", Sequence["T"]]]: ) -> Optional[Union["T", Sequence["T"]]]:
"""
Retrieves the related model/models from RelationshipManager.
:param item: name of the relation
:type item: str
:return: related model, list of related models or None
:rtype: Optional[Union[Model, List[Model]]]
"""
if item in self._orm: if item in self._orm:
return self._orm.get(item) return self._orm.get(item)
return None # pragma no cover return None # pragma no cover
def __eq__(self, other: object) -> bool: def __eq__(self, other: object) -> bool:
"""
Compares other model to this model. when == is called.
:param other: other model to compare
:type other: object
:return: result of comparison
:rtype: bool
"""
if isinstance(other, NewBaseModel): if isinstance(other, NewBaseModel):
return self.__same__(other) return self.__same__(other)
return super().__eq__(other) # pragma no cover return super().__eq__(other) # pragma no cover
def __same__(self, other: "NewBaseModel") -> bool: def __same__(self, other: "NewBaseModel") -> bool:
"""
Used by __eq__, compares other model to this model.
Compares:
* _orm_ids,
* primary key values if it's set
* dictionary of own fields (excluding relations)
:param other: model to compare to
:type other: NewBaseModel
:return: result of comparison
:rtype: bool
"""
return ( return (
self._orm_id == other._orm_id self._orm_id == other._orm_id
or (self.pk == other.pk and self.pk is not None) or (self.pk == other.pk and self.pk is not None)
@ -191,6 +313,14 @@ class NewBaseModel(pydantic.BaseModel, ModelTableProxy, metaclass=ModelMetaclass
@classmethod @classmethod
def get_name(cls, lower: bool = True) -> str: def get_name(cls, lower: bool = True) -> str:
"""
Returns name of the Model class, by default lowercase.
:param lower: flag if name should be set to lowercase
:type lower: bool
:return: name of the model
:rtype: str
"""
name = cls.__name__ name = cls.__name__
if lower: if lower:
name = name.lower() name = name.lower()
@ -198,6 +328,14 @@ class NewBaseModel(pydantic.BaseModel, ModelTableProxy, metaclass=ModelMetaclass
@property @property
def pk_column(self) -> sqlalchemy.Column: def pk_column(self) -> sqlalchemy.Column:
"""
Retrieves primary key sqlalchemy column from models Meta.table.
Each model has to have primary key.
Only one primary key column is allowed.
:return: primary key sqlalchemy column
:rtype: sqlalchemy.Column
"""
if object.__getattribute__(self, "_pk_column") is not None: if object.__getattribute__(self, "_pk_column") is not None:
return object.__getattribute__(self, "_pk_column") return object.__getattribute__(self, "_pk_column")
pk_columns = self.Meta.table.primary_key.columns.values() pk_columns = self.Meta.table.primary_key.columns.values()
@ -207,30 +345,51 @@ class NewBaseModel(pydantic.BaseModel, ModelTableProxy, metaclass=ModelMetaclass
@property @property
def saved(self) -> bool: def saved(self) -> bool:
"""Saved status of the model. Changed by setattr and loading from db"""
return self._orm_saved return self._orm_saved
@property @property
def signals(self) -> "SignalEmitter": def signals(self) -> "SignalEmitter":
"""Exposes signals from model Meta"""
return self.Meta.signals return self.Meta.signals
@classmethod @classmethod
def pk_type(cls) -> Any: def pk_type(cls) -> Any:
"""Shortcut to models primary key field type"""
return cls.Meta.model_fields[cls.Meta.pkname].__type__ return cls.Meta.model_fields[cls.Meta.pkname].__type__
@classmethod @classmethod
def db_backend_name(cls) -> str: def db_backend_name(cls) -> str:
"""Shortcut to database dialect,
cause some dialect require different treatment"""
return cls.Meta.database._backend._dialect.name return cls.Meta.database._backend._dialect.name
def remove(self, parent: "T", name: str) -> None: def remove(self, parent: "T", name: str) -> None:
"""Removes child from relation with given name in RelationshipManager"""
self._orm.remove_parent(self, parent, name) self._orm.remove_parent(self, parent, name)
def set_save_status(self, status: bool) -> None: def set_save_status(self, status: bool) -> None:
"""Sets value of the save status"""
object.__setattr__(self, "_orm_saved", status) object.__setattr__(self, "_orm_saved", status)
@classmethod @classmethod
def get_properties( def get_properties(
cls, include: Union[Set, Dict, None], exclude: Union[Set, Dict, None] cls, include: Union[Set, Dict, None], exclude: Union[Set, Dict, None]
) -> Set[str]: ) -> Set[str]:
"""
Returns a set of names of functions/fields decorated with
@property_field decorator.
They are added to dictionary when called directly and therefore also are
present in fastapi responses.
:param include: fields to include
:type include: Union[Set, Dict, None]
:param exclude: fields to exclude
:type exclude: Union[Set, Dict, None]
:return: set of property fields names
:rtype: Set[str]
"""
props = cls.Meta.property_fields props = cls.Meta.property_fields
if include: if include:
@ -242,6 +401,16 @@ class NewBaseModel(pydantic.BaseModel, ModelTableProxy, metaclass=ModelMetaclass
def _get_related_not_excluded_fields( def _get_related_not_excluded_fields(
self, include: Optional[Dict], exclude: Optional[Dict], self, include: Optional[Dict], exclude: Optional[Dict],
) -> List: ) -> List:
"""
Returns related field names applying on them include and exclude set.
:param include: fields to include
:type include: Union[Set, Dict, None]
:param exclude: fields to exclude
:type exclude: Union[Set, Dict, None]
:return:
:rtype: List of fields with relations that is not excluded
"""
fields = [field for field in self.extract_related_names()] fields = [field for field in self.extract_related_names()]
if include: if include:
fields = [field for field in fields if field in include] fields = [field for field in fields if field in include]
@ -259,6 +428,18 @@ class NewBaseModel(pydantic.BaseModel, ModelTableProxy, metaclass=ModelMetaclass
include: Union[Set, Dict, None], include: Union[Set, Dict, None],
exclude: Union[Set, Dict, None], exclude: Union[Set, Dict, None],
) -> List: ) -> List:
"""
Converts list of models into list of dictionaries.
:param models: List of models
:type models: List
:param include: fields to include
:type include: Union[Set, Dict, None]
:param exclude: fields to exclude
:type exclude: Union[Set, Dict, None]
:return: list of models converted to dictionaries
:rtype: List[Dict]
"""
result = [] result = []
for model in models: for model in models:
try: try:
@ -272,6 +453,18 @@ class NewBaseModel(pydantic.BaseModel, ModelTableProxy, metaclass=ModelMetaclass
def _skip_ellipsis( def _skip_ellipsis(
self, items: Union[Set, Dict, None], key: str self, items: Union[Set, Dict, None], key: str
) -> Union[Set, Dict, None]: ) -> Union[Set, Dict, None]:
"""
Helper to traverse the include/exclude dictionaries.
In dict() Ellipsis should be skipped as it indicates all fields required
and not the actual set/dict with fields names.
:param items: current include/exclude value
:type items: Union[Set, Dict, None]
:param key: key for nested relations to check
:type key: str
:return: nested value of the items
:rtype: Union[Set, Dict, None]
"""
result = self.get_child(items, key) result = self.get_child(items, key)
return result if result is not Ellipsis else None return result if result is not Ellipsis else None
@ -282,6 +475,21 @@ class NewBaseModel(pydantic.BaseModel, ModelTableProxy, metaclass=ModelMetaclass
include: Optional[Dict], include: Optional[Dict],
exclude: Optional[Dict], exclude: Optional[Dict],
) -> Dict: ) -> Dict:
"""
Traverse nested models and converts them into dictionaries.
Calls itself recursively if needed.
:param nested: flag if current instance is nested
:type nested: bool
:param dict_instance: current instance dict
:type dict_instance: Dict
:param include: fields to include
:type include: Optional[Dict]
:param exclude: fields to exclude
:type exclude: Optional[Dict]
:return: current model dict with child models converted to dictionaries
:rtype: Dict
"""
fields = self._get_related_not_excluded_fields(include=include, exclude=exclude) fields = self._get_related_not_excluded_fields(include=include, exclude=exclude)
@ -317,6 +525,34 @@ class NewBaseModel(pydantic.BaseModel, ModelTableProxy, metaclass=ModelMetaclass
exclude_none: bool = False, exclude_none: bool = False,
nested: bool = False, nested: bool = False,
) -> "DictStrAny": # noqa: A003' ) -> "DictStrAny": # noqa: A003'
"""
Generate a dictionary representation of the model,
optionally specifying which fields to include or exclude.
Nested models are also parsed to dictionaries.
Additionally fields decorated with @property_field are also added.
:param include: fields to include
:type include: Union[Set, Dict, None]
:param exclude: fields to exclude
:type exclude: Union[Set, Dict, None]
:param by_alias: flag to get values by alias - passed to pydantic
:type by_alias: bool
:param skip_defaults: flag to not set values - passed to pydantic
:type skip_defaults: bool
:param exclude_unset: flag to exclude not set values - passed to pydantic
:type exclude_unset: bool
:param exclude_defaults: flag to exclude default values - passed to pydantic
:type exclude_defaults: bool
:param exclude_none: flag to exclude None values - passed to pydantic
:type exclude_none: bool
:param nested: flag if the current model is nested
:type nested: bool
:return:
:rtype:
"""
dict_instance = super().dict( dict_instance = super().dict(
include=include, include=include,
exclude=self._update_excluded_with_related_not_required(exclude, nested), exclude=self._update_excluded_with_related_not_required(exclude, nested),
@ -348,11 +584,31 @@ class NewBaseModel(pydantic.BaseModel, ModelTableProxy, metaclass=ModelMetaclass
return dict_instance return dict_instance
def update_from_dict(self, value_dict: Dict) -> "NewBaseModel": def update_from_dict(self, value_dict: Dict) -> "NewBaseModel":
"""
Updates self with values of fields passed in the dictionary.
:param value_dict: dictionary of fields names and values
:type value_dict: Dict
:return: self
:rtype: NewBaseModel
"""
for key, value in value_dict.items(): for key, value in value_dict.items():
setattr(self, key, value) setattr(self, key, value)
return self return self
def _convert_json(self, column_name: str, value: Any, op: str) -> Union[str, Dict]: def _convert_json(self, column_name: str, value: Any, op: str) -> Union[str, Dict]:
"""
Converts value to/from json if needed (for Json columns).
:param column_name: name of the field
:type column_name: str
:param value: value fo the field
:type value: Any
:param op: operator on json
:type op: str
:return: converted value if needed, else original value
:rtype: Any
"""
if not self._is_conversion_to_json_needed(column_name): if not self._is_conversion_to_json_needed(column_name):
return value return value
@ -371,17 +627,41 @@ class NewBaseModel(pydantic.BaseModel, ModelTableProxy, metaclass=ModelMetaclass
return value.decode("utf-8") if isinstance(value, bytes) else value return value.decode("utf-8") if isinstance(value, bytes) else value
def _is_conversion_to_json_needed(self, column_name: str) -> bool: def _is_conversion_to_json_needed(self, column_name: str) -> bool:
"""
Checks if given column name is related to JSON field.
:param column_name: name of the field
:type column_name: str
:return: result of the check
:rtype: bool
"""
return ( return (
column_name in self.Meta.model_fields column_name in self.Meta.model_fields
and self.Meta.model_fields[column_name].__type__ == pydantic.Json and self.Meta.model_fields[column_name].__type__ == pydantic.Json
) )
def _extract_own_model_fields(self) -> Dict: def _extract_own_model_fields(self) -> Dict:
"""
Returns a dictionary with field names and values for fields that are not
relations fields (ForeignKey, ManyToMany etc.)
:return: dictionary of fields names and values.
:rtype: Dict
"""
related_names = self.extract_related_names() related_names = self.extract_related_names()
self_fields = self.dict(exclude=related_names) self_fields = self.dict(exclude=related_names)
return self_fields return self_fields
def _extract_model_db_fields(self) -> Dict: def _extract_model_db_fields(self) -> Dict:
"""
Returns a dictionary with field names and values for fields that are stored in
current model's table.
That includes own non-relational fields ang foreign key fields.
:return: dictionary of fields names and values.
:rtype: Dict
"""
self_fields = self._extract_own_model_fields() self_fields = self._extract_own_model_fields()
self_fields = { self_fields = {
k: v k: v
@ -395,6 +675,14 @@ class NewBaseModel(pydantic.BaseModel, ModelTableProxy, metaclass=ModelMetaclass
return self_fields return self_fields
def get_relation_model_id(self, target_field: Type["BaseField"]) -> Optional[int]: def get_relation_model_id(self, target_field: Type["BaseField"]) -> Optional[int]:
"""
Returns an id of the relation side model to use in prefetch query.
:param target_field: field with relation definition
:type target_field: Type["BaseField"]
:return: value of pk if set
:rtype: Optional[int]
"""
if target_field.virtual or issubclass( if target_field.virtual or issubclass(
target_field, ormar.fields.ManyToManyField target_field, ormar.fields.ManyToManyField
): ):

View File

@ -1,3 +1,7 @@
"""
Contains set of fields/methods etc names that are used to bypass the checks in
NewBaseModel __getattribute__ calls to speed the calls.
"""
quick_access_set = { quick_access_set = {
"Config", "Config",
"Meta", "Meta",