Coverage for src/lexigram/web/serialization/negotiator.py: 16%

1"""Content negotiation based on Accept header.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING, Any, cast 

6 

7from lexigram.web.serialization.serializers import ( 

8 AbstractMediaSerializer, 

9 HTMLSerializer, 

10 JSONSerializer, 

11 PlainTextSerializer, 

12 XMLSerializer, 

13) 

14 

15if TYPE_CHECKING: 

16 from starlette.requests import Request as StarletteRequest 

17 from starlette.responses import Response 

18 

19 

20class ContentNegotiator: 

21 """Handles content negotiation based on Accept header.""" 

22 

23 # Default priority order for content negotiation 

24 DEFAULT_PRIORITY = [ 

25 "application/json", 

26 "text/html", 

27 "text/plain", 

28 "application/xml", 

29 "*/*", 

30 ] 

31 

32 def __init__(self, serializers: list[AbstractMediaSerializer] | None = None): 

33 """Initialize with optional custom serializers.""" 

34 self._serializers: dict[str, AbstractMediaSerializer] = {} 

35 

36 # Register default serializers if none provided 

37 if serializers: 

38 for serializer in serializers: 

39 for media_type in serializer.supported_types(): 

40 self._serializers[media_type] = serializer 

41 else: 

42 # Register built-in serializers 

43 json_ser = JSONSerializer() 

44 html_ser = HTMLSerializer() 

45 text_ser = PlainTextSerializer() 

46 xml_ser = XMLSerializer() 

47 

48 for mt in json_ser.supported_types(): 

49 self._serializers[mt] = json_ser 

50 for mt in html_ser.supported_types(): 

51 self._serializers[mt] = html_ser 

52 for mt in text_ser.supported_types(): 

53 self._serializers[mt] = text_ser 

54 for mt in xml_ser.supported_types(): 

55 self._serializers[mt] = xml_ser 

56 

57 def add_serializer(self, serializer: AbstractMediaSerializer) -> None: 

58 """Add a custom serializer.""" 

59 for media_type in serializer.supported_types(): 

60 self._serializers[media_type] = serializer 

61 

62 def negotiate(self, request: StarletteRequest) -> AbstractMediaSerializer: 

63 """Determine the best serializer based on Accept header.""" 

64 accept_header = request.headers.get("accept", "*/*") 

65 

66 # Parse Accept header into (media_type, q-value) pairs 

67 options = [] 

68 for item in accept_header.split(","): 

69 item = item.strip() 

70 if not item: 

71 continue 

72 

73 # Parse quality factor 

74 parts = item.split(";") 

75 media_type = parts[0].strip() 

76 q_value = 1.0 

77 

78 for part in parts[1:]: 

79 if part.strip().startswith("q="): 

80 try: 

81 q_value = float(part.strip()[2:]) 

82 except ValueError: 

83 q_value = 1.0 

84 

85 options.append((media_type, q_value)) 

86 

87 # Sort by q-value descending 

88 options.sort(key=lambda x: x[1], reverse=True) 

89 

90 # Find first matching serializer 

91 for media_type, _ in options: 

92 # Direct match 

93 if media_type in self._serializers: 

94 return self._serializers[media_type] 

95 

96 # Wildcard match 

97 if media_type == "*/*": 

98 return self._serializers.get("application/json", JSONSerializer()) 

99 

100 # Type wildcard match (e.g., "text/*") 

101 if "/*" in media_type: 

102 prefix = media_type.split("/*")[0] 

103 for key in self._serializers: 

104 if key.startswith(prefix + "/"): 

105 return self._serializers[key] 

106 

107 # Default to JSON 

108 return self._serializers.get("application/json", JSONSerializer()) 

109 

110 async def serialize( 

111 self, 

112 data: Any, 

113 request: StarletteRequest, 

114 ) -> Response: 

115 """Serialize data based on Accept header.""" 

116 serializer = self.negotiate(request) 

117 return await serializer.serialize(data, request) 

118 

119 

120# Global default negotiator instance 

121_default_negotiator: ContentNegotiator | None = None 

122 

123 

124def get_negotiator(context: Any | None = None) -> ContentNegotiator: 

125 """Get the default content negotiator.""" 

126 from lexigram.di.resolution.context import get_resolver 

127 

128 resolver = get_resolver(context) 

129 if resolver: 

130 return cast( 

131 "ContentNegotiator", cast("Any", resolver).resolve_sync(ContentNegotiator) 

132 ) 

133 

134 global _default_negotiator 

135 if _default_negotiator is None: 

136 _default_negotiator = ContentNegotiator() 

137 return _default_negotiator