Coverage for polar/kit/extensions/sqlalchemy/types.py: 69%
39 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 17:15 +0000
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 17:15 +0000
1from enum import Enum, StrEnum 1ab
2from typing import TYPE_CHECKING, Any 1ab
4import sqlalchemy as sa 1ab
5from sqlalchemy.dialects.postgresql import UUID 1ab
6from sqlalchemy.engine.interfaces import Dialect 1ab
7from sqlalchemy.types import TypeDecorator as _TypeDecorator 1ab
9if TYPE_CHECKING: # pragma: no cover 9 ↛ 10line 9 didn't jump to line 10 because the condition on line 9 was never true1ab
10 GUIDTypeDecorator = _TypeDecorator[UUID] # type: ignore
11 TypeDecorator = _TypeDecorator[Any]
12else:
13 GUIDTypeDecorator = _TypeDecorator 1ab
14 TypeDecorator = _TypeDecorator 1ab
17class EnumType(TypeDecorator): 1ab
18 def __init__(self, enum_klass: type[Enum], **kwargs: Any) -> None: 1ab
19 super().__init__(**kwargs) 1acb
20 self.enum_klass = enum_klass 1acb
22 def process_bind_param(self, value: Any, dialect: Dialect) -> Any: 1ab
23 return value.value if isinstance(value, self.enum_klass) else value
25 def process_result_value(self, value: Any, dialect: Dialect) -> Any: 1ab
26 return value if value is None else self.enum_klass(value) 1d
29class IntEnum(EnumType): 1ab
30 impl = sa.Integer 1ab
31 cache_ok = True 1ab
34class StringEnum(EnumType): 1ab
35 impl = sa.Unicode 1ab
36 cache_ok = True 1ab
39class StrEnumType(TypeDecorator): 1ab
40 impl = sa.String 1ab
41 cache_ok = True 1ab
43 def __init__(self, enum_klass: type[StrEnum], **kwargs: Any) -> None: 1ab
44 super().__init__(**kwargs) 1ab
45 self.enum_klass = enum_klass 1ab
47 def process_bind_param(self, value: Any, dialect: Dialect) -> Any: 1ab
48 if isinstance(value, self.enum_klass):
49 return str(value)
50 return value
52 def process_result_value(self, value: Any, dialect: Dialect) -> Any: 1ab
53 if value is not None:
54 return self.enum_klass(value)
55 return value