时间戳

首页 > 解决方案 > 如何正确键入异步类实例变量

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

问题描述

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

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

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

标签： pythonpython-asynciotype-hintingmypypython-typing

解决方案

除非对它们调用某种方法，否则使类处于不可用状态通常不是好的设计。另一种方法是依赖注入和另一种构造函数：

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可以从（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!")

推荐阅读