We have a problem with @pipeline and unwrap() about returns HOT 14 CLOSED

It all translates to

Decorator:8(
  Var(test)
  NameExpr(pipeline [returns.pipeline.pipeline])
  FuncDef:9(
    test
    Args(
      Var(arg))
    def (arg: builtins.int) -> returns.result.Result[builtins.int, builtins.Exception]
    Block:9(
      ExpressionStmt:10(
        CallExpr:10(
          MemberExpr:10(
            CallExpr:10(
              NameExpr(Failure [returns.result.Failure])
              Args(
                StrExpr(a)))
            unwrap)
          Args()))
      ReturnStmt:11(
        CallExpr:11(
          NameExpr(Success [returns.result.Success])
          Args(
            NameExpr(arg [l])))))))

test.py:14: error: Revealed type is 'returns.result.Result[builtins.int, builtins.Exception]'

inside mypy, so we can run extra checks on the contents of the @pipeline.

from returns.

When used as pipeline(test)(1) it translates to:

CallExpr:13(
  NameExpr(pipeline [returns.pipeline.pipeline])
  Args(
    NameExpr(test [test.test])))

So, we can forbid to use @pipeline as a function. Only as a decorator.

from returns.

Some code samples:

if fullname == 'returns.pipeline.pipeline':
            def _analyze(function_ctx: FunctionContext):
                from mypy.subtypes import check_type_parameter
                from mypy.nodes import COVARIANT
                print(
                    function_ctx.default_return_type.arg_types[0],
                    function_ctx.default_return_type.ret_type,
                    function_ctx.default_return_type.ret_type.type.module_name,
                    function_ctx.default_return_type.ret_type.type.name(),
                )
                print(
                    'direct',
                    check_type_parameter(
                        function_ctx.default_return_type.arg_types[0],
                        function_ctx.default_return_type.ret_type,
                        COVARIANT,
                    )
                )

                # function_ctx.api.errors.report(1, 1, 'Test error')
                return function_ctx.default_return_type
            return _analyze

from returns.

I am still playing around with @pipeline:

from typing import TypeVar, Iterable, Generator, Iterator, Any

from returns.result import Result, Success, Failure

T = TypeVar('T', bound=Result)
V = TypeVar('V')
E = TypeVar('E')

def unwrap(monad: Result) -> Iterator[int]:
    yield monad.unwrap()

def example(number: int) -> Generator[Result[int, Exception], int, None]:
    a = yield Success(1)
    b = yield Success(2)
    # yield Failure('a')
    # reveal_type(a)
    print('a', a)
    print('b', b)
    # yield a
    yield Success(a + b + number)

gen = example(2)
monad = gen.send(None)  # type: ignore
print('send None', monad)

monad = gen.send(monad.unwrap())
print('send 1', monad)

monad = gen.send(monad.unwrap())
print('send 2', monad)

This code does not produce any type errors in the user's space.
And works correctly.

Try to uncomment to play around:

• reveal_type
• yield Failure('a')
• yield a

from returns.

Output:

» python test.py
send None <Success: 1>
send 1 <Success: 2>
a 1
b 2
send 2 <Success: 5>

And:

» mypy test.py --show-traceback
test.py:15: error: Revealed type is 'builtins.int'

from returns.

The thing is we cannot change the type of monads inside the pipeline:

def example(number: int) -> Generator[Result[int, Exception], int, None]:
    a = yield Success(1)
    b = yield Success('2')
    yield Success(a + int(b) + number)

Output:

» mypy test.py --show-traceback
test.py:14: error: Argument 1 to "Success" has incompatible type "str"; expected "int"

This is 100% required. Since without allowing different Result types inside the pipeline - you cannot compose functions. It is better to have type errors with Failure(bool).unwrap() than this.

from returns.

Now I will go for new mypy plugin to analyze @pipeline contents.

from returns.

Solutions:

• Try visit_ to find Failure.unwrap() inside the source code of @pipeline and hack the context with some dirty magic
• Tree transform .unwrap() to .bind() or .map()
• Try to add extra annotations to Failure and Success to possibly annotate the other value (now it is Any by default)

from returns.

There's literally nothing I can do about it.

0. Current APi has this bug. Possibly some others. We can still live with it
1. Generator-based APIs do not work, since they require inner-function to have Generator[] return type and do not allow different yield types. It is impossible to mix x = yield Success(1) and `y= yield Success('a') in one context. Dead end
2. visit_ is a not working for now. That looked promising: we need to find all .unwrap() calls on Failure type and match it to the function's error-return type. Like so: Failure(1).unwrap() matches -> Result[_, int]. So, it can be used. Otherwise - not. But, as I said I was not able to fix it. Since I cannot get context for the .unwrap() call neither I cannot traverse the tree of node inside @pipeline decorator. This might still work in the future
3. Transform is not possible, since it is too hard. Dead end
4. pipeline = Pipeline(), pipeline(callable...), and pipeline.unwrap(monad) did not work, since I was not able to match the types as I needed too. The logic is the same: we can only allow to pipeline.unwrap types with the same error-type as in Result[_, E]. However, this still might be an option

from returns.

Here's the deal:

1. we need to find returns.result.Failure.unwrap() calls in @pipeline decorator in a custom mypy plugin.
2. we get return type from the decorated function, we only care about the error type, because value type is checked with mypy correctly: -> Result[_, ErrorType]
3. we get failure types from Failure instances that call .unwrap()
4. we compare them with mypy's type checking: we allow things with the same type or more general type: raises int -> returns float is allowed, raises ValueError -> returns Exception is allowed, raises ValueError -> returns IndexError is not allowed
5. we raise warnings with ctx.api.fail for nodes that do not pass our check

Tree looks like so:

Decorator:8(
  Var(test)
  NameExpr(pipeline [returns.pipeline.pipeline])
  FuncDef:9(
    test
    Args(
      Var(arg))
    def (arg: builtins.int) -> returns.result.Result[builtins.int, builtins.Exception]
    Block:9(
      ExpressionStmt:10(
        CallExpr:10(
          MemberExpr:10(
            CallExpr:10(
              NameExpr(Failure [returns.result.Failure])
              Args(
                StrExpr(a)))
            unwrap)
          Args()))
      ReturnStmt:11(
        CallExpr:11(
          NameExpr(Success [returns.result.Success])
          Args(
            NameExpr(arg [l])))))))

from returns.

This works correctly with 0.10.0 and warn_unreachable = True:

from returns import Result, Success, Failure, pipeline


@pipeline
def call(a: str) -> Result[int, str]:
    Failure('a').unwrap()
    return Failure(str).unwrap()

call('a')

Call:

» mypy ex.py
ex.py:8: error: Statement is unreachable

from returns.

But still fails on:

from typing import Union
from returns import Result, Success, Failure, pipeline

x: Result[bool, bool]

@pipeline
def call(a: str) -> Result[int, str]:
    x.unwrap()
    return Failure(str).unwrap()

call('a')

Output:

» mypy ex.py

from returns.

The same happens with Success(...).failure()

from returns.

Nice idea: https://github.com/gcanti/fp-ts-contrib/blob/c94199e82fbf5c840a3bd4fb31653207ad053626/src/Do.ts

We can invest into creating a plugin that can pass TypedDict or Protocol between steps for better typing.

from returns.