2828from enum import IntEnum
2929from functools import reduce
3030
31- __version__ = "3.0.1 "
31+ __version__ = "3.1.0 "
3232
3333from typing import Any # noqa: F401
3434from typing import Generator # noqa: F401
@@ -103,6 +103,7 @@ class Classes(Asn1Enum):
103103
104104
105105class Encoding (Asn1Enum ):
106+ """ASN.1 encoding types."""
106107 DER = 1 # Distinguished Encoding Rules
107108 CER = 2 # Canonical Encoding Rules
108109
@@ -119,6 +120,12 @@ class ReadFlags(IntEnum):
119120`Decoder.read()`.
120121"""
121122
123+ EncoderStream = Union [io .RawIOBase , io .BufferedWriter , None ]
124+ """A stream that can be used for output."""
125+
126+ DecoderStream = Union [io .RawIOBase , io .BufferedIOBase , bytes ]
127+ """A stream that can be used for input."""
128+
122129
123130def to_int_2c (values ): # type: (bytes) -> int
124131 """Bytes to integer conversion in two's complement."""
@@ -205,19 +212,19 @@ class Encoder(object):
205212
206213 def __init__ (self ): # type: () -> None
207214 """Constructor."""
208- self ._stream = None # type: Union[io.RawIOBase, io.BufferedWriter, None] # Output stream
215+ self ._stream = None # type: EncoderStream # Output stream
209216 self ._encoding = None # type: Union[Encoding, None] # Encoding type (DER or CER)
210217 self ._stack = None # type: List[List[bytes]] | None # Stack of encoded data
211218
212219 def start (self , stream = None , encoding = None ):
213- # type: (Union[io.RawIOBase, io.BufferedWriter, Encoding, None] , Union[Encoding, None]) -> None
220+ # type: (EncoderStream , Union[Encoding, None]) -> None
214221 """
215222 This method instructs the encoder to start encoding a new ASN.1
216223 output. This method may be called at any time to reset the encoder
217224 and reset the current output (if any).
218225
219226 Args:
220- stream (io.RawIOBase or Encoding or None ): The output stream or the encoding with no stream.
227+ stream (EncoderStream ): The output stream or the encoding with no stream.
221228 encoding (Encoding or None): The encoding (DER or CER) to use.
222229 """
223230 # stream is an Encoding, encoding is an Encoding: this not allowed
@@ -724,7 +731,7 @@ def __init__(self): # type: () -> None
724731 self ._levels = 0 # type: int # Number of recursive calls
725732 self ._ends = [] # type: List[int] # End of the current element (or INDEFINITE_FORM) for enter / leave
726733
727- def start (self , stream ): # type: (Union[io.RawIOBase, io.BufferedIOBase, bytes] ) -> None
734+ def start (self , stream ): # type: (DecoderStream ) -> None
728735 """
729736 This method instructs the decoder to start decoding an ASN.1 input.
730737 This method may be called at any time to start a new decoding job.
@@ -736,7 +743,7 @@ def start(self, stream): # type: (Union[io.RawIOBase, io.BufferedIOBase, bytes]
736743 assumes the input is in BER or DER format.
737744
738745 Args:
739- stream (bytes or io.RawIOBase ): ASN.1 input, in BER or DER format, to be decoded.
746+ stream (DecoderStream ): ASN.1 input, in BER or DER format, to be decoded.
740747
741748 Returns:
742749 None
@@ -745,7 +752,7 @@ def start(self, stream): # type: (Union[io.RawIOBase, io.BufferedIOBase, bytes]
745752 `Error`
746753 """
747754 if not isinstance (stream , bytes ) and not isinstance (stream , io .RawIOBase ) and not isinstance (stream , io .BufferedIOBase ):
748- raise Error ('Expecting bytes or a subclass of io.RawIOBase. Get {} instead.' .format (type (stream )))
755+ raise Error ('Expecting bytes or a subclass of io.RawIOBase or BufferedIOBase . Get {} instead.' .format (type (stream )))
749756
750757 self ._stream = io .BytesIO (stream ) if isinstance (stream , bytes ) else stream # type: ignore
751758 self ._tag = None
0 commit comments