Source code for returns.contrib.mypy._features.flow
from mypy.plugin import FunctionContext
from mypy.types import Type as MypyType
from returns.contrib.mypy._typeops.inference import PipelineInference
[docs]def analyze(ctx: FunctionContext) -> MypyType:
"""
Helps to analyze ``flow`` function calls.
By default, ``mypy`` cannot infer and check this function call:
.. code:: python
>>> from returns.pipeline import flow
>>> assert flow(
... 1,
... lambda x: x + 1,
... lambda y: y / 2,
... ) == 1.0
But, this plugin can!
It knows all the types for all ``lambda`` functions in the pipeline.
How?
1. We use the first passed parameter as the first argument
to the first passed function
2. We use parameter + function to check the call and reveal
types of current pipeline step
3. We iterate through all passed function and use previous
return type as a new parameter to call current function
"""
if not ctx.arg_types[0]:
return ctx.default_return_type
if not ctx.arg_types[1]: # We do require to pass `*functions` arg.
ctx.api.fail('Too few arguments for "flow"', ctx.context)
return ctx.default_return_type
# We use custom argument type inference here,
# because for some reason, `mypy` does not do it correctly.
# It inferes `covariant` types incorrectly.
real_arg_types = tuple(
ctx.api.expr_checker.accept(arg) # type: ignore
for arg in ctx.args[1]
)
return PipelineInference(
ctx.arg_types[0][0],
).from_callable_sequence(
real_arg_types,
ctx.arg_kinds[1],
ctx,
)