tools/fidl/fidlgen_python/goldens/error.py.golden - fuchsia - Git at Google

 # WARNING: This file is machine generated by fidlgen.

 from __future__ import annotations

 from dataclasses import dataclass
 import sys
 import typing
 import enum

 import fidl


 from fidl._client import EventHandlerBase, FidlClient  # type: ignore[unused-ignore,unused-import]
 from fidl._server import ServerBase  # type: ignore[unused-ignore,unused-import]
 from fidl._fidl_common import DomainError, FrameworkError, MethodInfo, FidlProtocolMarker, normalize_identifier  # type: ignore[unused-ignore,unused-import]

 from abc import abstractmethod, ABC  # type: ignore[unused-ignore,unused-import]
 import collections.abc  # type: ignore[unused-ignore,unused-import]


 _UNDEFINED: typing.Any = object()


 @dataclass
 class ExampleFooRequest:
     s: str

     __fidl_kind__ = "struct"
     __fidl_type__ = "ExampleFooRequest"
     __fidl_raw_type__ = "test.error/ExampleFooRequest"

     # TODO(https://fxbug.dev/394421154): We should probably remove this method when we
     # start making breaking changes.
     def __getitem__(self, item: str):  # type: ignore
         if not isinstance(item, str):
             raise TypeError("Subscripted item must be a string")
         return getattr(self, item)

     def encode(self) -> tuple[bytes, list[typing.Any]]:
         return fidl.encode_struct(self, type(self)._encode, 16)

     @classmethod
     def decode(cls, data: bytes, handles: list[typing.Any]) -> typing.Self:
         return fidl.decode_struct(data, handles, cls._decode, 16)

     def _encode(self, encoder: fidl.Encoder, _offset: int) -> None:
         encoder.write_string(self.s, _offset + 0)

     @classmethod
     def _decode(cls, decoder: fidl.Decoder, _offset: int) -> typing.Self:
         s = decoder.read_string(_offset + 0, nullable=False)
         return cls(
             s=s,
         )

     # TODO(https://fxbug.dev/394421154): Assigning None (incorrectly) to each type is a consequence
     # of needing to support creation of a "default object" before decoding.
     @classmethod
     def make_default(cls) -> typing.Self:
         return cls(
             **{
                 "s": None,  # type: ignore[arg-type,unused-ignore]
             }
         )


 @dataclass
 class ExampleFooResponse:
     y: int

     __fidl_kind__ = "struct"
     __fidl_type__ = "ExampleFooResponse"
     __fidl_raw_type__ = "test.error/Example_foo_Response"

     # TODO(https://fxbug.dev/394421154): We should probably remove this method when we
     # start making breaking changes.
     def __getitem__(self, item: str):  # type: ignore
         if not isinstance(item, str):
             raise TypeError("Subscripted item must be a string")
         return getattr(self, item)

     def encode(self) -> tuple[bytes, list[typing.Any]]:
         return fidl.encode_struct(self, type(self)._encode, 8)

     @classmethod
     def decode(cls, data: bytes, handles: list[typing.Any]) -> typing.Self:
         return fidl.decode_struct(data, handles, cls._decode, 8)

     def _encode(self, encoder: fidl.Encoder, _offset: int) -> None:
         encoder.write_int64(self.y, _offset + 0)

     @classmethod
     def _decode(cls, decoder: fidl.Decoder, _offset: int) -> typing.Self:
         y = decoder.read_int64(_offset + 0)
         return cls(
             y=y,
         )

     # TODO(https://fxbug.dev/394421154): Assigning None (incorrectly) to each type is a consequence
     # of needing to support creation of a "default object" before decoding.
     @classmethod
     def make_default(cls) -> typing.Self:
         return cls(
             **{
                 "y": None,  # type: ignore[arg-type,unused-ignore]
             }
         )


 class ExampleFooResult:
     _response: typing.Optional[ExampleFooResponse]
     _err: typing.Optional[int]

     __fidl_kind__ = "union"
     __fidl_type__ = "ExampleFooResult"
     __fidl_raw_type__ = "test.error/Example_foo_Result"
     _is_result = True

     def __eq__(self, other: object) -> bool:
         if not isinstance(other, type(self)):
             return False
         if self.response != other.response:
             return False
         if self.err != other.err:
             return False
         return True

     def __repr__(self) -> str:
         """Returns the union repr in the format <'foo.bar.baz/FooUnion' object({value})>

         If {value} is not set, will write None."""
         variant = ""
         if self.response:
             variant = f"response={self.response!r}"
         if self.err:
             variant = f"err={self.err!r}"
         return f"<'{self.__fidl_type__}' object({variant})>"

     def __init__(
         self,
         response: typing.Optional[ExampleFooResponse] = _UNDEFINED,
         err: typing.Optional[int] = _UNDEFINED,
         _empty: typing.Optional[tuple[()]] = None,
     ):
         object.__init__(self)
         if _empty is not None:
             return

         number_of_variants = 0
         variants = []

         if response is not _UNDEFINED:
             if response is not None:
                 self._response = response
                 variants.append("response")
                 number_of_variants += 1
         if err is not _UNDEFINED:
             if err is not None:
                 self._err = err
                 variants.append("err")
                 number_of_variants += 1
         if number_of_variants > 1:
             raise TypeError(
                 f"Exactly one variant must be specified for {self.__fidl_raw_type__}: {variants}"
             )

     @property
     def response(self) -> ExampleFooResponse | None:
         return getattr(self, "_response", None)

     @property
     def err(self) -> int | None:
         return getattr(self, "_err", None)

     # TODO(https://fxbug.dev/394421154): We should probably remove this method when we
     # start making breaking changes.
     def __getitem__(self, item: str):  # type: ignore
         if not isinstance(item, str):
             raise TypeError("Subscripted item must be a string")
         return getattr(self, item)

     def encode(self) -> tuple[bytes, list[typing.Any]]:
         return fidl.encode_struct(self, type(self)._encode, 16)

     @classmethod
     def decode(cls, data: bytes, handles: list[typing.Any]) -> typing.Self:
         return fidl.decode_struct(data, handles, cls._decode, 16)

     def _encode(self, encoder: fidl.Encoder, _offset: int) -> None:
         if hasattr(self, "_response"):
             encoder.write_uint64(1, _offset)
             encoder.write_envelope(
                 getattr(self, "_response"),
                 _offset + 8,
                 ExampleFooResponse._encode,
                 8,
                 False,
             )
             return
         if hasattr(self, "_err"):
             encoder.write_uint64(2, _offset)
             encoder.write_envelope(
                 getattr(self, "_err"),
                 _offset + 8,
                 lambda val, enc, off: enc.write_uint32(val, off),
                 4,
                 True,
             )
             return
         encoder.write_inline(_offset, b"\x00" * 16)

     @classmethod
     def _decode(cls, decoder: fidl.Decoder, _offset: int) -> typing.Self:
         ordinal = decoder.read_uint64(_offset)
         if ordinal == 0:
             return cls(_empty=())
         if ordinal == 1:
             val = decoder.read_envelope(
                 _offset + 8,
                 ExampleFooResponse._decode,
                 8,
             )
             return cls(response=val)
         if ordinal == 2:
             val = decoder.read_envelope(
                 _offset + 8,
                 lambda dec, off: dec.read_uint32(off),
                 4,
             )
             return cls(err=val)

         # Skip unknown envelope
         decoder.skip_envelope(_offset + 8)
         raise ValueError(f"Unknown union ordinal {ordinal}")

     def unwrap(self) -> ExampleFooResponse:
         """Returns the response if result does not contain an error. Otherwise, raises an exception."""
         try:
             if (
                 hasattr(self, "_framework_err")
                 and self._framework_err is not None
             ):
                 raise AssertionError(
                     f"{self.__fidl_raw_type__} framework error {self._framework_err}"
                 )
         except AttributeError:
             pass
         try:
             if hasattr(self, "_err") and self._err is not None:
                 raise AssertionError(
                     f"{self.__fidl_raw_type__} error {self._err}"
                 )
         except AttributeError:
             pass

         assert (
             self._response is not None
         ), f"Failed to unwrap {self.__fidl_raw_type__}. Result does not contain an error or response."
         return self._response

     @classmethod
     def make_default(cls) -> typing.Self:
         return cls(_empty=())


 class ExampleServer(ServerBase, ABC):
     __fidl_kind__ = "server"
     library = "test.error"
     method_map: typing.Dict[int, MethodInfo] = {
         1107623248440401476: MethodInfo(
             name="foo",
             request_ident="test.error/ExampleFooRequest",
             requires_response=True,
             empty_response=False,
             has_result=True,
             response_identifier="test.error/Example_foo_Result",
         ),
     }

     FooResponse: typing.TypeAlias = typing.Union[
         DomainError,
         ExampleFooResponse,
     ]

     @abstractmethod
     def foo(
         self,
         request: ExampleFooRequest,
     ) -> (
         collections.abc.Coroutine[typing.Any, typing.Any, FooResponse]
         | FooResponse
     ):
         ...


 class ExampleClient(FidlClient):
     __fidl_kind__ = "client"
     library = "test.error"

     def foo(
         self,
         *,
         s: str,
     ) -> collections.abc.Coroutine[typing.Any, typing.Any, ExampleFooResult]:
         msg = ExampleFooRequest(
             s,
         )
         ret = self._send_two_way_fidl_request(
             1107623248440401476,
             "test.error",
             msg,
             ExampleFooResult,
         )
         return ret


 class ExampleEventHandler(EventHandlerBase, ABC):
     __fidl_kind__ = "event_handler"
     library = "test.error"
     method_map: typing.Dict[int, MethodInfo] = {}


 ExampleMarker = FidlProtocolMarker("(nondiscoverable) test.error/Example")


 fidl.register_method(
     1107623248440401476,
     ExampleFooRequest,
     ExampleFooResult,
 )
	# WARNING: This file is machine generated by fidlgen.

	from __future__ import annotations

	from dataclasses import dataclass
	import sys
	import typing
	import enum

	import fidl


	from fidl._client import EventHandlerBase, FidlClient # type: ignore[unused-ignore,unused-import]
	from fidl._server import ServerBase # type: ignore[unused-ignore,unused-import]
	from fidl._fidl_common import DomainError, FrameworkError, MethodInfo, FidlProtocolMarker, normalize_identifier # type: ignore[unused-ignore,unused-import]

	from abc import abstractmethod, ABC # type: ignore[unused-ignore,unused-import]
	import collections.abc # type: ignore[unused-ignore,unused-import]


	_UNDEFINED: typing.Any = object()


	@dataclass
	class ExampleFooRequest:
	s: str

	__fidl_kind__ = "struct"
	__fidl_type__ = "ExampleFooRequest"
	__fidl_raw_type__ = "test.error/ExampleFooRequest"

	# TODO(https://fxbug.dev/394421154): We should probably remove this method when we
	# start making breaking changes.
	def __getitem__(self, item: str): # type: ignore
	if not isinstance(item, str):
	raise TypeError("Subscripted item must be a string")
	return getattr(self, item)

	def encode(self) -> tuple[bytes, list[typing.Any]]:
	return fidl.encode_struct(self, type(self)._encode, 16)

	@classmethod
	def decode(cls, data: bytes, handles: list[typing.Any]) -> typing.Self:
	return fidl.decode_struct(data, handles, cls._decode, 16)

	def _encode(self, encoder: fidl.Encoder, _offset: int) -> None:
	encoder.write_string(self.s, _offset + 0)

	@classmethod
	def _decode(cls, decoder: fidl.Decoder, _offset: int) -> typing.Self:
	s = decoder.read_string(_offset + 0, nullable=False)
	return cls(
	s=s,
	)

	# TODO(https://fxbug.dev/394421154): Assigning None (incorrectly) to each type is a consequence
	# of needing to support creation of a "default object" before decoding.
	@classmethod
	def make_default(cls) -> typing.Self:
	return cls(
	**{
	"s": None, # type: ignore[arg-type,unused-ignore]
	}
	)


	@dataclass
	class ExampleFooResponse:
	y: int

	__fidl_kind__ = "struct"
	__fidl_type__ = "ExampleFooResponse"
	__fidl_raw_type__ = "test.error/Example_foo_Response"

	# TODO(https://fxbug.dev/394421154): We should probably remove this method when we
	# start making breaking changes.
	def __getitem__(self, item: str): # type: ignore
	if not isinstance(item, str):
	raise TypeError("Subscripted item must be a string")
	return getattr(self, item)

	def encode(self) -> tuple[bytes, list[typing.Any]]:
	return fidl.encode_struct(self, type(self)._encode, 8)

	@classmethod
	def decode(cls, data: bytes, handles: list[typing.Any]) -> typing.Self:
	return fidl.decode_struct(data, handles, cls._decode, 8)

	def _encode(self, encoder: fidl.Encoder, _offset: int) -> None:
	encoder.write_int64(self.y, _offset + 0)

	@classmethod
	def _decode(cls, decoder: fidl.Decoder, _offset: int) -> typing.Self:
	y = decoder.read_int64(_offset + 0)
	return cls(
	y=y,
	)

	# TODO(https://fxbug.dev/394421154): Assigning None (incorrectly) to each type is a consequence
	# of needing to support creation of a "default object" before decoding.
	@classmethod
	def make_default(cls) -> typing.Self:
	return cls(
	**{
	"y": None, # type: ignore[arg-type,unused-ignore]
	}
	)


	class ExampleFooResult:
	_response: typing.Optional[ExampleFooResponse]
	_err: typing.Optional[int]

	__fidl_kind__ = "union"
	__fidl_type__ = "ExampleFooResult"
	__fidl_raw_type__ = "test.error/Example_foo_Result"
	_is_result = True

	def __eq__(self, other: object) -> bool:
	if not isinstance(other, type(self)):
	return False
	if self.response != other.response:
	return False
	if self.err != other.err:
	return False
	return True

	def __repr__(self) -> str:
	"""Returns the union repr in the format <'foo.bar.baz/FooUnion' object({value})>

	If {value} is not set, will write None."""
	variant = ""
	if self.response:
	variant = f"response={self.response!r}"
	if self.err:
	variant = f"err={self.err!r}"
	return f"<'{self.__fidl_type__}' object({variant})>"

	def __init__(
	self,
	response: typing.Optional[ExampleFooResponse] = _UNDEFINED,
	err: typing.Optional[int] = _UNDEFINED,
	_empty: typing.Optional[tuple[()]] = None,
	):
	object.__init__(self)
	if _empty is not None:
	return

	number_of_variants = 0
	variants = []

	if response is not _UNDEFINED:
	if response is not None:
	self._response = response
	variants.append("response")
	number_of_variants += 1
	if err is not _UNDEFINED:
	if err is not None:
	self._err = err
	variants.append("err")
	number_of_variants += 1
	if number_of_variants > 1:
	raise TypeError(
	f"Exactly one variant must be specified for {self.__fidl_raw_type__}: {variants}"
	)

	@property
	def response(self) -> ExampleFooResponse \| None:
	return getattr(self, "_response", None)

	@property
	def err(self) -> int \| None:
	return getattr(self, "_err", None)

	# TODO(https://fxbug.dev/394421154): We should probably remove this method when we
	# start making breaking changes.
	def __getitem__(self, item: str): # type: ignore
	if not isinstance(item, str):
	raise TypeError("Subscripted item must be a string")
	return getattr(self, item)

	def encode(self) -> tuple[bytes, list[typing.Any]]:
	return fidl.encode_struct(self, type(self)._encode, 16)

	@classmethod
	def decode(cls, data: bytes, handles: list[typing.Any]) -> typing.Self:
	return fidl.decode_struct(data, handles, cls._decode, 16)

	def _encode(self, encoder: fidl.Encoder, _offset: int) -> None:
	if hasattr(self, "_response"):
	encoder.write_uint64(1, _offset)
	encoder.write_envelope(
	getattr(self, "_response"),
	_offset + 8,
	ExampleFooResponse._encode,
	8,
	False,
	)
	return
	if hasattr(self, "_err"):
	encoder.write_uint64(2, _offset)
	encoder.write_envelope(
	getattr(self, "_err"),
	_offset + 8,
	lambda val, enc, off: enc.write_uint32(val, off),
	4,
	True,
	)
	return
	encoder.write_inline(_offset, b"\x00" * 16)

	@classmethod
	def _decode(cls, decoder: fidl.Decoder, _offset: int) -> typing.Self:
	ordinal = decoder.read_uint64(_offset)
	if ordinal == 0:
	return cls(_empty=())
	if ordinal == 1:
	val = decoder.read_envelope(
	_offset + 8,
	ExampleFooResponse._decode,
	8,
	)
	return cls(response=val)
	if ordinal == 2:
	val = decoder.read_envelope(
	_offset + 8,
	lambda dec, off: dec.read_uint32(off),
	4,
	)
	return cls(err=val)

	# Skip unknown envelope
	decoder.skip_envelope(_offset + 8)
	raise ValueError(f"Unknown union ordinal {ordinal}")

	def unwrap(self) -> ExampleFooResponse:
	"""Returns the response if result does not contain an error. Otherwise, raises an exception."""
	try:
	if (
	hasattr(self, "_framework_err")
	and self._framework_err is not None
	):
	raise AssertionError(
	f"{self.__fidl_raw_type__} framework error {self._framework_err}"
	)
	except AttributeError:
	pass
	try:
	if hasattr(self, "_err") and self._err is not None:
	raise AssertionError(
	f"{self.__fidl_raw_type__} error {self._err}"
	)
	except AttributeError:
	pass

	assert (
	self._response is not None
	), f"Failed to unwrap {self.__fidl_raw_type__}. Result does not contain an error or response."
	return self._response

	@classmethod
	def make_default(cls) -> typing.Self:
	return cls(_empty=())


	class ExampleServer(ServerBase, ABC):
	__fidl_kind__ = "server"
	library = "test.error"
	method_map: typing.Dict[int, MethodInfo] = {
	1107623248440401476: MethodInfo(
	name="foo",
	request_ident="test.error/ExampleFooRequest",
	requires_response=True,
	empty_response=False,
	has_result=True,
	response_identifier="test.error/Example_foo_Result",
	),
	}

	FooResponse: typing.TypeAlias = typing.Union[
	DomainError,
	ExampleFooResponse,
	]

	@abstractmethod
	def foo(
	self,
	request: ExampleFooRequest,
	) -> (
	collections.abc.Coroutine[typing.Any, typing.Any, FooResponse]
	\| FooResponse
	):
	...


	class ExampleClient(FidlClient):
	__fidl_kind__ = "client"
	library = "test.error"

	def foo(
	self,
	*,
	s: str,
	) -> collections.abc.Coroutine[typing.Any, typing.Any, ExampleFooResult]:
	msg = ExampleFooRequest(
	s,
	)
	ret = self._send_two_way_fidl_request(
	1107623248440401476,
	"test.error",
	msg,
	ExampleFooResult,
	)
	return ret


	class ExampleEventHandler(EventHandlerBase, ABC):
	__fidl_kind__ = "event_handler"
	library = "test.error"
	method_map: typing.Dict[int, MethodInfo] = {}


	ExampleMarker = FidlProtocolMarker("(nondiscoverable) test.error/Example")


	fidl.register_method(
	1107623248440401476,
	ExampleFooRequest,
	ExampleFooResult,
	)