Add mypy type checking to CI + upgrade to Python 3.11 in CI (#2123)

* ci

* add types-requests dep

* pin boto3

* pin requests

* pin boto3 for now

* pin httpcore

* pin more to try

* try pip upgrade

* add constraints back

* upgrade to python 3.11

* cleanup
;

* upgrade semaphore builds to use 3.11

* one new type error

* whitespace

* Resolving setup issues

* Resolved all type hinting issues locally

* Applied unasync

* Resolved mypy failures

* Flake8 fix plus print for clang-format version

* Reverting admin.c to clang-format 18 variant

* Resolved new avro type checks

* Added mypy ignore for celpy calls

* Upped memory limit as mypy increased the overhead slightly

* Bumped memory for mypy overhead

* Adding file missed in prior commit

---------

Co-authored-by: Matthew Seal <mseal@confluent.io>

Author: fangnx

.semaphore/semaphore.yml

@@ -197,14 +197,14 @@ blocks:
       jobs:
         - name: Build and Tests with 'classic' group protocol
           commands:
-            - sem-version python 3.9
+            - sem-version python 3.11
             # use a virtualenv
             - python3 -m venv _venv && source _venv/bin/activate
             - chmod u+r+x tools/source-package-verification.sh
             - tools/source-package-verification.sh
         - name: Build and Tests with 'consumer' group protocol
           commands:
-            - sem-version python 3.9
+            - sem-version python 3.11
             - sem-version java 17
             # use a virtualenv
             - python3 -m venv _venv && source _venv/bin/activate
@@ -213,7 +213,7 @@ blocks:
             - tools/source-package-verification.sh
         - name: Build, Test, and Report coverage
           commands:
-            - sem-version python 3.9
+            - sem-version python 3.11
             # use a virtualenv
             - python3 -m venv _venv && source _venv/bin/activate
             - chmod u+r+x tools/source-package-verification.sh
@@ -237,7 +237,7 @@ blocks:
       jobs:
         - name: Build
           commands:
-            - sem-version python 3.9
+            - sem-version python 3.11
             # use a virtualenv
             - python3 -m venv _venv && source _venv/bin/activate
             - chmod u+r+x tools/source-package-verification.sh
@@ -256,7 +256,7 @@ blocks:
       jobs:
         - name: Build
           commands:
-            - sem-version python 3.9
+            - sem-version python 3.11
             # use a virtualenv
             - python3 -m venv _venv && source _venv/bin/activate
             - chmod u+r+x tools/source-package-verification.sh
@@ -275,7 +275,7 @@ blocks:
       jobs:
         - name: Build
           commands:
-            - sem-version python 3.9
+            - sem-version python 3.11
             # use a virtualenv
             - python3 -m venv _venv && source _venv/bin/activate
             - chmod u+r+x tools/source-package-verification.sh
@@ -302,7 +302,7 @@ blocks:
         - name: Build and Tests
           commands:
             # Setup Python environment
-            - sem-version python 3.9
+            - sem-version python 3.11
             - python3 -m venv _venv && source _venv/bin/activate
 
             # Install ducktape framework and additional dependencies

DEVELOPER.md

@@ -212,8 +212,11 @@ tox -e black,isort
 # Check linting
 tox -e flake8
 
+# Check typing
+tox -e mypy
+
 # Run all formatting and linting checks
-tox -e black,isort,flake8
+tox -e black,isort,flake8,mypy
 ```
 
 ## Documentation build

MANIFEST.in

@@ -1,5 +1,7 @@
 include README.md
 include LICENSE
 include src/confluent_kafka/src/*.[ch]
+include src/confluent_kafka/py.typed
+include src/confluent_kafka/cimpl.pyi
 prune tests
-prune docs
+prune docs

Makefile

@@ -2,6 +2,9 @@ all:
 	@echo "Targets:"
 	@echo " clean"
 	@echo " docs"
+	@echo " mypy"
+	@echo " style-check"
+	@echo " style-fix"
 
 
 clean:
@@ -14,6 +17,9 @@ clean:
 docs:
 	$(MAKE) -C docs html
 
+mypy:
+	python3 -m mypy src/confluent_kafka
+
 style-check:
 	@(tools/style-format.sh \
 		$$(git ls-tree -r --name-only HEAD | egrep '\.(c|h|py)$$') )

requirements/requirements-avro.txt

@@ -1,4 +1,4 @@
 fastavro < 1.8.0; python_version == "3.7"
 fastavro < 2; python_version > "3.7"
 requests
-avro>=1.11.1,<2
+avro>=1.11.1,<2

requirements/requirements-tests.txt

@@ -2,7 +2,9 @@
 urllib3<3
 flake8
 mypy
+attrs
 types-cachetools
+types-requests
 orjson
 pytest
 pytest-timeout

src/confluent_kafka/deserializing_consumer.py

@@ -105,7 +105,11 @@ def poll(self, timeout: float = -1) -> Optional[Message]:
         if error is not None:
             raise ConsumeError(error, kafka_message=msg)
 
-        ctx = SerializationContext(msg.topic(), MessageField.VALUE, msg.headers())
+        topic = msg.topic()
+        if topic is None:
+            raise TypeError("Message topic is None")
+        ctx = SerializationContext(topic, MessageField.VALUE, msg.headers())
+
         value = msg.value()
         if self._value_deserializer is not None:
             try:

src/confluent_kafka/schema_registry/common/avro.py

@@ -5,7 +5,7 @@
 from collections import defaultdict
 from copy import deepcopy
 from io import BytesIO
-from typing import Dict, Optional, Set, Tuple, Union
+from typing import Dict, Optional, Set, Tuple, Union, cast
 
 from fastavro import repository, validate
 from fastavro.schema import load_schema
@@ -217,11 +217,17 @@ def _resolve_union(schema: AvroSchema, message: AvroMessage) -> Tuple[Optional[A
     for subschema in schema:
         try:
             if is_wrapped_union:
-                if isinstance(subschema, dict) and subschema["name"] == message[0]:
-                    return (subschema, message[1])
+                if isinstance(subschema, dict):
+                    dict_schema = cast(dict, subschema)
+                    tuple_message = cast(tuple, message)
+                    if dict_schema["name"] == tuple_message[0]:
+                        return (dict_schema, tuple_message[1])
             elif is_typed_union:
-                if isinstance(subschema, dict) and subschema["name"] == message['-type']:
-                    return (subschema, message)
+                if isinstance(subschema, dict):
+                    dict_schema = cast(dict, subschema)
+                    dict_message = cast(dict, message)
+                    if dict_schema["name"] == dict_message['-type']:
+                        return (dict_schema, dict_message)
             else:
                 validate(message, subschema)
                 return (subschema, message)

src/confluent_kafka/schema_registry/rules/cel/string_format.py

@@ -44,11 +44,11 @@ def __init__(self, locale: str):
     def format(self, fmt: celtypes.Value, args: celtypes.Value) -> celpy.Result:
         if not isinstance(fmt, celtypes.StringType):
             return celpy.native_to_cel(  # type: ignore[attr-defined]
-                celpy.new_error("format() requires a string as the first argument")
+                celpy.new_error("format() requires a string as the first argument")  # type: ignore[attr-defined]
             )  # type: ignore[attr-defined]
         if not isinstance(args, celtypes.ListType):
             return celpy.native_to_cel(  # type: ignore[attr-defined]
-                celpy.new_error("format() requires a list as the second argument")
+                celpy.new_error("format() requires a list as the second argument")  # type: ignore[attr-defined]
             )  # type: ignore[attr-defined]
         # printf style formatting
         i = 0

src/confluent_kafka/schema_registry/rules/encryption/encrypt_executor.py

@@ -365,7 +365,7 @@ def _is_expired(self, ctx: RuleContext, dek: Optional[Dek]) -> bool:
             ctx.rule_mode != RuleMode.READ
             and self._dek_expiry_days > 0
             and dek is not None
-            and (now - dek.ts) / MILLIS_IN_DAY > self._dek_expiry_days
+            and (now - (dek.ts or 0)) / MILLIS_IN_DAY > self._dek_expiry_days
         )  # type: ignore[operator]
 
     def transform(self, ctx: RuleContext, field_type: FieldType, field_value: Any) -> Any: