remove .vscode settings, re-dump orjson choices to fix choices, move mypy config into pyproject.toml

This commit is contained in:
collerek
2021-10-08 20:14:03 +02:00
parent b2541bed1e
commit b637fc0774
14 changed files with 69 additions and 45 deletions

View File

@ -6,7 +6,7 @@ from typing import Any, Callable, Optional, TYPE_CHECKING, Type, Union
import sqlalchemy.types as types
from pydantic.utils import lenient_issubclass
from sqlalchemy.engine.default import DefaultDialect
from sqlalchemy.engine import Dialect
import ormar # noqa: I100, I202
from ormar import ModelDefinitionError # noqa: I202, I100
@ -146,14 +146,14 @@ class EncryptedString(types.TypeDecorator):
def __repr__(self) -> str: # pragma: nocover
return "TEXT()"
def load_dialect_impl(self, dialect: DefaultDialect) -> Any:
def load_dialect_impl(self, dialect: Dialect) -> Any:
return dialect.type_descriptor(types.TEXT())
def _refresh(self) -> None:
key = self._key() if callable(self._key) else self._key
self.backend._refresh(key)
def process_bind_param(self, value: Any, dialect: DefaultDialect) -> Optional[str]:
def process_bind_param(self, value: Any, dialect: Dialect) -> Optional[str]:
if value is None:
return value
self._refresh()
@ -167,7 +167,7 @@ class EncryptedString(types.TypeDecorator):
encrypted_value = self.backend.encrypt(value)
return encrypted_value
def process_result_value(self, value: Any, dialect: DefaultDialect) -> Any:
def process_result_value(self, value: Any, dialect: Dialect) -> Any:
if value is None:
return value
self._refresh()

View File

@ -2,7 +2,7 @@ import uuid
from typing import Any, Optional
from sqlalchemy import CHAR
from sqlalchemy.engine.default import DefaultDialect
from sqlalchemy.engine import Dialect
from sqlalchemy.types import TypeDecorator
@ -25,22 +25,20 @@ class UUID(TypeDecorator):
return "CHAR(36)"
return "CHAR(32)"
def load_dialect_impl(self, dialect: DefaultDialect) -> Any:
def load_dialect_impl(self, dialect: Dialect) -> Any:
return (
dialect.type_descriptor(CHAR(36))
if self.uuid_format == "string"
else dialect.type_descriptor(CHAR(32))
)
def process_bind_param(
self, value: uuid.UUID, dialect: DefaultDialect
) -> Optional[str]:
def process_bind_param(self, value: uuid.UUID, dialect: Dialect) -> Optional[str]:
if value is None:
return value
return str(value) if self.uuid_format == "string" else "%.32x" % value.int
def process_result_value(
self, value: Optional[str], dialect: DefaultDialect
self, value: Optional[str], dialect: Dialect
) -> Optional[uuid.UUID]:
if value is None:
return value

View File

@ -64,8 +64,11 @@ def convert_choices_if_needed( # noqa: CCR001
value = value.isoformat() if not isinstance(value, str) else value
choices = [o.isoformat() for o in field.choices]
elif field.__type__ == pydantic.Json:
value = json.dumps(value) if not isinstance(value, str) else value
value = (
json.dumps(value) if not isinstance(value, str) else re_dump_value(value)
)
value = value.decode("utf-8") if isinstance(value, bytes) else value
choices = [re_dump_value(x) for x in field.choices]
elif field.__type__ == uuid.UUID:
value = str(value) if not isinstance(value, str) else value
choices = [str(o) for o in field.choices]
@ -86,6 +89,21 @@ def convert_choices_if_needed( # noqa: CCR001
return value, choices
def re_dump_value(value: str) -> str:
"""
Rw-dumps choices due to different string representation in orjson and json
:param value: string to re-dump
:type value: str
:return: re-dumped choices
:rtype: List[str]
"""
try:
result: Union[str, bytes] = json.dumps(json.loads(value))
except json.JSONDecodeError:
result = value
return result.decode("utf-8") if isinstance(result, bytes) else result
def validate_choices(field: "BaseField", value: Any) -> None:
"""
Validates if given value is in provided choices.

View File

@ -1,5 +1,15 @@
import uuid
from typing import Callable, Collection, Dict, List, Optional, Set, TYPE_CHECKING, cast
from typing import (
Any,
Callable,
Collection,
Dict,
List,
Optional,
Set,
TYPE_CHECKING,
cast,
)
import ormar
from ormar.exceptions import ModelPersistenceError
@ -93,7 +103,7 @@ class SavePrepareMixin(RelationMixin, AliasMixin):
if field.__type__ == uuid.UUID and name in model_dict:
parsers = {"string": lambda x: str(x), "hex": lambda x: "%.32x" % x.int}
uuid_format = field.column_type.uuid_format
parser = parsers.get(uuid_format, lambda x: x)
parser: Callable[..., Any] = parsers.get(uuid_format, lambda x: x)
model_dict[name] = parser(model_dict[name])
return model_dict

View File

@ -1,7 +1,7 @@
from typing import Any, Dict, List, Optional, TYPE_CHECKING, Tuple, Type, Union, cast
try:
from sqlalchemy.engine.result import ResultProxy
from sqlalchemy.engine.result import ResultProxy # type: ignore
except ImportError: # pragma: no cover
from sqlalchemy.engine.result import Row as ResultProxy # type: ignore

View File

@ -1,8 +1,9 @@
from collections import OrderedDict
from typing import List, Optional, TYPE_CHECKING, Tuple, Type
from typing import List, Optional, TYPE_CHECKING, Tuple, Type, Union
import sqlalchemy
from sqlalchemy import text
from sqlalchemy import Table, text
from sqlalchemy.sql import Join
import ormar # noqa I100
from ormar.models.helpers.models import group_related_list
@ -41,7 +42,7 @@ class Query:
self.used_aliases: List[str] = []
self.select_from: List[str] = []
self.select_from: Union[Join, Table, List[str]] = []
self.columns = [sqlalchemy.Column]
self.order_columns = order_bys
self.sorted_orders: OrderedDict[OrderAction, text] = OrderedDict()