如何正确输入 asyncio class 实例变量

Question

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

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 可以是 None 在 send 方法和代码不是类型安全的。我无法在 __init__ 中初始化 _connection 变量，因为它需要在协程中异步运行。在 __init__ 之外声明变量也可能不是一个好主意。有什么办法可以解决这个问题吗？或者您推荐另一种 (OOP) 设计来解决这个问题？

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

Answer 1

It's probably a bad idea to declare the variable outside __init__ too

这很接近。您必须在 __init__.

之外对其进行注释

class Example:
    _connection: Connection

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

Answer 2

让类处于不可用状态通常不是好的设计，除非对它们调用某些方法。替代方案是 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!")

如何正确输入 asyncio class 实例变量

How to correctly type an asyncio class instance variables

python

type-hinting

python-asyncio

mypy

python-typing