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
« 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"""
5import typing
7from zserio.exception import PythonRuntimeException
8from zserio.bitwriter import BitStreamWriter
10class ServiceData:
11 """
12 Data abstraction to be sent or to be received in all Zserio services.
13 """
15 @property
16 def zserio_object(self) -> typing.Any:
17 """
18 Gets the Zserio object which represents the request if available.
20 :returns: The Zserio object or None.
21 """
23 raise NotImplementedError()
25 @property
26 def byte_array(self) -> bytes:
27 """
28 Gets the data which represent the request.
30 :returns: The request data.
31 """
33 raise NotImplementedError()
35class ObjectServiceData(ServiceData):
36 """
37 ServiceData implementation based on object generated by Zserio.
38 """
40 def __init__(self, zserio_object: typing.Any):
41 """
42 Constructor.
44 :param zserio_object: Zserio object from which to create service data.
45 """
47 self._is_byte_array_initialized = False
48 self._byte_array = bytes()
49 self._zserio_object = zserio_object
51 @property
52 def zserio_object(self) -> typing.Any:
53 """
54 Gets the Zserio object which represents the request.
56 :returns: The Zserio object.
57 """
59 return self._zserio_object
61 @property
62 def byte_array(self) -> bytes:
63 """
64 Gets the data which represent the request.
66 :returns: The request data which are created by serialization of Zserio object.
67 """
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
75 return self._byte_array
77class RawServiceData(ServiceData):
78 """
79 Service data implementation based on raw data.
80 """
82 def __init__(self, raw_data: bytes):
83 """
84 Constructor.
86 :param raw_data: Raw data to use.
87 """
89 self._raw_data = raw_data
91 @property
92 def zserio_object(self) -> None:
93 return None
95 @property
96 def byte_array(self) -> bytes:
97 return self._raw_data
99class ServiceInterface:
100 """
101 Generic interface for all Zserio services on server side.
102 """
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.
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()
116 @property
117 def service_full_name(self) -> str:
118 """
119 Gets service full name.
121 :returns: Service full name.
122 """
123 raise NotImplementedError()
125 @property
126 def method_names(self) -> typing.List:
127 """
128 Gets list of service method names.
130 :returns: List of service method names.
131 """
132 raise NotImplementedError()
134class ServiceClientInterface:
135 """
136 Generic interface for all Zserio services on client side.
137 """
139 def call_method(self, method_name: str, request: ServiceData, context: typing.Any = None) -> bytes:
140 """
141 Calls method with the given name synchronously.
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()
151class ServiceException(PythonRuntimeException):
152 """
153 Exception thrown in case of an error in Zserio Service
154 """