You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何让telnetlib3的TelnetReader通过StreamReader类型检查?

解决telnetlib3与asyncio StreamReader/Writer类型兼容问题

你的问题核心在于:工具函数注解了具体的asyncio.StreamReader类,但telnetlib3.TelnetReader只是实现了相同的接口(Protocol)而非子类,导致类型检查器不认可。以下是几种解决方案:

方案1:自定义Protocol抽象接口(推荐)

定义一个包含你需要的StreamReader核心方法的Protocol,让工具函数接收这个Protocol类型,这样TelnetReader和原生StreamReader都能通过类型检查:

import asyncio
import telnetlib3
from typing import Protocol, Awaitable, bytes

class ReadableStream(Protocol):
    def read(self, n: int = ...) -> Awaitable[bytes]:
        ...

async def utility_func(reader: ReadableStream):
    out = await reader.read(10000)
    print(out.decode())

async def main():
    tr = telnetlib3.TelnetReader()
    await utility_func(tr)
    # 也可以正常传入原生StreamReader
    reader, writer = await asyncio.open_connection('example.com', 80)
    await utility_func(reader)

asyncio.run(main())

这种方式保留了类型检查的安全性,同时让工具函数保持通用,不依赖具体实现类。

方案2:用typing.cast临时绕过(仅应急)

如果不想修改工具函数的类型注解,可以通过cast强制转换类型,告知类型检查器对象符合要求:

import asyncio
import telnetlib3
from typing import cast

async def utility_func(reader: asyncio.StreamReader):
    out = await reader.read(10000)
    print(out.decode())

async def main():
    tr = telnetlib3.TelnetReader()
    await utility_func(cast(asyncio.StreamReader, tr))

asyncio.run(main())

注意:这种方法会跳过类型检查的安全性验证,如果后续TelnetReader的接口与StreamReader出现差异,类型检查器无法提前发现问题。

方案3:使用telnetlib3提供的Protocol(若存在)

检查telnetlib3的官方文档,若库本身暴露了对应的Protocol类型(比如TelnetReaderProtocol),可以直接导入使用,让工具函数注解为该类型,同时兼容原生StreamReader:

import asyncio
import telnetlib3
from telnetlib3 import TelnetReaderProtocol

async def utility_func(reader: TelnetReaderProtocol):
    out = await reader.read(10000)
    print(out.decode())

async def main():
    tr = telnetlib3.TelnetReader()
    await utility_func(tr)

asyncio.run(main())

内容的提问来源于stack exchange,提问作者Wör Du Schnaffzig

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.06.12 06:33:14