Skip to content

Commit

Permalink
refactor: TypeAffinityRepr is not a subclass of str anymore, split ty…
Browse files Browse the repository at this point in the history
…pe_affinity mapping implement out of TypeAffinityRepr (#67)

This PR introduces refactor over TypeAffinityRepr as follow:
1. now TypeAffinityRepr is not a subclass of str, it is a simple standalone class with `origin` and `type_affinity` attrs exposed.
2. the type_affinity mapping implementation now is split from TypeAffinityRepr and becomes `map_type` helper function.

The behavior of TypeAffinityRepr is not changed, calling str over an TypeAffinityRepr object will still give you the affinity string.
  • Loading branch information
pga2rn authored Dec 25, 2024
1 parent 85c15cc commit c8d45ca
Show file tree
Hide file tree
Showing 2 changed files with 79 additions and 50 deletions.
125 changes: 76 additions & 49 deletions src/simple_sqlite3_orm/_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -50,64 +50,91 @@ def __new__(
"""For type check only, typing the _GenericAlias as GenericAlias."""


class TypeAffinityRepr(str):
def map_type(
_in: type[Any] | SQLiteTypeAffinityLiteral | Any,
) -> SQLiteTypeAffinity | str:
"""Mapping python types to corresponding sqlite storage classes.
Currently this function suports the following input:
1. sqlite3 native types(and wrapped in Optional).
2. Literal types.
3. Enum types with str or int as data type.
4. user defined affinity string, will be used as it.
"""
if _in is None or _in is type(None):
return SQLiteTypeAffinity.NULL

if isinstance(_in, str): # user-define type affinity, use as it
return _in

_origin = get_origin(_in)
if _origin is Literal:
return _map_from_literal(_in)
if (
_origin is Union
and len(_args := get_args(_in)) == 2
and _args[-1] is type(None)
):
# Optional[X] is actually Union[X, type(None)]
# after extract the actual types from Optional,
# do mapping from the beginning.
return map_type(_args[0])
if _origin is not None:
raise TypeError(f"not one of Literal or Optional: {_in}")

if not isinstance(_in, type):
raise TypeError(f"expecting type or str object, get {type(_in)=}")
return _map_from_type(_in)


def _map_from_literal(_in: Any) -> SQLiteTypeAffinity:
"""Support for literal of supported datatypes."""
_first_literal, *_literals = get_args(_in)
literal_type = type(_first_literal)

if any(not isinstance(_literal, literal_type) for _literal in _literals):
raise TypeError(f"mix types in literal is not allowed: {_in}")
return _map_from_type(literal_type)


def _map_from_type(_in: type[Any]) -> SQLiteTypeAffinity:
if issubclass(_in, int): # NOTE: also include IntEnum
return SQLiteTypeAffinity.INTEGER
elif issubclass(_in, str): # NOTE: also include StrEnum
return SQLiteTypeAffinity.TEXT
elif issubclass(_in, bytes):
return SQLiteTypeAffinity.BLOB
elif issubclass(_in, float):
return SQLiteTypeAffinity.REAL
raise TypeError(f"cannot map {_in} to any sqlite3 type affinity")


class TypeAffinityRepr:
"""Map python types to sqlite3 data types with type affinity.
Currently supports:
1. python sqlite3 lib supported native python types.
2. StrEnum and IntEnum, will map to TEXT and INT accordingly.
3. Optional types, will map against the args inside the Optional.
4. Literal types, will map against the values type inside the Literal.
Attrs:
type_affinity (SQLiteTypeAffinity | str)
origin (type[Any] | SQLiteTypeAffinityLiteral | Any)
"""

def __new__(cls, _in: type[Any] | SQLiteTypeAffinityLiteral | Any) -> Self:
"""Mapping python types to corresponding sqlite storage classes."""
if _in is None or _in is type(None):
return str.__new__(cls, SQLiteTypeAffinity.NULL.value)

if isinstance(_in, str): # user-define type affinity, use as it
return str.__new__(cls, _in)

_origin = get_origin(_in)
if _origin is Literal:
return cls._map_from_literal(_in)
if (
_origin is Union
and len(_args := get_args(_in)) == 2
and _args[-1] is type(None)
):
# Optional[X] is actually Union[X, type(None)]
# after extract the actual types from Optional,
# do mapping from the beginning.
return cls.__new__(cls, _args[0])
if _origin is not None:
raise TypeError(f"not one of Literal or Optional: {_in}")

if not isinstance(_in, type):
raise TypeError(f"expecting type or str object, get {type(_in)=}")
return cls._map_from_type(_in)

@classmethod
def _map_from_literal(cls, _in: Any) -> Self:
"""Support for literal of supported datatypes."""
_first_literal, *_literals = get_args(_in)
literal_type = type(_first_literal)

if any(not isinstance(_literal, literal_type) for _literal in _literals):
raise TypeError(f"mix types in literal is not allowed: {_in}")
return cls._map_from_type(literal_type)

@classmethod
def _map_from_type(cls, _in: type[Any]) -> Self:
if issubclass(_in, int): # NOTE: also include IntEnum
return str.__new__(cls, SQLiteTypeAffinity.INTEGER.value)
elif issubclass(_in, str): # NOTE: also include StrEnum
return str.__new__(cls, SQLiteTypeAffinity.TEXT.value)
elif issubclass(_in, bytes):
return str.__new__(cls, SQLiteTypeAffinity.BLOB.value)
elif issubclass(_in, float):
return str.__new__(cls, SQLiteTypeAffinity.REAL.value)
raise TypeError(f"cannot map {_in} to any sqlite3 type affinity")
def __init__(self, _in: type[Any] | SQLiteTypeAffinityLiteral | Any) -> None:
self.type_affinity = map_type(_in)
self.origin = _in

def __str__(self) -> str:
if isinstance(self.type_affinity, SQLiteTypeAffinity):
return self.type_affinity.value
return self.type_affinity

def __repr__(self) -> str: # pragma: no cover
return f"<{self.__qualname__}: {self}>"


class ConstrainRepr(str):
Expand Down
4 changes: 3 additions & 1 deletion tests/test__utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,9 @@
),
)
def test_typeafinityrepr(_in, expected):
assert TypeAffinityRepr(_in) == expected
_parsed = TypeAffinityRepr(_in)
assert _parsed.type_affinity == expected
assert str(_parsed) == expected.value


@pytest.mark.parametrize(
Expand Down

1 comment on commit c8d45ca

@github-actions
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Coverage

Coverage Report
FileStmtsMissCoverMissing
src/simple_sqlite3_orm
   __init__.py12375%28–30
   _sqlite_spec.py360100% 
   _table_spec.py1883481%49, 58, 69, 73–74, 86, 98, 176, 181–183, 223, 304, 306, 308–309, 357, 360, 364–366, 368–372, 374, 380, 423–424, 534–535, 549–550
   _types.py24195%34
   _utils.py731184%40, 42–43, 46–47, 69, 84, 87, 97, 110, 134
   utils.py1243670%127, 134–135, 170–173, 219, 232, 252–256, 282, 286, 333, 338–345, 357–359, 361–362, 365, 367–369, 378–379
src/simple_sqlite3_orm/_orm
   __init__.py40100% 
   _async.py91990%28, 70, 72–73, 86–87, 89, 124, 126
   _base.py1221389%98, 149, 171–173, 186–188, 287, 327, 409, 449, 472
   _multi_thread.py1031090%28, 30–31, 66, 68–69, 82–83, 85, 118
   _utils.py18288%18, 40
TOTAL79511985% 

Tests Skipped Failures Errors Time
78 0 💤 0 ❌ 0 🔥 1m 54s ⏱️

Please sign in to comment.