Coverage for fast_depends/library/serializer.py: 100%

1import inspect 20 ctx1abcdefghijklnopqrsmt

2import json 20 ctx1abcdefghijklnopqrsmt

3from abc import ABC, abstractmethod 20 ctx1abcdefghijklnopqrsmt

4from typing import Any, Protocol 20 ctx1abcdefghijklnopqrsmt

5 

6 

7class OptionItem: 20 ctx1abcdefghijklnopqrsmt

8 __slots__ = ( 20 ctx1abcdefghijklnopqrsmt

9 "field_name", 

10 "field_type", 

11 "default_value", 

12 "kind", 

13 "source", 

14 ) 

15 

16 def __init__( 20 ctx1abcdefghijklnopqrsmt

17 self, 

18 field_name: str, 

19 field_type: Any, 

20 source: Any = None, 

21 default_value: Any = ..., 

22 kind: inspect._ParameterKind = inspect.Parameter.POSITIONAL_OR_KEYWORD, 

23 ) -> None: 

24 self.field_name = field_name 20 ctx1abcdefghijklnopqrsmt

25 self.field_type = field_type 20 ctx1abcdefghijklnopqrsmt

26 self.default_value = default_value 20 ctx1abcdefghijklnopqrsmt

27 self.source = source 20 ctx1abcdefghijklnopqrsmt

28 self.kind = kind 20 ctx1abcdefghijklnopqrsmt

29 

30 def __repr__(self) -> str: 20 ctx1abcdefghijklnopqrsmt

31 type_name = getattr(self.field_type, "__name__", str(self.field_type)) 20 ctx1abcdefghijklnopqrsmt

32 content = f"{self.field_name}, type=`{type_name}`" 20 ctx1abcdefghijklnopqrsmt

33 if self.default_value is not Ellipsis: 20 ctx1abcdefghijklnopqrsmt

34 content = f"{content}, default=`{self.default_value}`" 

35 if self.source: 20 ctx1abcdefghijklnopqrsmt

36 content = f"{content}, source=`{self.source}`" 

37 return f"OptionItem[{content}]" 20 ctx1abcdefghijklnopqrsmt

38 

39 

40class Serializer(ABC): 20 ctx1abcdefghijklnopqrsmt

41 def __init__( 20 ctx1abcdefghijklnopqrsmt

42 self, 

43 *, 

44 name: str, 

45 options: list[OptionItem], 

46 response_type: Any, 

47 ): 

48 self.name = name 20 ctx1abcdefghijklnopqrsmt

49 self.options = {i.field_name: i for i in options} 20 ctx1abcdefghijklnopqrsmt

50 self.response_option = { 20 ctx1abcdefghijklnopqrsmt

51 "return": OptionItem(field_name="return", field_type=response_type), 

52 } 

53 

54 def get_aliases(self) -> tuple[str, ...]: 20 ctx1abcdefghijklnopqrsmt

55 return () 20 ctx1abcdefghijklnopqrsmt

56 

57 @abstractmethod 20 ctx1abcdefghijklnopqrsmt

58 def __call__(self, call_kwargs: dict[str, Any]) -> dict[str, Any]: 20 ctx1abcdefghijklnopqrsmt

59 raise NotImplementedError 

60 

61 def response(self, value: Any) -> Any: 20 ctx1abcdefghijklnopqrsmt

62 return value 20 ctx1abcdefghijklnopqrsmt

63 

64 

65class SerializerProto(Protocol): 20 ctx1abcdefghijklnopqrsmt

66 def __call__( 20 ctx1abcdefghijklnopqrsmt

67 self, 

68 *, 

69 name: str, 13 ctx1abcdefghijklm

70 options: list[OptionItem], 13 ctx1abcdefghijklm

71 response_type: Any, 13 ctx1abcdefghijklm

72 ) -> Serializer: ... 13 ctx1abcdefghijklm

73 

74 @staticmethod 20 ctx1abcdefghijklnopqrsmt

75 def encode(message: Any) -> bytes: 20 ctx1abcdefghijklnopqrsmt

76 return json.dumps(message).encode("utf-8") 20 ctx1abcdefghijklnopqrsmt