Skip to content

Commit

Permalink
Fix a mypy typing issue for re.Pattern
Browse files Browse the repository at this point in the history
  • Loading branch information
ronf committed Oct 14, 2023
1 parent e97b9d6 commit bb45b65
Showing 1 changed file with 7 additions and 6 deletions.
13 changes: 7 additions & 6 deletions asyncssh/stream.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,8 +24,8 @@
import inspect
import re
from typing import TYPE_CHECKING, Any, AnyStr, AsyncIterator
from typing import Callable, Dict, Generic, Iterable
from typing import List, Optional, Set, Tuple, Union, cast
from typing import Callable, Dict, Generic, Iterable, List
from typing import Optional, Pattern, Set, Tuple, Union, cast

from .constants import EXTENDED_DATA_STDERR
from .logging import SSHLogger
Expand Down Expand Up @@ -590,19 +590,20 @@ async def readuntil(self, separator: object, datatype: DataType,
if separator is _NEWLINE:
seplen = 1
separators = cast(AnyStr, '\n' if self._encoding else b'\n')
pat = re.compile(separators)
elif isinstance(separator, (bytes, str)):
seplen = len(separator)
separators = re.escape(cast(AnyStr, separator))
elif isinstance(separator, re.Pattern):
pat = re.compile(re.escape(cast(AnyStr, separator)))
elif isinstance(separator, Pattern):
seplen = max_separator_len
separators = separator
pat = cast(Pattern[AnyStr], separator)
else:
bar = cast(AnyStr, '|' if self._encoding else b'|')
seplist = list(cast(Iterable[AnyStr], separator))
seplen = max(len(sep) for sep in seplist)
separators = bar.join(re.escape(sep) for sep in seplist)
pat = re.compile(separators)

pat = re.compile(separators)
curbuf = 0
buflen = 0

Expand Down

0 comments on commit bb45b65

Please sign in to comment.