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

user_991805 0
Viewed 345

I create a class and setup the writer and reader as optionals set to None in __init__. Correct usage would require calling open before anything else, to ensure the writer and reader are correctly initialized.

In the read_write method I check for None and if needed initializing is automatically done. However Mypy still treats this as an Strict Optional Checking Error.

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()

Gives:

$ 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"

Now if I change the read_write method by replacing the call to self.open() with the call to open_connection Mypy no longer complains.

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)

I can obviously disable strict optional checking for that file in Mypy as a temporary solution, but is there a way to get this to work? This would require Mypy to follow the method call to open and check whether it sets the self.writer and self.reader attributes.

In this specific small example the open_connection could be put directly inside the read_write and completely remove the open method, but that hardly seems optimal if the open method grows.

mypy python-3.x python
1 Answers

MyPy is only aware of signatures, not side-effects. The signature of open cannot express that writer and reader are defined after calling it.

Instead of externally checking if the object has been "opened" and calling open otherwise, move the checking and opening into a method. This method can statically guarantee providing reader/writer, and dynamically memoize them.

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)
user_53499160
Related