Source code for py_gql.execution.runtime.blocking

# -*- coding: utf-8 -*-

from typing import Any, Callable, Iterable, Optional, Tuple, Type, TypeVar

from .base import Runtime


T = TypeVar("T")
E = TypeVar("E", bound=Exception)
TException = TypeVar("TException", bound=Type[Exception])
AnyFn = Callable[..., Any]



[docs]class BlockingRuntime(Runtime):
    """
    Default runtime implementation which blocks the current thread.
    """


[docs]    def submit(self, fn: AnyFn, *args: Any, **kwargs: Any) -> Any:
        return fn(*args, **kwargs)



[docs]    def ensure_wrapped(self, value: Any) -> Any:
        return value



[docs]    def gather_values(self, values: Iterable[Any]) -> Any:
        return list(values)



[docs]    def map_value(
        self,
        value: Any,
        then: Callable[[Any], T],
        else_: Optional[Tuple[Type[E], Callable[[E], T]]] = None,
    ) -> Any:
        try:
            return then(value)
        except Exception as err:
            if else_ and isinstance(err, else_[0]):
                return else_[1](err)
            raise



[docs]    def unwrap_value(self, value):
        return value



[docs]    def wrap_callable(self, func: AnyFn) -> AnyFn:
        return func