如何正确键入异步类实例变量

Question

考虑以下包含需要运行协程进行初始化的属性的示例类：

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)

如果我在此示例中运行 mypy（可能启用了严格可选），它会抱怨 _connection 在 send 方法中可以为 None 并且代码不是类型安全的。我无法初始化__init__ 中的_connection 变量，因为它需要在协程中异步运行。在__init__ 之外声明变量可能也是个坏主意。有没有办法解决这个问题？或者您是否推荐另一种 (OOP) 设计来解决该问题？

目前，我要么忽略 mypy 投诉，在每次使用前添加 assert self._connection，要么在使用后添加 # type: ignore。

Answer 1

除非对它们调用某种方法，否则使类处于不可用状态通常不是好的设计。一个替代方案是 dependency injection 和一个替代构造函数：

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)

这使__init__ 不再依赖于稍后调用的其他初始化程序；作为奖励，可以提供不同的连接，例如用于测试。

替代构造函数，这里是connect，仍然允许以独立的方式创建对象（被调用者不知道如何连接），但完全支持async。

async def example():
    # create instance asynchronously
    sender = await Example.connect()
    await sender.send(b"Hello ")
    await sender.send(b"World!")

---

要获得打开和关闭的完整生命周期，支持async with 是最直接的方法。这可以通过与替代构造函数类似的方式得到支持——通过提供替代构造作为上下文管理器：

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)

为简洁起见，省略了替代 connect 构造函数。对于 Python 3.6，asynccontextmanager 可以从 the asyncstdlib 获取（免责声明：我维护这个库）。

有一个普遍的警告：关闭确实会使对象处于不可用的状态 - 因此不一致 - 实际上根据定义。 Python 的类型系统无法将“打开的Connection”与“关闭的Connection”分开，尤其是无法检测到.close 或上下文从一个转换到另一个的结束。

通过使用async with 可以部分回避这个问题，因为通常认为上下文管理器在按照惯例阻止后无法使用。

async def example():
    async with Example.scope() as sender:
        await sender.send(b"Hello ")
        await sender.send(b"World!")

Answer 2

在__init__ 之外也声明变量可能是个坏主意

这很接近。您必须在 __init__ 之外对其进行注释。

class Example:
    _connection: Connection

    async def connect(self) -> None:
        self._connection = await connect_somewhere(…)