Coverage for amqtt/contrib/__init__.py: 72%

1"""Module for contributed plugins.""" 

2 

3from dataclasses import asdict, is_dataclass 

4from typing import Any, TypeVar 

5 

6from sqlalchemy import JSON, TypeDecorator 

7 

8T = TypeVar("T") 

9 

10 

11class DataClassListJSON(TypeDecorator[list[dict[str, Any]]]): 

12 impl = JSON 

13 cache_ok = True 

14 

15 def __init__(self, dataclass_type: type[T]) -> None: 

16 if not is_dataclass(dataclass_type): 16 ↛ 17line 16 didn't jump to line 17 because the condition on line 16 was never true

17 msg = f"{dataclass_type} must be a dataclass type" 

18 raise TypeError(msg) 

19 self.dataclass_type = dataclass_type 

20 super().__init__() 

21 

22 def process_bind_param( 

23 self, 

24 value: list[Any] | None, # Python -> DB 

25 dialect: Any 

26 ) -> list[dict[str, Any]] | None: 

27 if value is None: 27 ↛ 28line 27 didn't jump to line 28 because the condition on line 27 was never true

28 return None 

29 return [asdict(item) for item in value] 

30 

31 def process_result_value( 

32 self, 

33 value: list[dict[str, Any]] | None, # DB -> Python 

34 dialect: Any 

35 ) -> list[Any] | None: 

36 if value is None: 36 ↛ 37line 36 didn't jump to line 37 because the condition on line 36 was never true

37 return None 

38 return [self.dataclass_type(**item) for item in value] 

39 

40 def process_literal_param(self, value: Any, dialect: Any) -> Any: 

41 # Required by SQLAlchemy, typically used for literal SQL rendering. 

42 return value 

43 

44 @property 

45 def python_type(self) -> type: 

46 # Required by TypeEngine to indicate the expected Python type. 

47 return list