Source code for serdio.io_lifter

"""Tools for lifting arbitrary functions into functions mapping bytes to bytes.

This module deals with automatically converting between arbitrary functions and
"byte-handlers", i.e. functions mapping bytes to bytes. The :func:`.lift_io` decorator
allows a user to convert an arbitrary function to a bytes-handler. :func:`.lift_io`
returns an :class:`.IOLifter`, a Callable class that wraps an arbitrary function with
the machinery needed to convert it into a byte-handler via
:meth:`.IOLifter.as_bytes_handler`. :class:`.IOLifter` does not otherwise change the
function's call-behavior.

**Usage** ::

    def my_cool_function(x: int, y: float, b: float = 1.0) -> float:
        z = x * y
        z += b
        return z

    bytes_handler: Callable[bytes, bytes] = my_cool_function.as_bytes_handler()

    z = my_cool_function(2, 3.0)
    assert z == 7.0
import functools as ft
import inspect
from operator import xor
from typing import Callable
from typing import Optional

from serdio import serde

[docs]def lift_io( f: Callable = None, *, encoder_hook: Optional[Callable] = None, decoder_hook: Optional[Callable] = None, hook_bundle=None, as_handler: bool = False, ): """Lift a function into an :class:`.IOLifter`. The resulting :class:`.IOLifter` Callable is nearly identical to the original function, however it can also easily be converted to a bytes-handler with :meth:`~IOLifter.as_bytes_handler`. The bytes-handler expects Serdio bytes as input and returns Serdio bytes as its output. This decorator expects at most one of these sets of kwargs to be specified: - ``encoder_hook`` and ``decoder_hook`` - ``hook_bundle`` Args: f: A Callable to be IO-lifted into a bytes-handler. encoder_hook: An optional Callable that specifies how to convert custom-typed inputs or outputs into msgpack-able Python types (e.g. converting MyCustomClass into a dictionary of Python natives). See :class:`serdio.SerdeHookBundle` for details. decoder_hook: An optional Callable that specifies how to invert ``encoder_hook`` for custom-typed inputs and outputs. See :class:`serdio.SerdeHookBundle` for details. hook_bundle: An optional tuple, list, or :class:`~serdio.SerdeHookBundle` that simply packages up ``encoder_hook`` and ``decoder_hook`` Callables into a single object. as_handler: A boolean controlling the return type of the decorator. If ``False``, returns an :class:`.IOLifter` wrapping up ``f`` and the hook bundle specified by the supplied combination of ``encoder_hook``, ``decoder_hook``, and ``hook_bundle``. If True, returns the result of applying ``lambda x: x.as_bytes_handler()`` to the :class:`.IOLifter`. Returns: An :class:`.IOLifter` wrapping up ``f``, ``encoder_hook``, and ``decoder_hook`` with the machinery needed to convert arbitrary functions into bytes-handlers. If ``as_handler=True``, returns the bytes-handler version of ``f``. Raises: ValueError: if user supplies wrong combination of ``encoder_hook``, ``decoder_hook``, and ``hook_bundle``. TypeError: if ``hook_bundle`` is not coercible to :class:`~serdio.SerdeHookBundle`, or if ``encoder_hook``/``decoder_hook`` are not Callables. """ _check_lift_io_kwargs(encoder_hook, decoder_hook, hook_bundle) if encoder_hook is not None: _typecheck_hooks(encoder_hook, decoder_hook) hook_bundle = serde.SerdeHookBundle(encoder_hook, decoder_hook) elif hook_bundle is not None: hook_bundle = serde.bundle_serde_hooks(hook_bundle) _typecheck_hooks(hook_bundle.encoder_hook, hook_bundle.decoder_hook) if f is None: return ft.partial(lift_io, hook_bundle=hook_bundle, as_handler=as_handler) if as_handler: return IOLifter(f, hook_bundle=hook_bundle).as_bytes_handler() return IOLifter(f, hook_bundle=hook_bundle)
[docs]class IOLifter: """A Callable for lifting arbitrary functions into equivalent bytes-handlers. Args: f: The function we want to lift. hook_bundle: An optional :class:`~.serdio.serde.SerdeHookBundle` for dealing with user-defined types """ def __init__( self, f: Callable, hook_bundle: Optional[serde.SerdeHookBundle], ): self._func = f self._hook_bundle = hook_bundle def __call__(self, *args, **kwargs): return self._func(*args, **kwargs)
[docs] def as_cape_handler(self): """Alias of :meth:`.IOLifter.as_bytes_handler`.""" return self.as_bytes_handler()
[docs] def as_bytes_handler(self): """Lift the wrapped Callable into its functionally-equivalent bytes-handler. A bytes-handler is a Callable mapping Serdio bytes to Serdio bytes. """ if self.hook_bundle is not None: encoder_hook, decoder_hook = self.hook_bundle.unbundle() else: encoder_hook, decoder_hook = None, None def cape_handler(input_bytes): try: args, kwargs = serde.deserialize( input_bytes, decoder=decoder_hook, as_signature=True ) except ValueError: raise ValueError( "Couldn't deserialize the function's input with Serdio. " "Make sure your input is serialized according to the Serdio spec " "manually, or by setting use_serdio=True in / Cape.invoke." ) _check_inputs_match_signature(self._func, args, kwargs) output = self._func(*args, **kwargs) output_blob = serde.serialize(output, encoder=encoder_hook) return output_blob return cape_handler
@property def hook_bundle(self): return self._hook_bundle @property def encoder(self): return self._hook_bundle.encoder_hook @property def decoder(self): return self._hook_bundle.decoder_hook
def _check_lift_io_kwargs(encoder_hook, decoder_hook, hook_bundle): _check_missing_kwargs_combo(encoder_hook, decoder_hook, hook_bundle) _typecheck_hooks(encoder_hook, decoder_hook) _typecheck_bundle(hook_bundle) def _check_missing_kwargs_combo(encoder_hook, decoder_hook, hook_bundle): only_single_hook = xor(encoder_hook is not None, decoder_hook is not None) both_hooks_supplied = encoder_hook is not None and decoder_hook is not None bundle_supplied = hook_bundle is not None # not a true xor, since both sets of args are optional if not only_single_hook and not both_hooks_supplied and not bundle_supplied: return # either hooks are supplied or bundle is supplied, but not both if not only_single_hook and xor(both_hooks_supplied, bundle_supplied): return raise ValueError( "The `lift_io` decorator expects at most one of these sets of kwargs " "to be specified:\n" "\t - `encoder_hook` and `decoder_hook`\n" "\t - `hook_bundle`\n" "Found:\n" f"\t - `encoder_hook: {type(encoder_hook)}\n" f"\t - `decoder_hook: {type(decoder_hook)}\n" f"\t - `hook_bundle: {type(hook_bundle)}\n" ) def _typecheck_hooks(encoder_hook, decoder_hook): if encoder_hook is None and decoder_hook is None: return if not callable(encoder_hook): raise TypeError( f"Expected callable `encoder_hook`, found type: {type(encoder_hook)}" ) if not callable(decoder_hook): raise TypeError( f"Expected callable `decoder_hook`, found type: {type(decoder_hook)}" ) def _typecheck_bundle(hook_bundle): if hook_bundle is None: return if not isinstance(hook_bundle, (tuple, list, dict, serde.SerdeHookBundle)): raise TypeError( "`hook_bundle` keyword-argument must be one of:\n" "\t- tuple\n" "\t- list\n" "\t- dict\n" "\t- SerdeHookBundle\n" f"but found type: {type(hook_bundle)}." ) def _check_inputs_match_signature(f, args, kwargs): sig = inspect.signature(f) n_inputs = len(args) + len(kwargs) n_sig_parameters = len(sig.parameters) if n_inputs != n_sig_parameters: raise ValueError( f"The number of inputs {n_inputs} provided in or Cape invoke" f" doesn't match the number of inputs {n_sig_parameters} expected " "by the Cape handler" )