Cover azul.chalice with mypy

.mypy.ini

@@ -10,7 +10,8 @@ modules =
     azul.auth,
     azul.azulclient,
     azul.bytes,
-    azul.caching
+    azul.caching,
+    azul.chalice
 packages =
     azul.openapi
 

requirements.dev.txt

@@ -5,7 +5,7 @@ coverage==7.6.9
 docker==7.1.0
 flake8==7.1.1
 gevent==24.11.1
-git+https://github.com/DataBiosphere/[email protected]+15#egg=chalice
+git+https://github.com/DataBiosphere/[email protected]+16#egg=chalice
 git+https://github.com/hannes-ucsc/[email protected]+1#egg=requirements-parser
 gitpython==3.1.43
 google-api-python-client==2.156.0
@@ -24,6 +24,7 @@ python-gitlab==5.2.0
 pyyaml==6.0.2
 responses==0.25.3
 strict-rfc3339==0.7
+types-chevron==0.14.0  # match with chevron in requirements.txt
 types-urllib3==1.26.20  # match with urllib in requirements.txt
 watchdog==6.0.0
 -r requirements.dev.trans.txt

requirements.txt

@@ -3,7 +3,7 @@ aws-requests-auth==0.4.3
 bdbag==1.7.3
 boto3==1.35.84  # Match version of the `boto3-stubs` dev dependency
 botocore==1.35.84
-chevron==0.14.0
+chevron==0.14.0  # Match with types-chevron in requirements.dev.txt
 deprecated==1.2.15
 elasticsearch==7.17.12  # 7.x to match server-side
 elasticsearch-dsl==7.4.1  # 7.x to match server-side

src/azul/chalice.py

@@ -17,6 +17,7 @@
 import pathlib
 from typing import (
     Any,
+    Callable,
     Iterator,
     Literal,
     Self,
@@ -81,6 +82,9 @@
 from azul.types import (
     JSON,
     LambdaContext,
+    json_dict,
+    json_list,
+    json_str,
 )
 
 log = logging.getLogger(__name__)
@@ -254,17 +258,17 @@ def _http_cache_for(self, seconds: int):
 
     HttpMethod = Literal['GET', 'POST', 'PUT', 'PATCH', 'HEAD', 'OPTIONS', 'DELETE']
 
-    # noinspection PyMethodOverriding
-    def route(self,
-              path: str,
-              *,
-              methods: Sequence[HttpMethod] = ('GET',),
-              enabled: bool = True,
-              interactive: bool = True,
-              cache_control: str = 'no-store',
-              path_spec: JSON | None = None,
-              spec: JSON,
-              **kwargs):
+    def route[C: Callable](self,
+                           path: str,
+                           *,
+                           methods: Sequence[HttpMethod] = ('GET',),
+                           enabled: bool = True,
+                           interactive: bool = True,
+                           cache_control: str = 'no-store',
+                           path_spec: JSON | None = None,
+                           spec: JSON | None = None,
+                           **kwargs
+                           ) -> Callable[[C], C]:
         """
         Decorates a view handler function in a Chalice application.
 
@@ -297,8 +301,14 @@ def route(self,
 
                      https://github.com/OAI/OpenAPI-Specification/blob/main/versions/3.0.3.md#operation-object
 
-                     This must be specified for every `@app.route` invocation.
+                     Even though this keyword argument has a default value, it
+                     must be specified for every `@app.route` invocation. The
+                     reason for the default is so that the signature of the
+                     override is compatible with that of the overridden method,
+                     a mypy requirement.
         """
+        require(spec is not None, "Argument 'spec' is required")
+        assert spec is not None
         if enabled:
             if not interactive:
                 require(bool(methods), 'Must list methods with interactive=False')
@@ -328,17 +338,17 @@ def spec(self) -> JSON:
         Only call this method after all routes are registered.
         """
         used_tags = set(
-            tag
-            for path in self._specs['paths'].values()
-            for method in path.values() if isinstance(method, dict)
-            for tag in method.get('tags', [])
+            json_str(tag)
+            for path in json_dict(self._specs['paths']).values()
+            for method in json_dict(path).values() if isinstance(method, dict)
+            for tag in json_list(method.get('tags', []))
         )
         reject('servers' in self._specs, "The 'servers' entry is computed")
         return {
             **self._specs,
             'tags': [
-                tag for tag in self._specs.get('tags', [])
-                if tag['name'] in used_tags
+                tag for tag in json_list(self._specs.get('tags', []))
+                if json_dict(tag)['name'] in used_tags
             ],
             'servers': [{'url': str(self.base_url.add(path='/'))}]
         }
@@ -349,7 +359,9 @@ def self_url(self) -> mutable_furl:
         The URL of the current request, including the path, but without query
         arguments. Callers can safely modify the returned `furl` instance.
         """
-        path = self.current_request.context['path']
+        request = self.current_request
+        assert request is not None
+        path = request.context['path']
         return self.base_url.add(path=path)
 
     @property
@@ -372,7 +384,9 @@ def base_url(self) -> mutable_furl:
                 from chalice.constants import (
                     DEFAULT_HANDLER_NAME,
                 )
-                assert self.lambda_context.function_name == DEFAULT_HANDLER_NAME
+                lambda_context = self.lambda_context
+                assert lambda_context is not None
+                assert lambda_context.function_name == DEFAULT_HANDLER_NAME
                 scheme = 'http'
             else:
                 # Invocation via API Gateway
@@ -395,20 +409,20 @@ def _register_spec(self,
         """
         Add a route's specifications to the specification object.
         """
+        paths = json_dict(self._specs['paths'])
         if path_spec is not None:
-            reject(path in self._specs['paths'],
+            reject(path in paths,
                    'Only specify path_spec once per route path')
-            self._specs['paths'][path] = copy_json(path_spec)
+            paths[path] = copy_json(path_spec)
 
         for method in methods:
             # OpenAPI requires HTTP method names be lower case
             method = method.lower()
             # This may override duplicate specs from path_specs
-            if path not in self._specs['paths']:
-                self._specs['paths'][path] = {}
-            reject(method in self._specs['paths'][path],
+            path_methods = json_dict(paths.setdefault(path, {}))
+            reject(method in path_methods,
                    "Only specify 'spec' once per route path and method")
-            self._specs['paths'][path][method] = copy_json(spec)
+            path_methods[method] = copy_json(spec)
 
     class _LogJSONEncoder(JSONEncoder):
 
@@ -565,6 +579,7 @@ def bind(self, app: Chalice, handler_name: str | None = None) -> Self:
 
         @property
         def tf_function_resource_name(self) -> str:
+            assert self.handler_name is not None, 'Unbound decorator'
             if self.handler_name is None:
                 return self.app_name
             else:
@@ -879,8 +894,10 @@ class AppController:
 
     @property
     def lambda_context(self) -> LambdaContext:
+        assert self.app.lambda_context is not None
         return self.app.lambda_context
 
     @property
     def current_request(self) -> AzulRequest:
+        assert self.app.current_request is not None
         return self.app.current_request

src/azul/types.py

@@ -44,44 +44,73 @@
 type MutableFlatJSON = dict[str, PrimitiveJSON]
 
 
-class LambdaContext(object):
+def json_mapping(v: AnyJSON) -> JSON:
+    assert isinstance(v, Mapping)
+    return v
+
+
+def json_sequence(v: AnyJSON) -> JSONArray:
+    assert isinstance(v, Sequence)
+    return v
+
+
+def json_dict(v: AnyMutableJSON) -> MutableJSON:
+    assert isinstance(v, dict)
+    return v
+
+
+def json_list(v: AnyMutableJSON) -> MutableJSONArray:
+    assert isinstance(v, list)
+    return v
+
+
+def json_str(v: AnyMutableJSON | AnyJSON) -> str:
+    assert isinstance(v, str)
+    return v
+
+
+def json_int(v: AnyMutableJSON | AnyJSON) -> int:
+    assert isinstance(v, int)
+    return v
+
+
+def json_float(v: AnyMutableJSON | AnyJSON) -> float:
+    assert isinstance(v, float)
+    return v
+
+
+def json_bool(v: AnyMutableJSON | AnyJSON) -> bool:
+    assert isinstance(v, bool)
+    return v
+
+
+def json_none(v: AnyMutableJSON | AnyJSON) -> None:
+    assert v is None
+    return v
+
+
+class LambdaContext:
     """
     A stub for the AWS Lambda context
     """
 
-    @property
-    def aws_request_id(self) -> str:
-        raise NotImplementedError
+    aws_request_id: str
 
-    @property
-    def log_group_name(self) -> str:
-        raise NotImplementedError
+    log_group_name: str
 
-    @property
-    def log_stream_name(self) -> str:
-        raise NotImplementedError
+    log_stream_name: str
 
-    @property
-    def function_name(self) -> str:
-        raise NotImplementedError
+    function_name: str
 
-    @property
-    def memory_limit_in_mb(self) -> str:
-        raise NotImplementedError
+    memory_limit_in_mb: str
 
-    @property
-    def function_version(self) -> str:
-        raise NotImplementedError
+    function_version: str
 
-    @property
-    def invoked_function_arn(self) -> str:
-        raise NotImplementedError
+    invoked_function_arn: str
 
-    def get_remaining_time_in_millis(self) -> int:
-        raise NotImplementedError
+    def get_remaining_time_in_millis(self) -> int: ...  # type: ignore[empty-body]
 
-    def log(self, msg: str) -> None:
-        raise NotImplementedError
+    def log(self, msg: str) -> None: ...
 
 
 def is_optional(t) -> bool: