GV

GeezerVison

The World Wide Web, restored to 800 pixels of glory.

Home|About|Gaming cabinet|Archives|How it works|Starter sites

On this site

  • Sign in
  • ActionsCI/CD and automation
  • CodespacesCloud development environments
  • All Features
  • Documentation
  • Training
  • Certifications
  • Build with Copilot
  • Customize Copilot
  • Integrations
  • Trending repos
  • GitHub Sponsors
  • Developer articles
  • Enterprises

This view

  • Original page
  • New address

GeezerVison is showing github.com — What is the correct way to subclass asyncio.StreamReader? · Issue #12423 · python/typeshed

What is the correct way to subclass asyncio.StreamReader? #12423

New issue
Copy link
New issue
Copy link
Open
Open
What is the correct way to subclass asyncio.StreamReader?#12423
Copy link
Labels
topic: asyncioAsyncio-related issuesAsyncio-related issues

Description

@jgarvin
jgarvin
opened on Jul 24, 2024
Issue body actions

#11755 changed the signature of asyncio.StreamReader.readuntil to make the separator argument take an internal type _ReaduntilBuffer that just has ... for its definition. This is confusing because inside the python 3.12 stdlib code this argument is already typed as bytes, which is the obvious choice. When I try to write my own subclass that takes bytes for the argument mypy complains I'm violating Liskov. Since this class is specifically designed as a base class that users would want to subclass themselves, it seems like there should be a way to do this that doesn't use internal types, no?

import asyncio
import typing

class StreamReaderWrapper(asyncio.StreamReader):
    def __init__(self):
        self._reader: asyncio.StreamReader | None = None

    @typing.override
    async def read(self, n: int = -1) -> bytes:
        assert self._reader is not None
        return await self._reader.read(n)

    @typing.override
    async def readline(self) -> bytes:
        assert self._reader is not None
        return await self._reader.readline()

    @typing.override
    async def readexactly(self, n: int) -> bytes:
        assert self._reader is not None
        return await self._reader.readexactly(n)

    @typing.override
    async def readuntil(self, separator: bytes = b'\n') -> bytes:
        assert self._reader is not None
        return await self._reader.readuntil(separator)
foo.py:4: error: Argument 1 of "readuntil" is incompatible with supertype "StreamReader"; supertype defines the argument type as "_ReaduntilBuffer"  [override]
foo.py:4: note: This violates the Liskov substitution principle
foo.py:4: note: See https://mypy.readthedocs.io/en/stable/common_issues.html#incompatible-overrides

Note that using bytes | bytearray | memoryview as the type instead (taken from #11755) doesn't work either, same error

Activity

  1. AlexWaygood
    added
    topic: redisIssues related to the redis third-party stubsIssues related to the redis third-party stubs
    topic: asyncioAsyncio-related issuesAsyncio-related issues
    and removed
    topic: redisIssues related to the redis third-party stubsIssues related to the redis third-party stubs
    on Aug 1, 2024
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

No one assigned

    Labels

    topic: asyncioAsyncio-related issuesAsyncio-related issues

    Projects

    No projects

      Milestone

      No milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions

      You are visitor 01402503 — GeezerVison does not store the pages you look up.
      Best viewed with Netscape Navigator 4.7 or Internet Explorer 6 at 800×600. No cookies required.