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

added mypy checks and some typehint changes to conform

Browse Source
This commit is contained in:
collerek
2020-09-29 14:05:08 +02:00
parent 6d56ea5e30
commit 3caa87057e
23 changed files with 274 additions and 202 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
ormar/relations/alias_manager.py
View File

@ -1,7 +1,7 @@
import string
import uuid
from random import choices
from typing import List
from typing import List, Dict
import sqlalchemy
from sqlalchemy import text
@ -14,7 +14,7 @@ def get_table_alias() -> str:
class AliasManager:
def __init__(self) -> None:
self._aliases = dict()
self._aliases: Dict[str, str] = dict()
@staticmethod
def prefixed_columns(alias: str, table: sqlalchemy.Table) -> List[text]:

4
ormar/relations/querysetproxy.py
View File

@ -13,8 +13,8 @@ class QuerysetProxy:
relation: "Relation"
def __init__(self, relation: "Relation") -> None:
self.relation = relation
self.queryset = None
self.relation: Relation = relation
self.queryset: "QuerySet"
def _assign_child_to_parent(self, child: "Model") -> None:
owner = self.relation._owner

33
ormar/relations/relation.py
View File

@ -9,6 +9,7 @@ from ormar.relations.relation_proxy import RelationProxy
if TYPE_CHECKING: # pragma no cover
from ormar import Model
from ormar.relations import RelationsManager
from ormar.models import NewBaseModel
class RelationType(Enum):
@ -19,24 +20,26 @@ class RelationType(Enum):
class Relation:
def __init__(
self,
manager: "RelationsManager",
type_: RelationType,
to: Type["Model"],
through: Type["Model"] = None,
self,
manager: "RelationsManager",
type_: RelationType,
to: Type["Model"],
through: Type["Model"] = None,
) -> None:
self.manager = manager
self._owner = manager.owner
self._type = type_
self.to = to
self.through = through
self.related_models = (
self._owner: "Model" = manager.owner
self._type: RelationType = type_
self.to: Type["Model"] = to
self.through: Optional[Type["Model"]] = through
self.related_models: Optional[Union[RelationProxy, "Model"]] = (
RelationProxy(relation=self)
if type_ in (RelationType.REVERSE, RelationType.MULTIPLE)
else None
)
def _find_existing(self, child: "Model") -> Optional[int]:
if not isinstance(self.related_models, RelationProxy): # pragma nocover
raise ValueError("Cannot find existing models in parent relation type")
for ind, relation_child in enumerate(self.related_models[:]):
try:
if relation_child == child:
@ -52,7 +55,7 @@ class Relation:
self._owner.__dict__[relation_name] = child
else:
if self._find_existing(child) is None:
self.related_models.append(child)
self.related_models.append(child) # type: ignore
rel = self._owner.__dict__.get(relation_name, [])
rel = rel or []
if not isinstance(rel, list):
@ -60,19 +63,19 @@ class Relation:
rel.append(child)
self._owner.__dict__[relation_name] = rel
def remove(self, child: "Model") -> None:
def remove(self, child: Union["NewBaseModel", Type["NewBaseModel"]]) -> None:
relation_name = self._owner.resolve_relation_name(self._owner, child)
if self._type == RelationType.PRIMARY:
if self.related_models.__same__(child):
if self.related_models == child:
self.related_models = None
del self._owner.__dict__[relation_name]
else:
position = self._find_existing(child)
if position is not None:
self.related_models.pop(position)
self.related_models.pop(position) # type: ignore
del self._owner.__dict__[relation_name][position]
def get(self) -> Union[List["Model"], "Model"]:
def get(self) -> Optional[Union[List["Model"], "Model"]]:
return self.related_models
def __repr__(self) -> str: # pragma no cover

38
ormar/relations/relation_manager.py
View File

@ -1,6 +1,7 @@
from typing import List, Optional, TYPE_CHECKING, Type, Union
from typing import List, Optional, TYPE_CHECKING, Type, Union, Dict
from weakref import proxy
from ormar.fields import BaseField
from ormar.fields.foreign_key import ForeignKeyField
from ormar.fields.many_to_many import ManyToManyField
from ormar.relations.relation import Relation, RelationType
@ -11,25 +12,28 @@ from ormar.relations.utils import (
if TYPE_CHECKING: # pragma no cover
from ormar import Model
from ormar.models import NewBaseModel
class RelationsManager:
def __init__(
self, related_fields: List[Type[ForeignKeyField]] = None, owner: "Model" = None
self,
related_fields: List[Type[ForeignKeyField]] = None,
owner: "NewBaseModel" = None,
) -> None:
self.owner = proxy(owner)
self._related_fields = related_fields or []
self._related_names = [field.name for field in self._related_fields]
self._relations = dict()
self._relations: Dict[str, Relation] = dict()
for field in self._related_fields:
self._add_relation(field)
def _get_relation_type(self, field: Type[ForeignKeyField]) -> RelationType:
def _get_relation_type(self, field: Type[BaseField]) -> RelationType:
if issubclass(field, ManyToManyField):
return RelationType.MULTIPLE
return RelationType.PRIMARY if not field.virtual else RelationType.REVERSE
def _add_relation(self, field: Type[ForeignKeyField]) -> None:
def _add_relation(self, field: Type[BaseField]) -> None:
self._relations[field.name] = Relation(
manager=self,
type_=self._get_relation_type(field),
@ -44,15 +48,17 @@ class RelationsManager:
relation = self._relations.get(name, None)
if relation is not None:
return relation.get()
return None # pragma nocover
def _get(self, name: str) -> Optional[Relation]:
relation = self._relations.get(name, None)
if relation is not None:
return relation
return None
@staticmethod
def add(parent: "Model", child: "Model", child_name: str, virtual: bool) -> None:
to_field = child.resolve_relation_field(child, parent)
to_field: Type[BaseField] = child.resolve_relation_field(child, parent)
(parent, child, child_name, to_name,) = get_relations_sides_and_names(
to_field, parent, child, child_name, virtual
@ -61,18 +67,22 @@ class RelationsManager:
parent_relation = parent._orm._get(child_name)
if not parent_relation:
parent_relation = register_missing_relation(parent, child, child_name)
parent_relation.add(child)
child._orm._get(to_name).add(parent)
parent_relation.add(child) # type: ignore
def remove(self, name: str, child: "Model") -> None:
child_relation = child._orm._get(to_name)
if child_relation:
child_relation.add(parent)
def remove(self, name: str, child: Union["NewBaseModel", Type["NewBaseModel"]]) -> None:
relation = self._get(name)
relation.remove(child)
if relation:
relation.remove(child)
@staticmethod
def remove_parent(item: "Model", name: Union[str, "Model"]) -> None:
def remove_parent(item: Union["NewBaseModel", Type["NewBaseModel"]], name: "Model") -> None:
related_model = name
name = item.resolve_relation_name(item, related_model)
if name in item._orm:
rel_name = item.resolve_relation_name(item, related_model)
if rel_name in item._orm:
relation_name = item.resolve_relation_name(related_model, item)
item._orm.remove(name, related_model)
item._orm.remove(rel_name, related_model)
related_model._orm.remove(relation_name, item)

29
ormar/relations/relation_proxy.py
View File

@ -13,22 +13,30 @@ if TYPE_CHECKING: # pragma no cover
class RelationProxy(list):
def __init__(self, relation: "Relation") -> None:
super(RelationProxy, self).__init__()
self.relation = relation
self._owner = self.relation.manager.owner
self.relation: Relation = relation
self._owner: "Model" = self.relation.manager.owner
self.queryset_proxy = QuerysetProxy(relation=self.relation)
def __getattribute__(self, item: str) -> Any:
if item in ["count", "clear"]:
if not self.queryset_proxy.queryset:
self.queryset_proxy.queryset = self._set_queryset()
self._initialize_queryset()
return getattr(self.queryset_proxy, item)
return super().__getattribute__(item)
def __getattr__(self, item: str) -> Any:
if not self.queryset_proxy.queryset:
self.queryset_proxy.queryset = self._set_queryset()
self._initialize_queryset()
return getattr(self.queryset_proxy, item)
def _initialize_queryset(self) -> None:
if not self._check_if_queryset_is_initialized():
self.queryset_proxy.queryset = self._set_queryset()
def _check_if_queryset_is_initialized(self) -> bool:
return (
hasattr(self.queryset_proxy, "queryset")
and self.queryset_proxy.queryset is not None
)
def _set_queryset(self) -> "QuerySet":
owner_table = self.relation._owner.Meta.tablename
pkname = self.relation._owner.Meta.pkname
@ -45,10 +53,15 @@ class RelationProxy(list):
)
return queryset
async def remove(self, item: "Model") -> None:
async def remove(self, item: "Model") -> None: # type: ignore
super().remove(item)
rel_name = item.resolve_relation_name(item, self._owner)
item._orm._get(rel_name).remove(self._owner)
relation = item._orm._get(rel_name)
if relation is None: # pragma nocover
raise ValueError(
f"{self._owner.get_name()} does not have relation {rel_name}"
)
relation.remove(self._owner)
if self.relation._type == ormar.RelationType.MULTIPLE:
await self.queryset_proxy.delete_through_instance(item)

8
ormar/relations/utils.py
View File

@ -1,8 +1,8 @@
from typing import TYPE_CHECKING, Tuple, Type
from typing import TYPE_CHECKING, Tuple, Type, Optional
from weakref import proxy
import ormar
from ormar.fields.foreign_key import ForeignKeyField
from ormar.fields import BaseField
from ormar.fields.many_to_many import ManyToManyField
from ormar.relations import Relation
@ -12,7 +12,7 @@ if TYPE_CHECKING: # pragma no cover
def register_missing_relation(
parent: "Model", child: "Model", child_name: str
) -> Relation:
) -> Optional[Relation]:
ormar.models.expand_reverse_relationships(child.__class__)
name = parent.resolve_relation_name(parent, child)
field = parent.Meta.model_fields[name]
@ -22,7 +22,7 @@ def register_missing_relation(
def get_relations_sides_and_names(
to_field: Type[ForeignKeyField],
to_field: Type[BaseField],
parent: "Model",
child: "Model",
child_name: str,