Coverage for polar/kit/extensions/sqlalchemy/types.py: 69%

39 statements  

« prev     ^ index     » next       coverage.py v7.10.6, created at 2025-12-05 16:17 +0000

1from enum import Enum, StrEnum 1ab

2from typing import TYPE_CHECKING, Any 1ab

3 

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

8 

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

15 

16 

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

21 

22 def process_bind_param(self, value: Any, dialect: Dialect) -> Any: 1ab

23 return value.value if isinstance(value, self.enum_klass) else value 

24 

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

27 

28 

29class IntEnum(EnumType): 1ab

30 impl = sa.Integer 1ab

31 cache_ok = True 1ab

32 

33 

34class StringEnum(EnumType): 1ab

35 impl = sa.Unicode 1ab

36 cache_ok = True 1ab

37 

38 

39class StrEnumType(TypeDecorator): 1ab

40 impl = sa.String 1ab

41 cache_ok = True 1ab

42 

43 def __init__(self, enum_klass: type[StrEnum], **kwargs: Any) -> None: 1ab

44 super().__init__(**kwargs) 1ab

45 self.enum_klass = enum_klass 1ab

46 

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 

51 

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