Possible to "unwrap" a Promise result type?

[rchl]

I have a custom Promise implementation in Python and I'm struggling with an issue where in certain situations the type of the value passed to .then is not "unwrapped" from the Promise type.

I'll post all the code at the end but here are some examples (keep in mind that it should work more or less like Javascript implementation):

1. In this example the final type of result is an int as expected

Promise.resolve(None) \
  .then(lambda _: 123)
  .then(lambda result: reveal_type(result))

2. In this example the type of final result is Promise[str | int] and that's an issue because the value passed to then should always be an "unwrapped" value. So the whole chain of promises should have been resolved so that the result type should no longer be wrapped in Promise. The implementation works like that but I'm not sure how to express that with types.

Promise.resolve(None) \
  .then(lambda result: Promise.resolve('abc' or 123)) \
  .then(lambda result: reveal_type(result))

The problem seems to be specific to Promise.resolve which uses different TypeVar than the main Promise class. Reasoning for that is in this issue: microsoft/pyright#3963

So my question is: Is it possible to achieve the expected types behavior in Python?

The simplified Promise implementation is as follows:

promise.py

import functools
import threading
from typing import Callable, Generic, List, Protocol, TypeVar, Union

T = TypeVar('T')
S = TypeVar('S')
T_contra = TypeVar('T_contra', contravariant=True)
TResult = TypeVar('TResult')


class ResolveFunc(Protocol[T_contra]):
    def __call__(self, resolve_value: T_contra) -> None:
        ...


FullfillFunc = Callable[[T], Union[TResult, 'Promise[TResult]']]
ExecutorFunc = Callable[[ResolveFunc[T]], None]


class Promise(Generic[T]):

    @staticmethod
    def resolve(resolve_value: S) -> 'Promise[S]':

        def executor_func(resolve_fn: ResolveFunc[S]) -> None:
            resolve_fn(resolve_value)

        return Promise(executor_func)

    def __init__(self, executor_func: ExecutorFunc[T]) -> None:
        self.resolved = False
        self.mutex = threading.Lock()
        self.callbacks = []  # type: List[ResolveFunc[T]]
        executor_func(lambda resolve_value=None: self._do_resolve(resolve_value))

    def then(self, onfullfilled: FullfillFunc[T, TResult]) -> 'Promise[TResult]':

        def callback_wrapper(resolve_fn: ResolveFunc[TResult], resolve_value: T) -> None:
            result = onfullfilled(resolve_value)
            # If returned value is a promise then this promise needs to be
            # resolved with the value of returned promise.
            if isinstance(result, Promise):
                result.then(lambda value: resolve_fn(value))
            else:
                resolve_fn(result)

        def sync_wrapper(resolve_fn: ResolveFunc[TResult]) -> None:
            callback_wrapper(resolve_fn, self._get_value())

        def async_wrapper(resolve_fn: ResolveFunc[TResult]) -> None:
            self._add_callback(functools.partial(callback_wrapper, resolve_fn))

        if self._is_resolved():
            return Promise(sync_wrapper)
        return Promise(async_wrapper)

    def _do_resolve(self, new_value: T) -> None:
        # No need to block as we can't change from resolved to unresolved.
        if self.resolved:
            raise RuntimeError("cannot set the value of an already resolved promise")
        with self.mutex:
            self.resolved = True
            self.value = new_value
            for callback in self.callbacks:
                callback(new_value)

    def _add_callback(self, callback: ResolveFunc[T]) -> None:
        with self.mutex:
            self.callbacks.append(callback)

    def _is_resolved(self) -> bool:
        with self.mutex:
            return self.resolved

    def _get_value(self) -> T:
        with self.mutex:
            return self.value

I've also created a pyright playground link with the implementation and some examples (including the failing one) at the bottom.

[erictraut]

Your implementation includes the type subexpression Union[TResult, Promise[TResult]]. Any time you create a union between a "naked" TypeVar and a type that uses that that same TypeVar as a type argument, you create the potential for multiple solutions within the constraint solver. Consider the following simplified example:

from typing import TypeVar

T = TypeVar("T")

def func(x: T | list[T]) -> T:
    ...

reveal_type(func([1])) # int or list[int]?

In this case, both int and list[int] are acceptable solutions. However, the solution int is probably the one you expect. How does the constraint solver choose between multiple valid answers? There is no standardized technique. Pyright uses a heuristic whereby it computes all of the solutions and "scores" each one. The score attempts to take into account several factors including the number of TypeVars that are solved (or left unsolved) and the "complexity" of the resulting solutions. The "complexity" of a type is difficult to compute because there's subjectivity involved. For example, which is more complex: str | int or Promise[int]?

In this particular case, the complexity calculation in pyright is producing a result that is somewhat unintuitive. I've tweaked the complexity calculation algorithm slightly to nudge it in the right direction. This will be included in the next release of pyright.

[rchl]

Thank you for detailed explanation (as always) and looking forward to trying the new version.