Coverage for dynamodx / types.py: 59%

1from datetime import date, datetime 

2from ipaddress import IPv4Address 

3from typing import ( 

4 Any, 

5 Mapping, 

6) 

7from uuid import UUID 

8 

9from boto3.dynamodb.types import TypeDeserializer, TypeSerializer 

10 

11serializer = TypeSerializer() 

12deserializer = TypeDeserializer() 

13 

14 

15def _serialize_to_basic_types(data: Any) -> str | dict | set | list: 

16 match data: 

17 case datetime(): 

18 return data.isoformat() 

19 

20 case date(): 

21 return data.isoformat() 

22 

23 case UUID(): 

24 return str(data) 

25 

26 case IPv4Address(): 

27 return str(data) 

28 

29 case tuple() | list(): 

30 if not data: 

31 return [] 

32 

33 serialized = [_serialize_to_basic_types(v) for v in data] 

34 

35 if any(isinstance(v, (dict, list)) for v in serialized): 

36 return serialized 

37 

38 try: 

39 return set(serialized) 

40 except TypeError: 

41 return serialized 

42 

43 case set(): 

44 if not data: 

45 return [] 

46 

47 return set(_serialize_to_basic_types(v) for v in data) 

48 

49 case dict(): 

50 return {k: _serialize_to_basic_types(v) for k, v in data.items()} 

51 

52 case _: 

53 return data 

54 

55 

56def serialize(data: Mapping[str, Any], exclude_none: bool = False) -> dict: 

57 return { 

58 k: serializer.serialize(_serialize_to_basic_types(v)) 

59 for k, v in data.items() 

60 if not exclude_none or v is not None 

61 } 

62 

63 

64def deserialize(data: Mapping[str, Any]) -> dict: 

65 return {k: deserializer.deserialize(v) for k, v in data.items()} 

66 

67 

68def to_dict(data: Any | None) -> dict[str, Any] | None: 

69 # Convert from Pydantic v2 model 

70 if callable(getattr(data, 'model_dump', None)): 

71 return data.model_dump() # type: ignore 

72 

73 # Convert from dataclasses 

74 if hasattr(data, '__dataclass_fields__'): 

75 import dataclasses 

76 

77 return dataclasses.asdict(data) # type: ignore 

78 

79 return data