Source code for polywrap_plugin.resolution_context_override_client

"""This module defines the ResolutionContextOverrideClient class."""
from typing import Any, List, Optional

from polywrap_core import InvokerClient, Uri, UriResolutionContext


[docs]class ResolutionContextOverrideClient(InvokerClient): """A client that overrides the resolution context of the wrapped client. Args: client (InvokerClient): The wrapped client. resolution_context (Optional[UriResolutionContext]): \ The resolution context to use. """ client: InvokerClient resolution_context: Optional[UriResolutionContext] __slots__ = ("client", "resolution_context") def __init__( self, client: InvokerClient, resolution_context: Optional[UriResolutionContext] ): """Initialize a new ResolutionContextOverrideClient instance.""" self.client = client self.resolution_context = resolution_context
[docs] def invoke( self, uri: Uri, method: str, args: Optional[Any] = None, env: Optional[Any] = None, resolution_context: Optional[UriResolutionContext] = None, encode_result: Optional[bool] = False, ) -> Any: """Invoke the Wrapper based on the provided InvokerOptions. Args: uri (Uri): Uri of the wrapper method (str): Method to be executed args (Optional[Any]) : Arguments for the method, structured as a dictionary env (Optional[Any]): Override the client's config for all invokes within this invoke. resolution_context (Optional[UriResolutionContext]): A URI resolution context encode_result (Optional[bool]): If True, the result will be encoded Returns: Any: invocation result. Raises: WrapInvocationError: If the plugin method is not defined\ or is not callable. WrapAbortError: If the plugin method raises an exception. MsgpackDecodeError: If the plugin method returns invalid msgpack. """ return self.client.invoke( uri, method, args, env, self.resolution_context, encode_result, )
[docs] def get_implementations( self, uri: Uri, apply_resolution: bool = True ) -> Optional[List[Uri]]: """Get implementations of an interface with its URI. Args: uri (Uri): URI of the interface. apply_resolution (bool): If True, apply resolution to the URI and interfaces. Returns: Optional[List[Uri]]: List of implementations or None if not found. """ return self.client.get_implementations(uri, apply_resolution)
[docs] def try_resolve_uri( self, uri: Uri, resolution_context: UriResolutionContext | None = None ) -> Any: """Try to resolve a URI to a wrap package, a wrapper, or a URI. Args: uri (Uri): The URI to resolve. resolution_context (UriResolutionContext): The resolution context. Returns: Any: URI Resolution Result. """ return self.client.try_resolve_uri(uri, self.resolution_context)