docs(examples): enforce and fix all mypy errors

Makefile

@@ -99,4 +99,4 @@ changelog:
 	docker run -v "${PWD}":/workdir quay.io/git-chglog/git-chglog > CHANGELOG.md
 
 mypy:
-	poetry run mypy --pretty aws_lambda_powertools
+	poetry run mypy --pretty aws_lambda_powertools examples

aws_lambda_powertools/event_handler/__init__.py

@@ -2,7 +2,14 @@
 Event handler decorators for common Lambda events
 """
 
-from .api_gateway import ALBResolver, APIGatewayHttpResolver, ApiGatewayResolver, APIGatewayRestResolver, CORSConfig, Response
+from .api_gateway import (
+    ALBResolver,
+    APIGatewayHttpResolver,
+    ApiGatewayResolver,
+    APIGatewayRestResolver,
+    CORSConfig,
+    Response,
+)
 from .appsync import AppSyncResolver
 
 __all__ = [

aws_lambda_powertools/event_handler/appsync.py

@@ -142,6 +142,7 @@ def lambda_handler(event, context):
         ValueError
             If we could not find a field resolver
         """
+        # Maintenance: revisit generics/overload to fix [attr-defined] in mypy usage
         BaseRouter.current_event = data_model(event)
         BaseRouter.lambda_context = context
         resolver = self._get_resolver(BaseRouter.current_event.type_name, BaseRouter.current_event.field_name)

aws_lambda_powertools/tracing/base.py

@@ -2,39 +2,34 @@
 import numbers
 import traceback
 from contextlib import contextmanager
-from typing import Any, AsyncContextManager, ContextManager, List, NoReturn, Optional, Set, Union
+from typing import Any, Generator, List, NoReturn, Optional, Sequence, Union
 
 
-class BaseProvider(abc.ABC):
-    @abc.abstractmethod  # type: ignore
-    @contextmanager
-    def in_subsegment(self, name=None, **kwargs) -> ContextManager:
-        """Return a subsegment context manger.
+class BaseSegment(abc.ABC):
+    """Holds common properties and methods on segment and subsegment."""
+
+    @abc.abstractmethod
+    def close(self, end_time: Optional[int] = None):
+        """Close the trace entity by setting `end_time`
+        and flip the in progress flag to False.
 
         Parameters
         ----------
-        name: str
-            Subsegment name
-        kwargs: Optional[dict]
-            Optional parameters to be propagated to segment
+        end_time: int
+            Time in epoch seconds, by default current time will be used.
         """
 
-    @abc.abstractmethod  # type: ignore
-    @contextmanager
-    def in_subsegment_async(self, name=None, **kwargs) -> AsyncContextManager:
-        """Return a subsegment async context manger.
+    @abc.abstractmethod
+    def add_subsegment(self, subsegment: Any):
+        """Add input subsegment as a child subsegment."""
 
-        Parameters
-        ----------
-        name: str
-            Subsegment name
-        kwargs: Optional[dict]
-            Optional parameters to be propagated to segment
-        """
+    @abc.abstractmethod
+    def remove_subsegment(self, subsegment: Any):
+        """Remove input subsegment from child subsegments."""
 
     @abc.abstractmethod
     def put_annotation(self, key: str, value: Union[str, numbers.Number, bool]) -> NoReturn:
-        """Annotate current active trace entity with a key-value pair.
+        """Annotate segment or subsegment with a key-value pair.
 
         Note: Annotations will be indexed for later search query.
 
@@ -48,9 +43,8 @@ def put_annotation(self, key: str, value: Union[str, numbers.Number, bool]) -> N
 
     @abc.abstractmethod
     def put_metadata(self, key: str, value: Any, namespace: str = "default") -> NoReturn:
-        """Add metadata to the current active trace entity.
-
-        Note: Metadata is not indexed but can be later retrieved by BatchGetTraces API.
+        """Add metadata to segment or subsegment. Metadata is not indexed
+        but can be later retrieved by BatchGetTraces API.
 
         Parameters
         ----------
@@ -63,45 +57,52 @@ def put_metadata(self, key: str, value: Any, namespace: str = "default") -> NoRe
         """
 
     @abc.abstractmethod
-    def patch(self, modules: Set[str]) -> NoReturn:
-        """Instrument a set of supported libraries
+    def add_exception(self, exception: BaseException, stack: List[traceback.StackSummary], remote: bool = False):
+        """Add an exception to trace entities.
 
         Parameters
         ----------
-        modules: Set[str]
-            Set of modules to be patched
-        """
-
-    @abc.abstractmethod
-    def patch_all(self) -> NoReturn:
-        """Instrument all supported libraries"""
+        exception: Exception
+            Caught exception
+        stack: List[traceback.StackSummary]
+            List of traceback summaries
 
+            Output from `traceback.extract_stack()`.
+        remote: bool
+            Whether it's a client error (False) or downstream service error (True), by default False
+        """
 
-class BaseSegment(abc.ABC):
-    """Holds common properties and methods on segment and subsegment."""
 
+class BaseProvider(abc.ABC):
     @abc.abstractmethod
-    def close(self, end_time: Optional[int] = None):
-        """Close the trace entity by setting `end_time`
-        and flip the in progress flag to False.
+    @contextmanager
+    def in_subsegment(self, name=None, **kwargs) -> Generator[BaseSegment, None, None]:
+        """Return a subsegment context manger.
 
         Parameters
         ----------
-        end_time: int
-            Time in epoch seconds, by default current time will be used.
+        name: str
+            Subsegment name
+        kwargs: Optional[dict]
+            Optional parameters to be propagated to segment
         """
 
     @abc.abstractmethod
-    def add_subsegment(self, subsegment: Any):
-        """Add input subsegment as a child subsegment."""
+    @contextmanager
+    def in_subsegment_async(self, name=None, **kwargs) -> Generator[BaseSegment, None, None]:
+        """Return a subsegment async context manger.
 
-    @abc.abstractmethod
-    def remove_subsegment(self, subsegment: Any):
-        """Remove input subsegment from child subsegments."""
+        Parameters
+        ----------
+        name: str
+            Subsegment name
+        kwargs: Optional[dict]
+            Optional parameters to be propagated to segment
+        """
 
     @abc.abstractmethod
     def put_annotation(self, key: str, value: Union[str, numbers.Number, bool]) -> NoReturn:
-        """Annotate segment or subsegment with a key-value pair.
+        """Annotate current active trace entity with a key-value pair.
 
         Note: Annotations will be indexed for later search query.
 
@@ -115,8 +116,9 @@ def put_annotation(self, key: str, value: Union[str, numbers.Number, bool]) -> N
 
     @abc.abstractmethod
     def put_metadata(self, key: str, value: Any, namespace: str = "default") -> NoReturn:
-        """Add metadata to segment or subsegment. Metadata is not indexed
-        but can be later retrieved by BatchGetTraces API.
+        """Add metadata to the current active trace entity.
+
+        Note: Metadata is not indexed but can be later retrieved by BatchGetTraces API.
 
         Parameters
         ----------
@@ -129,17 +131,15 @@ def put_metadata(self, key: str, value: Any, namespace: str = "default") -> NoRe
         """
 
     @abc.abstractmethod
-    def add_exception(self, exception: BaseException, stack: List[traceback.StackSummary], remote: bool = False):
-        """Add an exception to trace entities.
+    def patch(self, modules: Sequence[str]) -> NoReturn:
+        """Instrument a set of supported libraries
 
         Parameters
         ----------
-        exception: Exception
-            Caught exception
-        stack: List[traceback.StackSummary]
-            List of traceback summaries
-
-            Output from `traceback.extract_stack()`.
-        remote: bool
-            Whether it's a client error (False) or downstream service error (True), by default False
+        modules: Set[str]
+            Set of modules to be patched
         """
+
+    @abc.abstractmethod
+    def patch_all(self) -> NoReturn:
+        """Instrument all supported libraries"""

aws_lambda_powertools/tracing/tracer.py

@@ -155,7 +155,7 @@ def __init__(
         self.__build_config(
             service=service, disabled=disabled, auto_patch=auto_patch, patch_modules=patch_modules, provider=provider
         )
-        self.provider = self._config["provider"]
+        self.provider: BaseProvider = self._config["provider"]
         self.disabled = self._config["disabled"]
         self.service = self._config["service"]
         self.auto_patch = self._config["auto_patch"]

docs/core/event_handler/api_gateway.md

@@ -201,7 +201,7 @@ You can use **`not_found`** decorator to override this behavior, and return a cu
 
 You can use **`exception_handler`** decorator with any Python exception. This allows you to handle a common exception outside your route, for example validation errors.
 
-```python hl_lines="14 15" title="Exception handling"
+```python hl_lines="13-14" title="Exception handling"
 --8<-- "examples/event_handler_rest/src/exception_handling.py"
 ```
 

docs/core/event_handler/appsync.md

@@ -58,7 +58,7 @@ Here's an example where we have two separate functions to resolve `getTodo` and
 
 === "getting_started_graphql_api_resolver.py"
 
-    ```python hl_lines="7 13 23 25-26 35 37 48"
+    ```python hl_lines="14 20 30 32-33 42 44 55"
     --8<-- "examples/event_handler_graphql/src/getting_started_graphql_api_resolver.py"
     ```
 
@@ -112,7 +112,7 @@ You can nest `app.resolver()` decorator multiple times when resolving fields wit
 
 === "nested_mappings.py"
 
-    ```python hl_lines="4 10 20-21 23 30"
+    ```python hl_lines="11 17 27-28 28 30 37"
     --8<-- "examples/event_handler_graphql/src/nested_mappings.py"
     ```
 
@@ -126,7 +126,7 @@ You can nest `app.resolver()` decorator multiple times when resolving fields wit
 
 For Lambda Python3.8+ runtime, this utility supports async functions when you use in conjunction with `asyncio.run`.
 
-```python hl_lines="7 14 24-25 34 36" title="Resolving GraphQL resolvers async"
+```python hl_lines="14 21 31-32 41 43" title="Resolving GraphQL resolvers async"
 --8<-- "examples/event_handler_graphql/src/async_resolvers.py"
 ```
 
@@ -151,13 +151,13 @@ Use the following code for `merchantInfo` and `searchMerchant` functions respect
 
 === "graphql_transformer_merchant_info.py"
 
-    ```python hl_lines="4 6 22-23 27-28 36"
+    ```python hl_lines="11 13 29-30 34-35 43"
     --8<-- "examples/event_handler_graphql/src/graphql_transformer_merchant_info.py"
     ```
 
 === "graphql_transformer_search_merchant.py"
 
-    ```python hl_lines="4 6 21-22 36 42"
+    ```python hl_lines="11 13 28-29 43 49"
     --8<-- "examples/event_handler_graphql/src/graphql_transformer_search_merchant.py"
     ```
 
@@ -185,7 +185,7 @@ You can subclass [AppSyncResolverEvent](../../utilities/data_classes.md#appsync-
 
 === "custom_models.py.py"
 
-    ```python hl_lines="4 7 23-25 28-29 36 43"
+    ```python hl_lines="11 14 30-32 35-36 43 50"
     --8<-- "examples/event_handler_graphql/src/custom_models.py"
     ```
 
@@ -214,7 +214,7 @@ Let's assume you have `split_operation.py` as your Lambda function entrypoint an
 
     We import **Router** instead of **AppSyncResolver**; syntax wise is exactly the same.
 
-  	```python hl_lines="4 8 18-19"
+  	```python hl_lines="11 15 25-26"
     --8<-- "examples/event_handler_graphql/src/split_operation_module.py"
   	```
 
@@ -242,7 +242,7 @@ Here's an example of how you can test your synchronous resolvers:
 
 === "assert_graphql_response_module.py"
 
-    ```python hl_lines="10"
+    ```python hl_lines="17"
     --8<-- "examples/event_handler_graphql/src/assert_graphql_response_module.py"
     ```
 
@@ -259,13 +259,13 @@ And an example for testing asynchronous resolvers. Note that this requires the `
 
 === "assert_async_graphql_response.py"
 
-    ```python hl_lines="27"
+    ```python hl_lines="28"
     --8<-- "examples/event_handler_graphql/src/assert_async_graphql_response.py"
     ```
 
 === "assert_async_graphql_response_module.py"
 
-    ```python hl_lines="14"
+    ```python hl_lines="21"
     --8<-- "examples/event_handler_graphql/src/assert_async_graphql_response_module.py"
     ```
 

examples/event_handler_graphql/src/assert_async_graphql_response.py

@@ -1,9 +1,10 @@
 import json
 from dataclasses import dataclass
 from pathlib import Path
+from typing import List
 
 import pytest
-from assert_async_graphql_response_module import Location, app  # instance of AppSyncResolver
+from assert_async_graphql_response_module import Todo, app  # instance of AppSyncResolver
 
 
 @pytest.fixture
@@ -24,7 +25,7 @@ async def test_async_direct_resolver(lambda_context):
     fake_event = json.loads(Path("assert_async_graphql_response.json").read_text())
 
     # WHEN
-    result: list[Location] = await app(fake_event, lambda_context)
+    result: List[Todo] = await app(fake_event, lambda_context)
     # alternatively, you can also run a sync test against `lambda_handler`
     # since `lambda_handler` awaits the coroutine to complete
 

examples/event_handler_graphql/src/assert_async_graphql_response_module.py

@@ -1,5 +1,12 @@
+import sys
+
+if sys.version_info >= (3, 8):
+    from typing import TypedDict
+else:
+    from typing_extensions import TypedDict
+
 import asyncio
-from typing import TypedDict
+from typing import List
 
 import aiohttp
 
@@ -22,11 +29,11 @@ class Todo(TypedDict, total=False):
 
 
 @app.resolver(type_name="Query", field_name="listTodos")
-async def list_todos() -> list[Todo]:
+async def list_todos() -> List[Todo]:
     async with aiohttp.ClientSession(trace_configs=[aiohttp_trace_config()]) as session:
         async with session.get("https://jsonplaceholder.typicode.com/todos") as resp:
-            # first two results to demo assertion
-            return await resp.json()[:2]
+            result: List[Todo] = await resp.json()
+            return result[:2]  # first two results to demo assertion
 
 
 @logger.inject_lambda_context(correlation_id_path=correlation_paths.APPSYNC_RESOLVER)

examples/event_handler_graphql/src/assert_graphql_response_module.py

@@ -1,4 +1,11 @@
-from typing import TypedDict
+import sys
+
+if sys.version_info >= (3, 8):
+    from typing import TypedDict
+else:
+    from typing_extensions import TypedDict
+
+from typing import List
 
 from aws_lambda_powertools import Logger, Tracer
 from aws_lambda_powertools.event_handler import AppSyncResolver
@@ -20,7 +27,7 @@ class Location(TypedDict, total=False):
 @app.resolver(field_name="listLocations")
 @app.resolver(field_name="locations")
 @tracer.capture_method
-def get_locations(name: str, description: str = "") -> list[Location]:  # match GraphQL Query arguments
+def get_locations(name: str, description: str = "") -> List[Location]:  # match GraphQL Query arguments
     return [{"name": name, "description": description}]