Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-relay/src/lexigram/ai/relay/mappers/gemini.py: 78%

1"""Google Gemini ``generateContent`` request and response mapper. 

2 

3Converts the Gemini wire DTOs (:class:`GeminiRequest` / 

4:class:`GeminiResponse`) into the canonical relay IR and back. Stream 

5conversion is handled by the shared stream lifecycle task and reports 

6``unsupported_feature`` until then. 

7 

8Request conversion lives in :mod:`gemini_request`; response conversion 

9in :mod:`gemini_response`; this module wires both into 

10:class:`GeminiMapper`. 

11""" 

12 

13from __future__ import annotations 

14 

15from typing import Any 

16 

17from lexigram.ai.relay.context import ConversionContext 

18from lexigram.ai.relay.errors import unsupported_feature 

19from lexigram.ai.relay.mappers.gemini_request import ( 

20 _TARGET, 

21) 

22from lexigram.ai.relay.mappers.gemini_request import ( 

23 ir_to_request as _ir_to_request, 

24) 

25from lexigram.ai.relay.mappers.gemini_request import ( 

26 request_to_ir as _request_to_ir, 

27) 

28from lexigram.ai.relay.mappers.gemini_response import ( 

29 ir_to_response as _ir_to_response, 

30) 

31from lexigram.ai.relay.mappers.gemini_response import ( 

32 response_to_ir as _response_to_ir, 

33) 

34from lexigram.contracts.ai.exceptions import RelayError 

35from lexigram.contracts.ai.relay.ir import ( 

36 RelayRequest, 

37 RelayResponse, 

38 StreamDelta, 

39 StreamState, 

40) 

41from lexigram.contracts.core.result import Err, Result 

42 

43__all__ = ["GeminiMapper"] 

44 

45 

46class GeminiMapper: 

47 """Bidirectional Google Gemini ``generateContent`` converter. 

48 

49 Attributes: 

50 format: The wire format this mapper handles. 

51 """ 

52 

53 format = _TARGET 

54 

55 def request_to_ir( 

56 self, payload: Any, *, context: ConversionContext 

57 ) -> Result[RelayRequest, RelayError]: 

58 """Convert a ``GeminiRequest`` into canonical ``RelayRequest``. 

59 

60 Args: 

61 payload: A wire request DTO. 

62 context: Per-conversion context with loss sink. 

63 

64 Returns: 

65 Ok(request) on success, Err(relay_error) on malformed payload. 

66 """ 

67 return _request_to_ir(payload, context=context) 

68 

69 def ir_to_request( 

70 self, request: RelayRequest, *, context: ConversionContext 

71 ) -> Result[Any, RelayError]: 

72 """Convert canonical ``RelayRequest`` into a ``GeminiRequest``. 

73 

74 Args: 

75 request: Canonical request IR. 

76 context: Per-conversion context with loss sink. 

77 

78 Returns: 

79 Ok(request) on success, Err(relay_error) on failure. 

80 """ 

81 return _ir_to_request(request, context=context) 

82 

83 def response_to_ir( 

84 self, payload: Any, *, context: ConversionContext 

85 ) -> Result[RelayResponse, RelayError]: 

86 """Convert a ``GeminiResponse`` into canonical ``RelayResponse``. 

87 

88 Args: 

89 payload: A wire response DTO. 

90 context: Per-conversion context with loss sink. 

91 

92 Returns: 

93 Ok(response) on success, Err(relay_error) on malformed payload. 

94 """ 

95 return _response_to_ir(payload, context=context) 

96 

97 def ir_to_response( 

98 self, response: RelayResponse, *, context: ConversionContext 

99 ) -> Result[Any, RelayError]: 

100 """Convert canonical ``RelayResponse`` into a ``GeminiResponse``. 

101 

102 Args: 

103 response: Canonical response IR. 

104 context: Per-conversion context with loss sink. 

105 

106 Returns: 

107 Ok(response) on success, Err(relay_error) on failure. 

108 """ 

109 return _ir_to_response(response, context=context) 

110 

111 def stream_to_delta( 

112 self, event: Any, *, state: StreamState 

113 ) -> Result[tuple[StreamDelta, ...], RelayError]: 

114 """Stream conversion is deferred to the shared stream lifecycle task.""" 

115 return Err( 

116 unsupported_feature("gemini stream conversion is not implemented yet") 

117 ) 

118 

119 def delta_to_stream( 

120 self, delta: StreamDelta, *, state: StreamState 

121 ) -> Result[tuple[Any, ...], RelayError]: 

122 """Stream conversion is deferred to the shared stream lifecycle task.""" 

123 return Err( 

124 unsupported_feature("gemini stream conversion is not implemented yet") 

125 )