refactor: initial type annotations

.github/workflows/test.yml

@@ -48,6 +48,9 @@ jobs:
       - name: "Install python dependencies"
         run: |
           pip install ".[ci]"
+      - name: "Run type checking"
+        run: |
+          mypy stream_zip.py
       - name: "Run tests"
         run: |
           pytest --cov

pyproject.toml

@@ -27,6 +27,9 @@ dev = [
     "pytest-cov>=3.0.0",
     "stream-unzip>=0.0.86",
     "pyzipper>=0.3.6",
+    # Type checking
+    "mypy>=0.971",
+    'types-contextvars>=2.4.7.3; python_version<"3.7"',
 ]
 ci = [
     "pycryptodome==3.10.1",
@@ -35,6 +38,9 @@ ci = [
     "pytest-cov==3.0.0",
     "stream-unzip==0.0.86",
     "pyzipper==0.3.6",
+    # Type checking
+    "mypy==0.971",
+    'types-contextvars==2.4.7.3; python_version<"3.7"',
 ]
 
 [project.urls]

stream_zip.py

@@ -3,6 +3,7 @@
 import asyncio
 import secrets
 import zlib
+from typing import Any, Iterable, Tuple, Optional, Deque
 
 from Crypto.Cipher import AES
 from Crypto.Hash import HMAC, SHA1
@@ -70,14 +71,14 @@ def method_compressobj(offset, default_get_compressobj):
     return method_compressobj
 
 
-def stream_zip(files, chunk_size=65536,
+def stream_zip(files: Iterable[Tuple[Any, Any, Any, Any, Any]], chunk_size: int=65536,
                get_compressobj=lambda: zlib.compressobj(wbits=-zlib.MAX_WBITS, level=9),
-               extended_timestamps=True,
-               password=None,
+               extended_timestamps: bool=True,
+               password: Optional[str]=None,
                get_crypto_random=lambda num_bytes: secrets.token_bytes(num_bytes),
-):
+) -> Iterable[bytes]:
 
-    def evenly_sized(chunks):
+    def evenly_sized(chunks: Iterable[bytes]) -> Iterable[bytes]:
         chunk = b''
         offset = 0
         it = iter(chunks)
@@ -104,7 +105,7 @@ def up_to(num):
                 break
             yield block
 
-    def get_zipped_chunks_uneven():
+    def get_zipped_chunks_uneven() -> Iterable[bytes]:
         local_header_signature = b'PK\x03\x04'
         local_header_struct = Struct('<HHH4sIIIHH')
 
@@ -140,7 +141,7 @@ def get_zipped_chunks_uneven():
         data_descriptor_flag = 0b0000000000001000
         utf8_flag = 0b0000100000000000
 
-        central_directory = deque()
+        central_directory: Deque[Tuple[bytes, bytes, bytes]] = deque()
         central_directory_size = 0
         central_directory_start_offset = 0
         zip_64_central_directory = False