itamarst
mypy plugin and coroutines
Hi,
We just started integrating the 2.0 release into our codebase, and hit one issue. The crochet.mypy plugin is not checking if the function return value is a coroutine, it only propagates the return types exactly as is. So we're getting a lot of:
"Coroutine[Any, Any, List[Dict[str, Any]]]" has no attribute "__iter__" (not iterable)
...
BTW the release looks great. Many thanks.
EDIT: these are using crochet.wait_for, which I now see the plugin doesn't support
Ah, probably need to do a similar thing where we remove the Coroutine wrapper which is being added erroneously.
Can you give me a small reproducer that mypy complains about?
This should do it:
$ more mypy.ini
[mypy]
plugins = crochet.mypy
$ cat wait_for_it.py
import crochet
@crochet.wait_for(10)
async def foo() -> str:
return ''
def bar() -> str:
return foo()
$ mypy --strict --config=mypy.ini wait_for_it.py
wait_for_it.py:10: error: Incompatible return value type (got "Coroutine[Any, Any, str]", expected "str")
Found 1 error in 1 file (checked 1 source file)
The stub for wait_for may need to change as well, I guess something like:
@overload
def wait_for(timeout: float) -> Callable[[Callable[..., Coroutine[Any, Any, _T]]], _T]: ...
@overload
def wait_for(timeout: float) -> Callable[[_F], _F]: ...
Thanks for the reproducer!
@wait_for stub will not need to change, cause the return function deliberately does not return coroutines, those get wrapped into a blocking call. Just need to unwrap the types somehow in the mypy plugin.
Oh actually I'm probably wrong and it does need to change the typing stub, I forget it's a thing that returns a decorator.
... but unfortunately that won't work due to https://github.com/python/typing/issues/256, so probably will need to extend the plugin.
I attempted to update the plugin with limited results. I did discover that the annotation for wait_for needs to be:
@overload
def wait_for(timeout: float) -> Callable[[Callable[..., Coroutine[Any, Any, _T]]], Callable[..., _T]]: ...
@overload
def wait_for(timeout: float) -> Callable[[_F], _F]: ...
Hi again @itamarst, I was playing around with the new ParamSpec support in mypy==0.950 and had success with this:
...
from typing_extensions import ParamSpec
...
@overload
def wait_for(timeout: float) -> Callable[
[Callable[_P, Deferred[_T]]],
Callable[_P, _T]
]: ...
@overload
def wait_for(timeout: float) -> Callable[
[Callable[_P, Coroutine[Any, Any, _T]]],
Callable[_P, _T]
]: ...
@overload
def wait_for(timeout: float) -> Callable[
[Callable[_P, _T]],
Callable[_P, _T]
]: ...
I suspect that the need for the mypy plugin is gone.
Happy to accept a PR, I don't have a lot of time to spend on writing this myself.
Happy to accept a PR, I don't have a lot of time to spend on writing this myself.
I started a test-bed at https://github.com/itamarst/crochet/pull/144
Looks like we have to wait for https://github.com/python/mypy/issues/12595 though...
I managed to get this to work well:
@overload
def wait_for_x(f: Callable[_P, Deferred[_T]]) -> Callable[_P, _T]:
...
@overload
def wait_for_x(f: Callable[_P, Coroutine[None, None, _T]]) -> Callable[_P, _T]:
...
@overload
def wait_for_x(f: Callable[_P, _T]) -> Callable[_P, _T]:
...
But (correctly) making wait_for return a callable which returns a callable... did not:
@overload
def wait_for(f: float) -> Callable[[Callable[_P, Deferred[_T]]], Callable[_P, _T]]:
...
@overload
def wait_for(f: float) -> Callable[[Callable[_P, Coroutine[None, None, _T]]], Callable[_P, _T]]:
...
@overload
def wait_for(f: float) -> Callable[[Callable[_P, _T]], Callable[_P, _T]]:
...