Description
Bug Report
In the following synchronous code, mypy is correctly able to perform type narrowing to determine that the resource
variable must be an instance of Resource
when resource.do_thing()
is called inside the foo()
function:
from contextlib import AbstractContextManager
from typing import TypeVar
T = TypeVar("T")
Self = TypeVar("Self")
class ExitStack:
def __enter__(self: Self) -> Self:
return self
def __exit__(self, *args: object) -> None:
pass
def enter_context(self, cm: AbstractContextManager[T]) -> T:
return cm.__enter__()
class Resource:
def __enter__(self: Self) -> Self:
return self
def __exit__(self, *args: object) -> None:
pass
def do_thing(self) -> None:
pass
def foo(resource: Resource | None) -> None:
with ExitStack() as stack:
if resource is None:
resource = stack.enter_context(Resource())
resource.do_thing()
However, in an asynchronous version of this code, mypy is not able to narrow the type in the same way, leading to a false positive being emitted:
from contextlib import AbstractAsyncContextManager
from typing import TypeVar
T = TypeVar("T")
Self = TypeVar("Self")
class AsyncExitStack:
async def __aenter__(self: Self) -> Self:
return self
async def __aexit__(self, *args: object) -> None:
pass
async def enter_async_context(self, cm: AbstractAsyncContextManager[T]) -> T:
return await cm.__aenter__()
class AsyncResource:
async def __aenter__(self: Self) -> Self:
return self
async def __aexit__(self, *args: object) -> None:
pass
def do_thing(self) -> None:
pass
async def async_foo(resource: AsyncResource | None) -> None:
async with AsyncExitStack() as stack:
if resource is None:
resource = await stack.enter_async_context(AsyncResource())
resource.do_thing() # error: Item "None" of "Optional[AsyncResource]" has no attribute "do_thing"
Curiously, if we rewrite async_foo()
like this, the false-positive error goes away:
async def async_foo(resource: AsyncResource | None) -> None:
async with AsyncExitStack() as stack:
if resource is None:
res = await stack.enter_async_context(AsyncResource())
resource = res
resource.do_thing()
Real-life use case
My use case is to write a function like this using aiohttp
, in which a user can optionally supply an existing aiohttp.ClientSession()
instance if they want to, or let the function create one on the fly if not.
import aiohttp
from contextlib import AsyncExitStack
async def foo(session: aiohttp.ClientSession | None = None):
async with AsyncExitStack() as stack:
if session is None:
session = await stack.enter_async_context(aiohttp.ClientSession())
async with session.get("https://foo.com/json") as response:
data = await response.json()
To Reproduce
https://mypy-play.net/?mypy=latest&python=3.10&gist=0ee494765b5abc08fc9f161bc63656a7
Your Environment
- Mypy version used: 0.982; also occurs on mypy master (commit 8d5b641)