varun-r-mallya/py-libp2p
1
0
Fork 0
mirror of https://github.com/varun-r-mallya/py-libp2p.git synced 2026-02-10 07:00:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: improve types according to new typecheck

Browse Source
This commit is contained in:
Akash Mondal
2025-06-09 19:10:15 +00:00
parent 47ae20d29c
commit 13d730ae5c
2 changed files with 7 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
libp2p/network/stream/net_stream.py
View File

@ -1,9 +1,6 @@
from enum import (
Enum,
)
from typing import (
Optional,
)
import trio
@ -102,11 +99,11 @@ class NetStream(INetStream):
"""
muxed_stream: IMuxedStream
protocol_id: Optional[TProtocol]
protocol_id: TProtocol | None
__stream_state: StreamState
def __init__(
self, muxed_stream: IMuxedStream, nursery: Optional[trio.Nursery] = None
self, muxed_stream: IMuxedStream, nursery: trio.Nursery | None = None
) -> None:
super().__init__()
@ -142,7 +139,7 @@ class NetStream(INetStream):
async with self._state_lock:
return self.__stream_state
async def read(self, n: Optional[int] = None) -> bytes:
async def read(self, n: int | None = None) -> bytes:
"""
Read from stream.
@ -279,7 +276,7 @@ class NetStream(INetStream):
if hasattr(swarm, "refs") and hasattr(swarm.refs, "done"):
swarm.refs.done()
def get_remote_address(self) -> Optional[tuple[str, int]]:
def get_remote_address(self) -> tuple[str, int] | None:
"""Delegate to the underlying muxed stream."""
return self.muxed_stream.get_remote_address()