Mypy 严格的可选检查失败，调用另一个 class 方法来设置可选

Question

我创建了一个 class 并将 writer 和 reader 设置为选项，在 __init__ 中设置为 None。正确的用法需要先调用 open，以确保正确初始化 writer 和 reader。

在 read_write 方法中，我检查 None 并在需要时自动完成初始化。但是 Mypy 仍然将此视为严格可选检查错误。

import asyncio
from typing import Optional


class AsyncTest:
    def __init__(self, host: str, port: int = 502) -> None:
        self.host = host
        self.port = port
        self.writer: Optional[asyncio.StreamWriter] = None
        self.reader: Optional[asyncio.StreamReader] = None

    async def open(self) -> None:
        self.reader, self.writer = await asyncio.open_connection(self.host, self.port)

    async def read_write(self) -> None:
        if self.writer is None or self.reader is None:
            await self.open()

        self.writer.write(b'')
        await self.writer.drain()
        resp = await self.reader.readexactly(100)

    async def close(self) -> None:
        if self.writer:
            self.writer.close()
            await self.writer.wait_closed()

给出：

$ mypy test.py
test.py:19: error: Item "None" of "Optional[StreamWriter]" has no attribute "write"
test.py:20: error: Item "None" of "Optional[StreamWriter]" has no attribute "drain"
test.py:21: error: Item "None" of "Optional[StreamReader]" has no attribute "readexactly"

现在，如果我通过将对 self.open() 的调用替换为对 open_connection 的调用来更改 read_write 方法，Mypy 将不再抱怨。

async def read_write(self) -> None:
    if self.writer is None or self.reader is None:
        self.reader, self.writer = await asyncio.open_connection(self.host, self.port)

    self.writer.write(b'')
    await self.writer.drain()
    resp = await self.reader.readexactly(100)

作为临时解决方案，我显然可以在 Mypy 中禁用对该文件的严格可选检查，但有没有办法让它工作？这将要求 Mypy 跟随对 open 的方法调用并检查它是否设置了 self.writer 和 self.reader 属性。

在这个具体的小例子中，open_connection 可以直接放在 read_write 中并完全删除 open 方法，但如果 open方法增长。

Answer 1

MyPy 只知道签名，不知道副作用。 open的签名不能表示writer和reader是调用后定义的

不是在外部检查对象是否已经 "opened" 并调用 open 否则，将检查和打开移动到一个方法中。该方法可以静态保证提供reader/writer，动态记忆

class AsyncTest:
    ...

    # method always provides valid reader/writer
    async def _connection(self) -> Tuple[asyncio.StreamReader, asyncio.StreamWriter]:
        # internally check if already opened
        if self.reader is None or self.writer is None:
            self.reader, self.writer = await asyncio.open_connection(self.host, self.port)
        return self.reader, self.writer

    async def read_write(self) -> None:
        # other methods call connection method unconditionally
        reader, writer = self._connection()
        writer.write(b'')
        await writer.drain()
        resp = await reader.readexactly(100)

Mypy 严格的可选检查失败，调用另一个 class 方法来设置可选

Mypy strict optional checking fails with call to another class method to set optionals

python

python-3.x

mypy