Source code for pyobs.comm.proxy

from __future__ import annotations
import inspect
import types
from typing import TYPE_CHECKING, Any, Type, List, Dict, get_type_hints

from pyobs.interfaces import Interface
from pyobs.utils.types import cast_bound_arguments_to_simple

if TYPE_CHECKING:
    from pyobs.comm import Comm


class Proxy:
    """A proxy for remote pyobs modules."""

    __module__ = "pyobs.comm"

    def __init__(self, comm: Comm, client: str, interfaces: List[Type[Interface]]):
        """Creates a new proxy.

        Args:
            comm: Comm object to use for connection.
            client: Name of client to connect to.
            interfaces: List of interfaces supported by client.
        """

        # set client and interfaces
        self._comm: Comm = comm
        self._client = client
        self._interfaces = interfaces

        # remove interfaces that are implemented by others
        to_delete = []
        for i1 in interfaces:
            for i2 in interfaces:
                if i1 != i2 and issubclass(i1, i2):
                    # i1 implements i2, so remove i2
                    to_delete.append(i2)
        interfaces = [i for i in interfaces if i not in to_delete]

        # add interfaces as base classes
        cls = self.__class__
        self.__class__ = cls.__class__("Proxy", tuple([cls] + interfaces), {})  # type: ignore

        # create methods
        self._methods = self._create_methods()

    @property
    def name(self) -> str:
        """Name of the client."""
        return self._client

    @property
    def method_names(self) -> List[str]:
        """List of method names."""
        return list(sorted(self._methods.keys()))

    @property
    def interfaces(self) -> List[Type[Interface]]:
        """List of interfaces."""
        return self._interfaces

[docs] def signature(self, method: str) -> inspect.Signature: """Returns the signature of a given method. Args: method: Name of the method. Returns: Signature of the given method. """ return inspect.signature(self._methods[method][0])
[docs] def interface_method(self, method: str) -> Any: """Returns the method of the given name from the interface and not from the object itself. Args: method: Name of method. Returns: The interface method. """ return self._methods[method][0]
[docs] async def execute(self, method: str, *args: Any, **kwargs: Any) -> Any: """Execute a method on the remote client. Args: method: Name of method to call. *args: Parameters for method call. **kwargs: Parameters for method call. Returns: Result of method call. """ # add 'self' to args args = tuple([self] + list(args)) # get signature and type hints _, signature, _, type_hints = self._methods[method] # bind signature ba = signature.bind(*args, **kwargs) ba.apply_defaults() # cast to simple types cast_bound_arguments_to_simple( ba, type_hints, pre=self._comm.cast_to_simple_pre, post=self._comm.cast_to_simple_post ) # do request and return future return await self._comm.execute(self._client, method, type_hints, *ba.args[1:])
def _create_methods(self) -> Dict[str, Any]: """Create local methods for the remote client.""" # loop all interfaces and get methods methods = {} for interface in self._interfaces: # loop all methods: for func_name, func in inspect.getmembers(interface, predicate=inspect.isfunction): # set method my_func = types.MethodType(self._remote_function_wrapper(func_name), self) setattr(self, func_name, my_func) type_hints = get_type_hints(func) signature = inspect.signature(func) # store func methods[func_name] = (func, signature, getattr(self, func_name), type_hints) # return methods return methods def _remote_function_wrapper(self, method: str) -> Any: """Function wrapper for remote calls. Args: method: Name of method to wrap. Returns: Wrapper. """ async def inner(this: Proxy, *args: Any, **kwargs: Any) -> Any: return await this.execute(method, *args, **kwargs) return inner __all__ = ["Proxy"]