Coverage for /home/runner/work/zserio/zserio/compiler/extensions/python/runtime/src/zserio/service.py: 100%
48 statements
« prev ^ index » next coverage.py v6.5.0, created at 2024-07-18 11:41 +0000
1"""
2The module provides classes for Zserio services.
3"""
5import typing
7from zserio.exception import PythonRuntimeException
8from zserio.bitwriter import BitStreamWriter
11class ServiceData:
12 """
13 Data abstraction to be sent or to be received in all Zserio services.
14 """
16 @property
17 def zserio_object(self) -> typing.Any:
18 """
19 Gets the Zserio object which represents the request if available.
21 :returns: The Zserio object or None.
22 """
24 raise NotImplementedError()
26 @property
27 def byte_array(self) -> bytes:
28 """
29 Gets the data which represent the request.
31 :returns: The request data.
32 """
34 raise NotImplementedError()
37class ObjectServiceData(ServiceData):
38 """
39 ServiceData implementation based on object generated by Zserio.
40 """
42 def __init__(self, zserio_object: typing.Any):
43 """
44 Constructor.
46 :param zserio_object: Zserio object from which to create service data.
47 """
49 self._is_byte_array_initialized = False
50 self._byte_array = bytes()
51 self._zserio_object = zserio_object
53 @property
54 def zserio_object(self) -> typing.Any:
55 """
56 Gets the Zserio object which represents the request.
58 :returns: The Zserio object.
59 """
61 return self._zserio_object
63 @property
64 def byte_array(self) -> bytes:
65 """
66 Gets the data which represent the request.
68 :returns: The request data which are created by serialization of Zserio object.
69 """
71 if not self._is_byte_array_initialized:
72 writer = BitStreamWriter()
73 self._zserio_object.write(writer)
74 self._byte_array = writer.byte_array
75 self._is_byte_array_initialized = True
77 return self._byte_array
80class RawServiceData(ServiceData):
81 """
82 Service data implementation based on raw data.
83 """
85 def __init__(self, raw_data: bytes):
86 """
87 Constructor.
89 :param raw_data: Raw data to use.
90 """
92 self._raw_data = raw_data
94 @property
95 def zserio_object(self) -> None:
96 return None
98 @property
99 def byte_array(self) -> bytes:
100 return self._raw_data
103class ServiceInterface:
104 """
105 Generic interface for all Zserio services on server side.
106 """
108 def call_method(self, method_name: str, request_data: bytes, context: typing.Any = None) -> ServiceData:
109 """
110 Calls method with the given name synchronously.
112 :param method_name: Name of the service method to call.
113 :param request_data: Request data to be passed to the method.
114 :param context: Context specific for particular service.
115 :returns: Response service data.
116 :raises ServiceException: If the call fails.
117 """
118 raise NotImplementedError()
120 @property
121 def service_full_name(self) -> str:
122 """
123 Gets service full name.
125 :returns: Service full name.
126 """
127 raise NotImplementedError()
129 @property
130 def method_names(self) -> typing.List:
131 """
132 Gets list of service method names.
134 :returns: List of service method names.
135 """
136 raise NotImplementedError()
139class ServiceClientInterface:
140 """
141 Generic interface for all Zserio services on client side.
142 """
144 def call_method(self, method_name: str, request: ServiceData, context: typing.Any = None) -> bytes:
145 """
146 Calls method with the given name synchronously.
148 :param method_name: Name of the service method to call.
149 :param request: Request service data to be passed to the method.
150 :param context: Context specific for particular service.
151 :returns: Response data.
152 :raises ServiceException: If the call fails.
153 """
154 raise NotImplementedError()
157class ServiceException(PythonRuntimeException):
158 """
159 Exception thrown in case of an error in Zserio Service
160 """