v0.1.0

Added type validation, removed support for Python 3.6

.github/workflows/python-pytest.yml

@@ -16,7 +16,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python-version: ["3.6", "3.7", "3.8", "3.9", "3.10"]
+        python-version: ["3.7", "3.8", "3.9", "3.10"]
 
     steps:
     - uses: actions/checkout@v2

README.md

@@ -1,6 +1,6 @@
 # `dag-cbor`: A Python implementation of [DAG-CBOR](https://ipld.io/specs/codecs/dag-cbor/spec/)
 
-[![Generic badge](https://img.shields.io/badge/python-3.6+-green.svg)](https://docs.python.org/3.6/)
+[![Generic badge](https://img.shields.io/badge/python-3.7+-green.svg)](https://docs.python.org/3.7/)
 ![PyPI version shields.io](https://img.shields.io/pypi/v/dag-cbor.svg)
 [![PyPI status](https://img.shields.io/pypi/status/dag-cbor.svg)](https://pypi.python.org/pypi/dag-cbor/)
 [![Checked with Mypy](http://www.mypy-lang.org/static/mypy_badge.svg)](https://github.com/python/mypy)

dag_cbor/__init__.py

@@ -38,7 +38,7 @@
     data, which is not natively handled by `cbor2`).
 """
 
-__version__ = "0.0.7"
+__version__ = "0.1.0"
 
 from .encoding import encode as encode
 from .decoding import decode as decode

dag_cbor/decoding.py

@@ -28,6 +28,8 @@
 import math
 import struct
 from typing import Any, Dict, Callable, List, Optional, Tuple, Union
+from typing_validation import validate
+
 
 import cid # type: ignore
 
@@ -93,6 +95,9 @@ def decode(stream_or_bytes: Union[BufferedIOBase, bytes], *,
             [0, 1]
         ```
     """
+    validate(stream_or_bytes, Union[BufferedIOBase, bytes])
+    validate(allow_concat, bool)
+    # validate(callback, Optional[DecodeCallback]) # not yet supported by typing_validation
     if isinstance(stream_or_bytes, bytes):
         stream: BufferedIOBase = BytesIO(stream_or_bytes)
     else:

dag_cbor/encoding.py

@@ -28,6 +28,7 @@
 import math
 import struct
 from typing import Any, Dict, List, Optional, overload, Union
+from typing_validation import validate
 
 import cid # type: ignore
 
@@ -69,6 +70,8 @@ def encode(data: EncodableType, stream: None = None) -> bytes:
                 ...
         ```
     """
+    validate(data, EncodableType)
+    validate(stream, Optional[BufferedIOBase])
     if stream is None:
         internal_stream = BytesIO()
         _encode(internal_stream, data)

dag_cbor/random.py

@@ -82,6 +82,7 @@
 import sys
 from types import MappingProxyType
 from typing import Any, Dict, Iterator, List, Mapping, Optional
+from typing_validation import validate
 
 import cid # type: ignore
 import multihash # type: ignore
@@ -232,6 +233,13 @@ def set_options(*,
 
     """
     # pylint: disable = too-many-branches, too-many-locals, too-many-statements
+    for iarg in (seed, min_int, max_int, min_bytes, max_bytes, min_chars, max_chars,
+                 min_codepoint, max_codepoint, min_len, max_len, max_nesting, float_decimals):
+        validate(iarg, Optional[int])
+    for barg in (canonical, include_cid):
+        validate(barg, Optional[bool])
+    for farg in (min_float, max_float):
+        validate(farg, Optional[float])
     global _options
     global _rand
     # set newly passed options
@@ -336,6 +344,11 @@ def rand_data(n: Optional[int] = None, *,
         - integer values >= 0 mean that containers will be generated, with items generated by `random_data(max_nesting=max_nesting-1)`
         - no other values are valid
     """
+    validate(n, Optional[int])
+    validate(max_nesting, Optional[int])
+    return _rand_data(n, max_nesting=max_nesting)
+
+def _rand_data(n: Optional[int] = None, *, max_nesting: Optional[int] = None) -> Iterator[EncodableType]:
     if n is not None and n < 0:
         raise ValueError()
     if max_nesting is None:
@@ -344,14 +357,14 @@ def rand_data(n: Optional[int] = None, *,
         raise ValueError("Value for max_nesting must be >= -1 (with -1 indicating no containers).")
     include_cid = _options["include_cid"]
     data_generators: List[Iterator[Any]] = [
-        rand_list(max_nesting=max_nesting) if max_nesting >= 0 else iter([]),
-        rand_dict(max_nesting=max_nesting) if max_nesting >= 0 else iter([]),
-        rand_int(),
-        rand_bytes(),
-        rand_str(),
-        rand_bool_none(),
-        rand_float(),
-        rand_cid()
+        _rand_list(max_nesting=max_nesting) if max_nesting >= 0 else iter([]),
+        _rand_dict(max_nesting=max_nesting) if max_nesting >= 0 else iter([]),
+        _rand_int(),
+        _rand_bytes(),
+        _rand_str(),
+        _rand_bool_none(),
+        _rand_float(),
+        _rand_cid()
     ]
     num_data_generators = len(data_generators) if include_cid else len(data_generators)-1
     i = 0
@@ -383,6 +396,12 @@ def rand_list(n: Optional[int] = None, *, length: Optional[int] = None, max_nest
         - integer values > 0 mean that containers will be generated as items, with maximum nesting level `max_nesting-1`
         - no other values are valid
     """
+    validate(n, Optional[int])
+    validate(length, Optional[int])
+    validate(max_nesting, Optional[int])
+    return _rand_list(n, length=length, max_nesting=max_nesting)
+
+def _rand_list(n: Optional[int] = None, *, length: Optional[int] = None, max_nesting: Optional[int] = None) -> Iterator[List[Any]]:
     if n is not None and n < 0:
         raise ValueError()
     if length is not None and length < 0:
@@ -396,7 +415,7 @@ def rand_list(n: Optional[int] = None, *, length: Optional[int] = None, max_nest
     i = 0
     while n is None or i < n:
         _length = length if length is not None else _rand.randint(min_len, max_len)
-        yield list(rand_data(_length, max_nesting=max_nesting-1))
+        yield list(_rand_data(_length, max_nesting=max_nesting-1))
         i += 1
 
 def rand_dict(n: Optional[int] = None, *, length: Optional[int] = None, max_nesting: Optional[int] = None) -> Iterator[Dict[str, Any]]:
@@ -414,6 +433,12 @@ def rand_dict(n: Optional[int] = None, *, length: Optional[int] = None, max_nest
         - integer values > 0 mean that containers will be generated as values, with maximum nesting level `max_nesting-1`
         - no other values are valid
     """
+    validate(n, Optional[int])
+    validate(length, Optional[int])
+    validate(max_nesting, Optional[int])
+    return _rand_dict(n, length=length, max_nesting=max_nesting)
+
+def _rand_dict(n: Optional[int] = None, *, length: Optional[int] = None, max_nesting: Optional[int] = None) -> Iterator[Dict[str, Any]]:
     # pylint: disable = too-many-locals, too-many-branches
     if n is not None and n < 0:
         raise ValueError()
@@ -447,7 +472,7 @@ def rand_dict(n: Optional[int] = None, *, length: Optional[int] = None, max_nest
         else:
             keys = []
             keys_set = set()
-            str_generator = rand_str()
+            str_generator = _rand_str()
             while len(keys) < _length:
                 try:
                     s = next(str_generator)
@@ -457,7 +482,7 @@ def rand_dict(n: Optional[int] = None, *, length: Optional[int] = None, max_nest
                     keys.append(s)
                     keys_set.add(s)
         # generate dictionary
-        raw_dict = dict(zip(keys, rand_data(_length, max_nesting=max_nesting-1)))
+        raw_dict = dict(zip(keys, _rand_data(_length, max_nesting=max_nesting-1)))
         if canonical:
             yield _canonical_order_dict(raw_dict)
         else:
@@ -469,6 +494,10 @@ def rand_int(n: Optional[int] = None) -> Iterator[int]:
         Generates a stream of random `int` data.
         If a number `n` is given, that number of samples is yelded.
     """
+    validate(n, Optional[int])
+    return _rand_int(n)
+
+def _rand_int(n: Optional[int] = None) -> Iterator[int]:
     if n is not None and n < 0:
         raise ValueError()
     min_int = _options["min_int"]
@@ -485,6 +514,11 @@ def rand_bytes(n: Optional[int] = None, *, length: Optional[int] = None) -> Iter
 
         The optional `length` keyword argument can be used to fix the number of bytes generated.
     """
+    validate(n, Optional[int])
+    validate(length, Optional[int])
+    return _rand_bytes(n, length=length)
+
+def _rand_bytes(n: Optional[int] = None, *, length: Optional[int] = None) -> Iterator[bytes]:
     if n is not None and n < 0:
         raise ValueError()
     if length is not None and length < 0:
@@ -504,6 +538,11 @@ def rand_str(n: Optional[int] = None, *, length: Optional[int] = None) -> Iterat
 
         The optional `length` keyword argument can be used to fix the number of characters generated.
     """
+    validate(n, Optional[int])
+    validate(length, Optional[int])
+    return _rand_str(n, length=length)
+
+def _rand_str(n: Optional[int] = None, *, length: Optional[int] = None) -> Iterator[str]:
     if n is not None and n < 0:
         raise ValueError()
     if length is not None and length < 0:
@@ -529,6 +568,10 @@ def rand_bool(n: Optional[int] = None) -> Iterator[bool]:
         Generates a stream of random `bool` data.
         If a number `n` is given, that number of samples is yelded.
     """
+    validate(n, Optional[int])
+    return _rand_bool(n)
+
+def _rand_bool(n: Optional[int] = None) -> Iterator[bool]:
     if n is not None and n < 0:
         raise ValueError()
     i = 0
@@ -542,6 +585,10 @@ def rand_bool_none(n: Optional[int] = None) -> Iterator[Optional[bool]]:
         Generates a stream of random `Optional[bool]` data.
         If a number `n` is given, that number of samples is yelded.
     """
+    validate(n, Optional[int])
+    return _rand_bool_none(n)
+
+def _rand_bool_none(n: Optional[int] = None) -> Iterator[Optional[bool]]:
     if n is not None and n < 0:
         raise ValueError()
     i = 0
@@ -555,6 +602,10 @@ def rand_float(n: Optional[int] = None) -> Iterator[float]:
         Generates a stream of random `float` data.
         If a number `n` is given, that number of samples is yelded.
     """
+    validate(n, Optional[int])
+    return _rand_float(n)
+
+def _rand_float(n: Optional[int] = None) -> Iterator[float]:
     if n is not None and n < 0:
         raise ValueError()
     min_float = _options["min_float"]
@@ -592,14 +643,18 @@ def rand_cid(n: Optional[int] = None) -> Iterator[None]:
 
         If a number `n` is given, that number of samples is yelded.
     """
+    validate(n, Optional[int])
+    return _rand_cid(n)
+
+def _rand_cid(n: Optional[int] = None) -> Iterator[None]:
     if n is not None and n < 0:
         raise ValueError()
     hashfun = sha3_512
     hashcode = "sha3-512"
     hashlen = 0x40
     cid_version = 1
     cid_codec = "dag-cbor"
-    bytes_generator = rand_bytes(length=hashlen)
+    bytes_generator = _rand_bytes(length=hashlen)
     i = 0
     while n is None or i < n:
         try:

dag_cbor/utils.py

@@ -14,6 +14,7 @@
 """
 
 from typing import Any, Dict
+from typing_validation import validate
 
 class CBORError(Exception):
     """
@@ -71,6 +72,7 @@ def _check_key_compliance(value: Dict[str, Any]) -> None:
 
 def check_key_compliance(value: Dict[str, Any]) -> None:
     """ Check keys for DAG-CBOR compliance. """
+    validate(value, Dict[str, Any])
     _check_key_compliance(value)
 
 def canonical_order_dict(value: Dict[str, Any]) -> Dict[str, Any]:
@@ -79,6 +81,7 @@ def canonical_order_dict(value: Dict[str, Any]) -> Dict[str, Any]:
         Specifically, keys are sorted increasingly by the lexicographic ordering of the corresponding
         UTF-8 bytestrings.
     """
+    validate(value, Dict[str, Any])
     _check_key_compliance(value)
     # sort keys canonically
     return _canonical_order_dict(value)

docs/dag_cbor/decoding.html

@@ -72,7 +72,9 @@ <h1 class="title">Module <code>dag_cbor.decoding</code></h1>
 from io import BufferedIOBase, BytesIO
 import math
 import struct
-from typing import Callable, Optional, Tuple, Union
+from typing import Any, Dict, Callable, List, Optional, Tuple, Union
+from typing_validation import validate
+
 
 import cid # type: ignore
 
@@ -138,6 +140,9 @@ <h1 class="title">Module <code>dag_cbor.decoding</code></h1>
             [0, 1]
         ```
     &#34;&#34;&#34;
+    validate(stream_or_bytes, Union[BufferedIOBase, bytes])
+    validate(allow_concat, bool)
+    # validate(callback, Optional[DecodeCallback]) # not yet supported by typing_validation
     if isinstance(stream_or_bytes, bytes):
         stream: BufferedIOBase = BytesIO(stream_or_bytes)
     else:
@@ -249,8 +254,8 @@ <h1 class="title">Module <code>dag_cbor.decoding</code></h1>
     return (res.decode(encoding=&#34;utf-8&#34;, errors=&#34;strict&#34;), length)
 
 def _decode_list(stream: BufferedIOBase, length: int, *,
-                 callback: Optional[DecodeCallback]) -&gt; Tuple[list, int]:
-    l: list = []
+                 callback: Optional[DecodeCallback]) -&gt; Tuple[List[Any], int]:
+    l: List[Any] = []
     for i in range(length):
         try:
             item, _ = _decode_item(stream, callback=callback)
@@ -260,8 +265,8 @@ <h1 class="title">Module <code>dag_cbor.decoding</code></h1>
     return (l, 0)
 
 def _decode_dict(stream: BufferedIOBase, length: int,
-                 callback: Optional[DecodeCallback]) -&gt; Tuple[dict, int]:
-    d: dict = {}
+                 callback: Optional[DecodeCallback]) -&gt; Tuple[Dict[str, Any], int]:
+    d: Dict[str, Any] = {}
     for i in range(length):
         try:
             k, _ = _decode_item(stream, callback=callback)
@@ -411,6 +416,9 @@ <h2 class="section-title" id="header-functions">Functions</h2>
             [0, 1]
         ```
     &#34;&#34;&#34;
+    validate(stream_or_bytes, Union[BufferedIOBase, bytes])
+    validate(allow_concat, bool)
+    # validate(callback, Optional[DecodeCallback]) # not yet supported by typing_validation
     if isinstance(stream_or_bytes, bytes):
         stream: BufferedIOBase = BytesIO(stream_or_bytes)
     else:

docs/dag_cbor/encoding.html

@@ -72,7 +72,8 @@ <h1 class="title">Module <code>dag_cbor.encoding</code></h1>
 from io import BufferedIOBase, BytesIO
 import math
 import struct
-from typing import Optional, overload, Union
+from typing import Any, Dict, List, Optional, overload, Union
+from typing_validation import validate
 
 import cid # type: ignore
 
@@ -114,6 +115,8 @@ <h1 class="title">Module <code>dag_cbor.encoding</code></h1>
                 ...
         ```
     &#34;&#34;&#34;
+    validate(data, EncodableType)
+    validate(stream, Optional[BufferedIOBase])
     if stream is None:
         internal_stream = BytesIO()
         _encode(internal_stream, data)
@@ -197,13 +200,13 @@ <h1 class="title">Module <code>dag_cbor.encoding</code></h1>
     stream.write(utf8_value)
     return num_head_bytes+len(utf8_value)
 
-def _encode_list(stream: BufferedIOBase, value: list) -&gt; int:
+def _encode_list(stream: BufferedIOBase, value: List[Any]) -&gt; int:
     num_bytes_written = _encode_head(stream, 0x4, len(value))
     for item in value:
         num_bytes_written += _encode(stream, item)
     return num_bytes_written
 
-def _encode_dict(stream: BufferedIOBase, value: dict) -&gt; int:
+def _encode_dict(stream: BufferedIOBase, value: Dict[str, Any]) -&gt; int:
     _check_key_compliance(value)
     # sort keys canonically
     try:
@@ -297,6 +300,8 @@ <h2 class="section-title" id="header-functions">Functions</h2>
                 ...
         ```
     &#34;&#34;&#34;
+    validate(data, EncodableType)
+    validate(stream, Optional[BufferedIOBase])
     if stream is None:
         internal_stream = BytesIO()
         _encode(internal_stream, data)

docs/dag_cbor/index.html

@@ -99,9 +99,11 @@ <h1 class="title">Package <code>dag_cbor</code></h1>
     data, which is not natively handled by `cbor2`).
 &#34;&#34;&#34;
 
-from .encoding import encode
-from .decoding import decode
-from . import random</code></pre>
+__version__ = &#34;0.1.0&#34;
+
+from .encoding import encode as encode
+from .decoding import decode as decode
+from . import random as random</code></pre>
 </details>
 </section>
 <section>