Consider the following example class containing attributes that require running a coroutine for initialization:
class Example:
def __init__(self) -> None:
self._connection: Optional[Connection] = None
async def connect() -> None:
self._connection = await connect_somewhere(...)
async def send(data: bytes) -> None:
self._connection.send(data)
If I run mypy (perhaps with strict-optional enabled) on this example, it will complain that _connection can be None in send method and the code is not type-safe. I can't initialize the _connection variable in __init__, as it needs to be run asynchronously in a coroutine. It's probably a bad idea to declare the variable outside __init__ too. Is there any way to solve this? Or do you recommend another (OOP) design that would solve the issue?
Currently, I either ignore the mypy complaints, prepend assert self._connection before each usage or append # type: ignore after the usage.
It is generally not good design to have classes in an unusable state unless some method is called on them. An alternative is dependency injection and an alternative constructor:
from typing import TypeVar, Type
# not strictly needed – one can also use just 'Example'
# if inheritance is not needed
T = TypeVar('T')
class Example:
# class always receives a fully functioning connection
def __init__(self, connection: Connection) -> None:
self._connection = connection
# class can construct itself asynchronously without a connection
@classmethod
async def connect(cls: Type[T]) -> T:
return cls(await connect_somewhere(...))
async def send(self, data: bytes) -> None:
self._connection.send(data)
This frees __init__ from relying on some other initialiser to be called later on; as a bonus, it is possible to provide a different connection, e.g. for testing.
The alternative constructor, here connect, still allows to create the object in a self-contained way (without the callee knowing how to connect) but with full async support.
async def example():
# create instance asynchronously
sender = await Example.connect()
await sender.send(b"Hello ")
await sender.send(b"World!")
To get the full life-cycle of opening and closing, supporting async with is the most straightforward approach. This can be supported in a similar way to the alternative constructor – by providing an alternative construct as a context manager:
from typing import TypeVar, Type, AsyncIterable
from contextlib import asynccontextmanager
T = TypeVar('T')
class Example:
def __init__(self, connection: Connection) -> None:
self._connection = connection
@asynccontextmanager
@classmethod
async def scope(cls: Type[T]) -> AsyncIterable[T]:
connection = await connect_somewhere(...) # use `async with` if possible!
try:
yield cls(connection)
finally:
connection.close()
async def send(self, data: bytes) -> None:
self._connection.send(data)
Alternative connect constructor omitted for brevity. For Python 3.6, asynccontextmanager can be fetched from the asyncstdlib (Disclaimer: I maintain this library).
There is a general caveat: closing does leave objects in an unusable – thus inconsistent – state practically by definition. Python's type system has no way to separate "open Connection" from "closed Connection", and especially not to detect that .close or the end of a context transitions from one to the other.
By using async with one partially side-steps this issue, since context managers are generally understood not to be useable after their block by convention.
async def example():
async with Example.scope() as sender:
await sender.send(b"Hello ")
await sender.send(b"World!")