Coverage for /home/jenkins/workspace/NDS/Zserio/NDS_ZSERIO-linux-build/compiler/extensions/python/runtime/src/zserio/service.py: 100%

48 statements  

« prev     ^ index     » next       coverage.py v6.5.0, created at 2023-12-13 15:12 +0000

1""" 

2The module provides classes for Zserio services. 

3""" 

4 

5import typing 

6 

7from zserio.exception import PythonRuntimeException 

8from zserio.bitwriter import BitStreamWriter 

9 

10class ServiceData: 

11 """ 

12 Data abstraction to be sent or to be received in all Zserio services. 

13 """ 

14 

15 @property 

16 def zserio_object(self) -> typing.Any: 

17 """ 

18 Gets the Zserio object which represents the request if available. 

19 

20 :returns: The Zserio object or None. 

21 """ 

22 

23 raise NotImplementedError() 

24 

25 @property 

26 def byte_array(self) -> bytes: 

27 """ 

28 Gets the data which represent the request. 

29 

30 :returns: The request data. 

31 """ 

32 

33 raise NotImplementedError() 

34 

35class ObjectServiceData(ServiceData): 

36 """ 

37 ServiceData implementation based on object generated by Zserio. 

38 """ 

39 

40 def __init__(self, zserio_object: typing.Any): 

41 """ 

42 Constructor. 

43 

44 :param zserio_object: Zserio object from which to create service data. 

45 """ 

46 

47 self._is_byte_array_initialized = False 

48 self._byte_array = bytes() 

49 self._zserio_object = zserio_object 

50 

51 @property 

52 def zserio_object(self) -> typing.Any: 

53 """ 

54 Gets the Zserio object which represents the request. 

55 

56 :returns: The Zserio object. 

57 """ 

58 

59 return self._zserio_object 

60 

61 @property 

62 def byte_array(self) -> bytes: 

63 """ 

64 Gets the data which represent the request. 

65 

66 :returns: The request data which are created by serialization of Zserio object. 

67 """ 

68 

69 if not self._is_byte_array_initialized: 

70 writer = BitStreamWriter() 

71 self._zserio_object.write(writer) 

72 self._byte_array = writer.byte_array 

73 self._is_byte_array_initialized = True 

74 

75 return self._byte_array 

76 

77class RawServiceData(ServiceData): 

78 """ 

79 Service data implementation based on raw data. 

80 """ 

81 

82 def __init__(self, raw_data: bytes): 

83 """ 

84 Constructor. 

85 

86 :param raw_data: Raw data to use. 

87 """ 

88 

89 self._raw_data = raw_data 

90 

91 @property 

92 def zserio_object(self) -> None: 

93 return None 

94 

95 @property 

96 def byte_array(self) -> bytes: 

97 return self._raw_data 

98 

99class ServiceInterface: 

100 """ 

101 Generic interface for all Zserio services on server side. 

102 """ 

103 

104 def call_method(self, method_name: str, request_data: bytes, context: typing.Any = None) -> ServiceData: 

105 """ 

106 Calls method with the given name synchronously. 

107 

108 :param method_name: Name of the service method to call. 

109 :param request_data: Request data to be passed to the method. 

110 :param context: Context specific for particular service. 

111 :returns: Response service data. 

112 :raises ServiceException: If the call fails. 

113 """ 

114 raise NotImplementedError() 

115 

116 @property 

117 def service_full_name(self) -> str: 

118 """ 

119 Gets service full name. 

120 

121 :returns: Service full name. 

122 """ 

123 raise NotImplementedError() 

124 

125 @property 

126 def method_names(self) -> typing.List: 

127 """ 

128 Gets list of service method names. 

129 

130 :returns: List of service method names. 

131 """ 

132 raise NotImplementedError() 

133 

134class ServiceClientInterface: 

135 """ 

136 Generic interface for all Zserio services on client side. 

137 """ 

138 

139 def call_method(self, method_name: str, request: ServiceData, context: typing.Any = None) -> bytes: 

140 """ 

141 Calls method with the given name synchronously. 

142 

143 :param method_name: Name of the service method to call. 

144 :param request: Request service data to be passed to the method. 

145 :param context: Context specific for particular service. 

146 :returns: Response data. 

147 :raises ServiceException: If the call fails. 

148 """ 

149 raise NotImplementedError() 

150 

151class ServiceException(PythonRuntimeException): 

152 """ 

153 Exception thrown in case of an error in Zserio Service 

154 """