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

clean code

Browse Source
This commit is contained in:
collerek
2020-08-11 19:54:54 +02:00
parent 45653d36c7
commit 4aadc9fac6
9 changed files with 88 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files

BIN
.coverage
View File

Binary file not shown.

5
orm/models/__init__.py
View File

@ -1,5 +1,4 @@
from orm.models.fakepydantic import FakePydantic
from orm.models.model import Model
__all__ = [
"Model"
]
__all__ = ["FakePydantic", "Model"]

28
orm/models/fakepydantic.py
View File

@ -1,19 +1,29 @@
import inspect
import json
import uuid
from typing import TYPE_CHECKING, Dict, TypeVar, Type, Any, Optional, Callable, Set, List
from typing import (
Any,
Callable,
Dict,
List,
Optional,
Set,
TYPE_CHECKING,
Type,
TypeVar,
)
import databases
import pydantic
import sqlalchemy
from pydantic import BaseModel
import orm
import orm # noqa I100
from orm.fields import BaseField
from orm.models.metaclass import ModelMetaclass
from orm.relations import RelationshipManager
if TYPE_CHECKING: #pragma no cover
if TYPE_CHECKING: # pragma no cover
from orm.models.model import Model
@ -74,9 +84,9 @@ class FakePydantic(list, metaclass=ModelMetaclass):
item = getattr(self.values, key, None)
if (
item is not None
and self._is_conversion_to_json_needed(key)
and isinstance(item, str)
item is not None
and self._is_conversion_to_json_needed(key)
and isinstance(item, str)
):
try:
item = json.loads(item)
@ -92,7 +102,7 @@ class FakePydantic(list, metaclass=ModelMetaclass):
if self.__class__ != other.__class__: # pragma no cover
return False
return self._orm_id == other._orm_id or (
self.values is not None and other.values is not None and self.pk == other.pk
self.values is not None and other.values is not None and self.pk == other.pk
)
def __repr__(self) -> str: # pragma no cover
@ -148,7 +158,7 @@ class FakePydantic(list, metaclass=ModelMetaclass):
related_names = set()
for name, field in cls.__fields__.items():
if inspect.isclass(field.type_) and issubclass(
field.type_, pydantic.BaseModel
field.type_, pydantic.BaseModel
):
related_names.add(name)
return related_names
@ -180,7 +190,7 @@ class FakePydantic(list, metaclass=ModelMetaclass):
for field in one.__model_fields__.keys():
# print(field, one.dict(), other.dict())
if isinstance(getattr(one, field), list) and not isinstance(
getattr(one, field), orm.Model
getattr(one, field), orm.Model
):
setattr(other, field, getattr(one, field) + getattr(other, field))
elif isinstance(getattr(one, field), orm.Model):

7
orm/models/metaclass.py
View File

@ -1,14 +1,17 @@
import copy
from typing import Dict, Tuple, Type, Optional, List, Any
from typing import Any, Dict, List, Optional, TYPE_CHECKING, Tuple, Type
import sqlalchemy
from pydantic import BaseConfig, create_model
from pydantic.fields import ModelField
from orm import ForeignKey, ModelDefinitionError
from orm import ForeignKey, ModelDefinitionError # noqa I100
from orm.fields import BaseField
from orm.relations import RelationshipManager
if TYPE_CHECKING: # pragma no cover
from orm import Model
relationship_manager = RelationshipManager()

8
orm/models/model.py
View File

@ -1,15 +1,15 @@
from typing import List, Any
from typing import Any, List
import sqlalchemy
import orm.queryset.queryset
from orm.models.fakepydantic import FakePydantic
import orm.queryset # noqa I100
from orm.models import FakePydantic # noqa I100
class Model(FakePydantic):
__abstract__ = True
objects = orm.queryset.queryset.QuerySet()
objects = orm.queryset.QuerySet()
@classmethod
def from_row(

3
orm/queryset/__init__.py
View File

@ -0,0 +1,3 @@
from orm.queryset.queryset import QuerySet
__all__ = ["QuerySet"]

26
orm/queryset/clause.py
View File

@ -1,9 +1,9 @@
from typing import Type, List, Any, Tuple, Dict, Union, Optional, TYPE_CHECKING
from typing import Any, Dict, List, Optional, TYPE_CHECKING, Tuple, Type, Union
import sqlalchemy
from sqlalchemy import text
import orm
import orm # noqa I100
from orm.exceptions import QueryDefinitionError
if TYPE_CHECKING: # pragma no cover
@ -25,7 +25,7 @@ ESCAPE_CHARACTERS = ["%", "_"]
class QueryClause:
def __init__(
self, model_cls: Type["Model"], filter_clauses: List, select_related: List,
self, model_cls: Type["Model"], filter_clauses: List, select_related: List,
) -> None:
self._select_related = select_related
@ -35,7 +35,7 @@ class QueryClause:
self.table = self.model_cls.__table__
def filter( # noqa: A003
self, **kwargs: Any
self, **kwargs: Any
) -> Tuple[List[sqlalchemy.sql.expression.TextClause], List[str]]:
filter_clauses = self.filter_clauses
select_related = list(self._select_related)
@ -92,7 +92,7 @@ class QueryClause:
return filter_clauses, select_related
def _determine_filter_target_table(
self, related_parts: List[str], select_related: List[str]
self, related_parts: List[str], select_related: List[str]
) -> Tuple[List[str], str, "Model"]:
table_prefix = ""
@ -116,12 +116,12 @@ class QueryClause:
return select_related, table_prefix, model_cls
def _compile_clause(
self,
clause: sqlalchemy.sql.expression.BinaryExpression,
column: sqlalchemy.Column,
table: sqlalchemy.Table,
table_prefix: str,
modifiers: Dict,
self,
clause: sqlalchemy.sql.expression.BinaryExpression,
column: sqlalchemy.Column,
table: sqlalchemy.Table,
table_prefix: str,
modifiers: Dict,
) -> sqlalchemy.sql.expression.TextClause:
for modifier, modifier_value in modifiers.items():
clause.modifiers[modifier] = modifier_value
@ -140,7 +140,7 @@ class QueryClause:
@staticmethod
def _escape_characters_in_clause(
op: str, value: Union[str, "Model"]
op: str, value: Union[str, "Model"]
) -> Tuple[str, bool]:
has_escaped_character = False
@ -162,7 +162,7 @@ class QueryClause:
@staticmethod
def _extract_operator_field_and_related(
parts: List[str],
parts: List[str],
) -> Tuple[str, str, Optional[List]]:
if parts[-1] in FILTER_OPERATORS:
op = parts[-1]

58
orm/queryset/query.py
View File

@ -1,9 +1,9 @@
from typing import NamedTuple, Type, List, Tuple, TYPE_CHECKING
from typing import List, NamedTuple, TYPE_CHECKING, Tuple, Type
import sqlalchemy
from sqlalchemy import text
import orm
import orm # noqa I100
from orm import ForeignKey
from orm.fields import BaseField
@ -20,12 +20,12 @@ class JoinParameters(NamedTuple):
class Query:
def __init__(
self,
model_cls: Type["Model"],
filter_clauses: List,
select_related: List,
limit_count: int,
offset: int,
self,
model_cls: Type["Model"],
filter_clauses: List,
select_related: List,
limit_count: int,
offset: int,
) -> None:
self.query_offset = offset
@ -50,12 +50,12 @@ class Query:
for key in self.model_cls.__model_fields__:
if (
not self.model_cls.__model_fields__[key].nullable
and isinstance(
self.model_cls.__model_fields__[key],
orm.fields.foreign_key.ForeignKey,
)
and key not in self._select_related
not self.model_cls.__model_fields__[key].nullable
and isinstance(
self.model_cls.__model_fields__[key],
orm.fields.foreign_key.ForeignKey,
)
and key not in self._select_related
):
self._select_related = [key] + self._select_related
@ -97,12 +97,12 @@ class Query:
@staticmethod
def _field_is_a_foreign_key_and_no_circular_reference(
field: BaseField, field_name: str, rel_part: str
field: BaseField, field_name: str, rel_part: str
) -> bool:
return isinstance(field, ForeignKey) and field_name not in rel_part
def _field_qualifies_to_deeper_search(
self, field: ForeignKey, parent_virtual: bool, nested: bool, rel_part: str
self, field: ForeignKey, parent_virtual: bool, nested: bool, rel_part: str
) -> bool:
prev_part_of_related = "__".join(rel_part.split("__")[:-1])
partial_match = any(
@ -110,19 +110,19 @@ class Query:
)
already_checked = any([x.startswith(rel_part) for x in self.auto_related])
return (
(field.virtual and parent_virtual)
or (partial_match and not already_checked)
) or not nested
(field.virtual and parent_virtual)
or (partial_match and not already_checked)
) or not nested
def on_clause(
self, previous_alias: str, alias: str, from_clause: str, to_clause: str,
self, previous_alias: str, alias: str, from_clause: str, to_clause: str,
) -> text:
left_part = f"{alias}_{to_clause}"
right_part = f"{previous_alias + '_' if previous_alias else ''}{from_clause}"
return text(f"{left_part}={right_part}")
def _build_join_parameters(
self, part: str, join_params: JoinParameters
self, part: str, join_params: JoinParameters
) -> JoinParameters:
model_cls = join_params.model_cls.__model_fields__[part].to
to_table = model_cls.__table__.name
@ -165,15 +165,15 @@ class Query:
return JoinParameters(prev_model, previous_alias, from_table, model_cls)
def _extract_auto_required_relations(
self,
prev_model: Type["Model"],
rel_part: str = "",
nested: bool = False,
parent_virtual: bool = False,
self,
prev_model: Type["Model"],
rel_part: str = "",
nested: bool = False,
parent_virtual: bool = False,
) -> None:
for field_name, field in prev_model.__model_fields__.items():
if self._field_is_a_foreign_key_and_no_circular_reference(
field, field_name, rel_part
field, field_name, rel_part
):
rel_part = field_name if not rel_part else rel_part + "__" + field_name
if not field.nullable:
@ -181,7 +181,7 @@ class Query:
self.auto_related.append("__".join(rel_part.split("__")[:-1]))
rel_part = ""
elif self._field_qualifies_to_deeper_search(
field, parent_virtual, nested, rel_part
field, parent_virtual, nested, rel_part
):
self._extract_auto_required_relations(
prev_model=field.to,
@ -201,7 +201,7 @@ class Query:
self._select_related = new_joins + self.auto_related
def _apply_expression_modifiers(
self, expr: sqlalchemy.sql.select
self, expr: sqlalchemy.sql.select
) -> sqlalchemy.sql.select:
if self.filter_clauses:
if len(self.filter_clauses) == 1:

22
orm/queryset/queryset.py
View File

@ -1,10 +1,10 @@
from typing import Type, List, Any, Union, Tuple, TYPE_CHECKING
from typing import Any, List, TYPE_CHECKING, Tuple, Type, Union
import databases
import sqlalchemy
import orm # noqa I100
from orm import NoMatch, MultipleMatches
from orm import MultipleMatches, NoMatch
from orm.queryset.clause import QueryClause
from orm.queryset.query import Query
@ -14,12 +14,12 @@ if TYPE_CHECKING: # pragma no cover
class QuerySet:
def __init__(
self,
model_cls: Type["Model"] = None,
filter_clauses: List = None,
select_related: List = None,
limit_count: int = None,
offset: int = None,
self,
model_cls: Type["Model"] = None,
filter_clauses: List = None,
select_related: List = None,
limit_count: int = None,
offset: int = None,
) -> None:
self.model_cls = model_cls
self.filter_clauses = [] if filter_clauses is None else filter_clauses
@ -151,9 +151,9 @@ class QuerySet:
pkname = self.model_cls.__pkname__
pk = self.model_cls.__model_fields__[pkname]
if (
pkname in new_kwargs
and new_kwargs.get(pkname) is None
and (pk.nullable or pk.autoincrement)
pkname in new_kwargs
and new_kwargs.get(pkname) is None
and (pk.nullable or pk.autoincrement)
):
del new_kwargs[pkname]