type checking & co

.github/workflows/ci.yml

@@ -13,7 +13,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python-version: ["3.8", "3.9", "3.10", "3.11", "3.12", "3.13", "3.14"]
+        python-version: ["3.9", "3.10", "3.11", "3.12", "3.13", "3.14"]
 
     steps:
       - uses: actions/checkout@v6
@@ -26,7 +26,7 @@ jobs:
       - name: Install dependencies
         run: |
           python -m pip install --upgrade pip
-          pip install -Ue .[test]
+          pip install -Ue . --group test
           pip install -Ue .
 
       - name: Run tests

.pre-commit-config.yaml

@@ -25,7 +25,7 @@ repos:
       - id: rst-directive-colons
       - id: rst-inline-touching-normal
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.15.21
+    rev: v0.16.0
     hooks:
       - id: ruff-check
         alias: "ruff"
@@ -54,8 +54,13 @@ repos:
       - id: conventional-pre-commit
         stages: [commit-msg]
         args: ["--strict"]
+  - repo: https://github.com/astral-sh/ty-pre-commit
+    rev: v0.0.63
+    hooks:
+      - id: ty
+        stages: [manual]
   - repo: https://github.com/pre-commit/mirrors-mypy
-    rev: v2.2.0
+    rev: v2.3.0
     hooks:
       - id: mypy
         stages: [manual]

Makefile

@@ -1,13 +1,30 @@
 #!/usr/bin/make -f
 
+.PHONY: dist pushrelease pushreleasetest clean lint lint-fix format
+
 dist:
-    python3 setup.py sdist bdist_wheel
+	python3 setup.py sdist bdist_wheel
 
 pushreleasetest:
-    python3 -m twine upload --repository testpypi dist/*
+	python3 -m twine upload --repository testpypi dist/*
 
 pushrelease:
-    python3 -m twine upload dist/*
+	python3 -m twine upload dist/*
 
 clean:
-    $(RM) -r dist
+	$(RM) -r dist
+
+lint:
+	-prek run -a
+
+lint-fix:
+	-prek run -a --hook-stage manual ruff-fix
+
+format:
+	-prek run -a --hook-stage manual ruff-format
+
+ty:
+	-prek run -a --hook-stage manual ty
+
+mypy:
+	-prek run -a --hook-stage manual mypy

httoop/__main__.py

@@ -9,7 +9,8 @@
 
 import pathlib
 import sys
-from argparse import ArgumentParser, FileType
+from argparse import ArgumentParser
+from typing import IO
 
 from httoop import Request, Response, __name__ as name, __version__ as version
 from httoop.client import ClientStateMachine
@@ -51,45 +52,49 @@ def add_subparsers(self) -> None:
         request = parse_message_subparsers.add_parser('request', parents=[self.parent_parser])
         request.set_defaults(func=self.parse_request)
         add = request.add_argument
-        add('--file', default='-', type=FileType('rb'))
+        add('--file', default='-')
         add('--scheme', default='http')
         add('--host', default='www.example.net')
         add('--port', default=80, type=int)
 
         response = parse_message_subparsers.add_parser('response', parents=[self.parent_parser])
         add = response.add_argument
         response.set_defaults(func=self.parse_response)
-        add('--file', default='-', type=FileType('rb'))
+        add('--file', default='-')
 
     def parse_arguments(self) -> None:
         self.arguments = self.parser.parse_args()
-
-        if self.arguments.action == 'parse' and hasattr(self.arguments.file, 'buffer'):
-            # https://bugs.python.org/issue14156
-            self.arguments.file = self.arguments.file.buffer
         self.arguments.func()
 
-    def add_common_arguments(self, add) -> None:
+    @classmethod
+    def add_common_arguments(cls, add) -> None:
         add('--protocol')
         add('-H', '--header', action='append', default=[])
         add('-b', '--body', default='')
 
+    @classmethod
+    def get_file(cls, file: str) -> IO:
+        if file == '-':
+            return sys.stdin.buffer
+        return pathlib.Path(file).open('rb')
+
     def parse_request(self) -> None:
         server = ServerStateMachine(self.arguments.scheme, self.arguments.host, self.arguments.port)
-        for _request, response in server.parse(self.arguments.file.read()):
+        for _request, response in server.parse(self.get_file(self.arguments.file).read()):
             print(repr(response))
             print(repr(response.headers))
             print(repr(response.body))
 
     def parse_response(self) -> None:
         client = ClientStateMachine()
         client.request = Request()
-        for response in client.parse(self.arguments.file.read()):
+        for response in client.parse(self.get_file(self.arguments.file).read()):
             print(repr(response))
             print(repr(response.headers))
             print(repr(response.body))
             print(repr(bytes(response.body)))
         if client.buffer:
+            assert client.message  # noqa: S101
             print('WARNING: response not yet complete!:')
             print(repr(client.message))
             print(repr(client.message.headers))
@@ -115,10 +120,11 @@ def compose_response(self) -> None:
         self.common()
 
     def common(self) -> None:
+        assert self.message is not None  # noqa: S101
         if self.arguments.protocol:
             protocol = self.arguments.protocol
             try:
-                protocol = [int(x) for x in protocol.split('.', 1)]
+                protocol = tuple(int(x) for x in protocol.split('.', 1))
             except ValueError:
                 pass
             else:
@@ -133,14 +139,15 @@ def common(self) -> None:
         if body == '-':
             body = sys.stdin.read()
         elif body.startswith('@'):
-            body = pathlib.Path(body[1:]).open('rb')
+            body = pathlib.Path(body[1:]).open('rb')  # noqa: SIM115
         self.message.body = body
 
         sys.stdout.write(self.decode(bytes(self.message)))
         sys.stdout.write(self.decode(bytes(self.message.headers)))
         sys.stdout.write(self.decode(bytes(self.message.body)))
 
-    def decode(self, data):
+    @classmethod
+    def decode(cls, data):
         if str is not bytes:
             data = data.decode('ISO8859-1')
         return data

httoop/authentication/__init__.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import re
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, ClassVar
 
 from httoop.authentication.basic import BasicAuthRequestScheme, BasicAuthResponseScheme
 from httoop.authentication.digest import DigestAuthRequestScheme, DigestAuthResponseScheme, SecureDigestAuthRequestScheme
@@ -16,7 +16,7 @@
 
 class AuthElement(HeaderElement):
 
-    schemes = {}
+    schemes: ClassVar = {}
     RE_SPACE_SPLIT = re.compile(rb'\s+(?=(?:[^"]*"[^"]*")*[^"]*$)')
 
     def sanitize(self) -> None:
@@ -31,29 +31,29 @@ def parseparams(cls, elementstr: bytes) -> tuple[bytes, dict[bytes, str]] | tupl
         try:
             scheme, authinfo = elementstr.split(b' ', 1)
         except ValueError:
-            raise InvalidHeader(_('Authorization headers must contain authentication scheme'))
+            raise InvalidHeader(_('Authorization headers must contain authentication scheme')) from None
         try:
             parser = cls.schemes[scheme.decode('ISO8859-1').lower()]
         except KeyError:
-            raise InvalidHeader(_('Unsupported authentication scheme: %r'), scheme)
+            raise InvalidHeader(_('Unsupported authentication scheme: %r'), scheme) from None
 
         try:
             authinfo = parser.parse(authinfo)
         except KeyError as key:
-            raise InvalidHeader(_('Missing parameter %r for authentication scheme %r'), str(key), scheme)
+            raise InvalidHeader(_('Missing parameter %r for authentication scheme %r'), str(key), scheme) from None
 
         return scheme.title(), authinfo
 
     def compose(self) -> bytes:
         try:
             scheme = self.schemes[self.value.lower()]
         except KeyError:
-            raise InvalidHeader(_('Unsupported authentication scheme: %r'), self.value)
+            raise InvalidHeader(_('Unsupported authentication scheme: %r'), self.value) from None
 
         try:
             authinfo = scheme.compose(self.params)
         except KeyError as key:
-            raise InvalidHeader(_('Missing parameter %r for authentication scheme %r'), str(key), self.value)
+            raise InvalidHeader(_('Missing parameter %r for authentication scheme %r'), str(key), self.value) from None
 
         return b'%s %s' % (self.value.encode('ASCII').title(), authinfo)
 
@@ -68,7 +68,7 @@ class AuthRequestElement(AuthElement):
 
     encoding = 'ASCII'
 
-    schemes = {
+    schemes: ClassVar = {
         'basic': BasicAuthRequestScheme,
         'digest': SecureDigestAuthRequestScheme,
     }
@@ -103,7 +103,7 @@ def password(self, password) -> None:
 
 class AuthResponseElement(AuthElement):
 
-    schemes = {
+    schemes: ClassVar = {
         'basic': BasicAuthResponseScheme,
         'digest': DigestAuthResponseScheme,
     }

httoop/authentication/basic.py

@@ -19,9 +19,9 @@ def parse(authinfo: bytes) -> dict[str, bytes]:
             if not username:
                 raise ValueError()
         except Base64Error:
-            raise InvalidHeader(_('Basic authentication contains invalid base64'))
+            raise InvalidHeader(_('Basic authentication contains invalid base64')) from None
         except ValueError:
-            raise InvalidHeader(_('No username:password provided'))
+            raise InvalidHeader(_('No username:password provided')) from None
 
         return {
             # 'username': username.decode('ISO8859-1'),
@@ -42,11 +42,11 @@ def compose(authinfo: ByteUnicodeDict) -> bytes:
 class BasicAuthResponseScheme:
 
     @staticmethod
-    def parse(authinfo: bytes) -> dict[bytes, str | bytes]:
+    def parse(authinfo: bytes) -> dict[bytes, str]:
         params = HeaderElement.parseparams(b'X;%s' % authinfo)[1]
-        params.setdefault(b'realm', b'')
+        params.setdefault(b'realm', '')
         return params
 
     @staticmethod
     def compose(authinfo: ByteUnicodeDict) -> bytes:
-        return HeaderElement.formatparam(b'realm', authinfo['realm'], True)
+        return HeaderElement.formatparam(b'realm', authinfo['realm'], quote=True)

httoop/authentication/digest.py

@@ -1,21 +1,22 @@
+# ruff: file-ignore[N806,N802]
 from __future__ import annotations
 
 from hashlib import md5, new, sha256
 from hmac import compare_digest
 from secrets import token_bytes
-from typing import Callable
+from typing import Callable, ClassVar
 
 from httoop.exceptions import InvalidHeader
 from httoop.header.element import HeaderElement
 from httoop.util import ByteUnicodeDict, _
 
 
 class DigestAuthScheme:
-    algorithms = {
-        'MD5': lambda: md5(),  # noqa: S324
-        'MD5-sess': lambda: md5(),  # noqa: S324
-        'SHA-256': lambda: sha256(),
-        'SHA-256-sess': lambda: sha256(),
+    algorithms: ClassVar = {
+        'MD5': md5,
+        'MD5-sess': md5,
+        'SHA-256': sha256,
+        'SHA-256-sess': sha256,
         'SHA-512-256': lambda: new('sha512_256'),
         'SHA-512-256-sess': lambda: new('sha512_256'),
     }  # not case insensitive per RFC

httoop/client/__init__.py

@@ -1,15 +1,17 @@
 from __future__ import annotations
 
 from httoop.exceptions import InvalidLine
-from httoop.messages import Response
+from httoop.messages import Request, Response
 from httoop.parser import NOT_RECEIVED_YET, StateMachine
 
 
 class ClientStateMachine(StateMachine):
 
     Message = Response
+    request: Request
+    message: Response
 
-    def __init__(self, *, strict: bool = True, max_status_line_length: int | float = 256) -> None:
+    def __init__(self, *, strict: bool = True, max_status_line_length: float = 256) -> None:
         super().__init__(strict=strict)
         self.max_status_line_length = max_status_line_length
 

httoop/codecs/__init__.py

@@ -18,7 +18,7 @@
 __all__ = ['CODECS', 'Codec', 'application', 'audio', 'example', 'image', 'message', 'model', 'multipart', 'text', 'video']
 
 
-def lookup(encoding: str, raise_errors: bool = True) -> Any:
+def lookup(encoding: str, *, raise_errors: bool = True) -> Any:
     type_ = '%s/*' % (encoding.split('/', 1)[0],)
     return CODECS.get(encoding) or CODECS.get(type_) or (raise_errors and CODECS[encoding]) or None
 

httoop/codecs/application/gzip.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import gzip
 import io
 import zlib
@@ -20,30 +22,30 @@ def encode(cls, data: bytes, charset: None = None, mimetype: None = None) -> byt
                 fd.write(Codec.encode(data, charset))
             return out.getvalue()
         except zlib.error:  # pragma: no cover
-            raise EncodeError(_('Invalid gzip data.'))
+            raise EncodeError(_('Invalid gzip data.')) from None
 
     @classmethod
-    def decode(cls, data: bytes, charset: None = None, mimetype: None = None) -> str:
+    def decode(cls, data: bytes, charset: str | None = None, mimetype: None = None) -> str:
         try:
             with gzip.GzipFile(fileobj=io.BytesIO(data)) as fd:
                 data = fd.read()
         except (zlib.error, OSError, EOFError):
-            raise DecodeError(_('Invalid gzip data.'))
+            raise DecodeError(_('Invalid gzip data.')) from None
         return Codec.decode(data, charset)
 
     @classmethod
     def iterencode(cls, data, charset=None, mimetype=None):
+        out = io.BytesIO()
         try:
-            out = io.BytesIO()
             with gzip.GzipFile(fileobj=out, mode='w', compresslevel=cls.compression_level) as fd:
                 for part in data:
                     fd.write(Codec.encode(part, charset))
                     yield out.getvalue()
                     out.seek(0)
                     out.truncate()
-            yield out.getvalue()
         except zlib.error:  # pragma: no cover
-            raise EncodeError(_('Invalid gzip data.'))
+            raise EncodeError(_('Invalid gzip data.')) from None
+        yield out.getvalue()
 
     @classmethod
     def iterdecode(cls, data, charset=None, mimetype=None):
@@ -62,4 +64,4 @@ def iterdecode(cls, data, charset=None, mimetype=None):
                 fd.seek(0)
                 yield Codec.decode(gzfd.read(), charset)
         except (zlib.error, OSError, EOFError):
-            raise DecodeError(_('Invalid gzip data.'))
+            raise DecodeError(_('Invalid gzip data.')) from None