fix(mypy): addresses lack of optional types

aws_lambda_powertools/event_handler/api_gateway.py

@@ -126,7 +126,11 @@ class Response:
     """Response data class that provides greater control over what is returned from the proxy event"""
 
     def __init__(
-        self, status_code: int, content_type: Optional[str], body: Union[str, bytes, None], headers: Dict = None
+        self,
+        status_code: int,
+        content_type: Optional[str],
+        body: Union[str, bytes, None],
+        headers: Optional[Dict] = None,
     ):
         """
 
@@ -167,7 +171,7 @@ def __init__(
 class ResponseBuilder:
     """Internally used Response builder"""
 
-    def __init__(self, response: Response, route: Route = None):
+    def __init__(self, response: Response, route: Optional[Route] = None):
         self.response = response
         self.route = route
 
@@ -199,7 +203,7 @@ def _route(self, event: BaseProxyEvent, cors: Optional[CORSConfig]):
         if self.route.compress and "gzip" in (event.get_header_value("accept-encoding", "") or ""):
             self._compress()
 
-    def build(self, event: BaseProxyEvent, cors: CORSConfig = None) -> Dict[str, Any]:
+    def build(self, event: BaseProxyEvent, cors: Optional[CORSConfig] = None) -> Dict[str, Any]:
         """Build the full response dict to be returned by the lambda"""
         self._route(event, cors)
 
@@ -250,7 +254,7 @@ def lambda_handler(event, context):
     def __init__(
         self,
         proxy_type: Enum = ProxyEventType.APIGatewayProxyEvent,
-        cors: CORSConfig = None,
+        cors: Optional[CORSConfig] = None,
         debug: Optional[bool] = None,
     ):
         """
@@ -270,10 +274,10 @@ def __init__(
         self._cors_enabled: bool = cors is not None
         self._cors_methods: Set[str] = {"OPTIONS"}
         self._debug = resolve_truthy_env_var_choice(
-            choice=debug, env=os.getenv(constants.EVENT_HANDLER_DEBUG_ENV, "false")
+            env=os.getenv(constants.EVENT_HANDLER_DEBUG_ENV, "false"), choice=debug
         )
 
-    def get(self, rule: str, cors: bool = None, compress: bool = False, cache_control: str = None):
+    def get(self, rule: str, cors: Optional[bool] = None, compress: bool = False, cache_control: Optional[str] = None):
         """Get route decorator with GET `method`
 
         Examples
@@ -298,7 +302,7 @@ def lambda_handler(event, context):
         """
         return self.route(rule, "GET", cors, compress, cache_control)
 
-    def post(self, rule: str, cors: bool = None, compress: bool = False, cache_control: str = None):
+    def post(self, rule: str, cors: Optional[bool] = None, compress: bool = False, cache_control: Optional[str] = None):
         """Post route decorator with POST `method`
 
         Examples
@@ -324,7 +328,7 @@ def lambda_handler(event, context):
         """
         return self.route(rule, "POST", cors, compress, cache_control)
 
-    def put(self, rule: str, cors: bool = None, compress: bool = False, cache_control: str = None):
+    def put(self, rule: str, cors: Optional[bool] = None, compress: bool = False, cache_control: Optional[str] = None):
         """Put route decorator with PUT `method`
 
         Examples
@@ -350,7 +354,9 @@ def lambda_handler(event, context):
         """
         return self.route(rule, "PUT", cors, compress, cache_control)
 
-    def delete(self, rule: str, cors: bool = None, compress: bool = False, cache_control: str = None):
+    def delete(
+        self, rule: str, cors: Optional[bool] = None, compress: bool = False, cache_control: Optional[str] = None
+    ):
         """Delete route decorator with DELETE `method`
 
         Examples
@@ -375,7 +381,9 @@ def lambda_handler(event, context):
         """
         return self.route(rule, "DELETE", cors, compress, cache_control)
 
-    def patch(self, rule: str, cors: bool = None, compress: bool = False, cache_control: str = None):
+    def patch(
+        self, rule: str, cors: Optional[bool] = None, compress: bool = False, cache_control: Optional[str] = None
+    ):
         """Patch route decorator with PATCH `method`
 
         Examples
@@ -403,7 +411,14 @@ def lambda_handler(event, context):
         """
         return self.route(rule, "PATCH", cors, compress, cache_control)
 
-    def route(self, rule: str, method: str, cors: bool = None, compress: bool = False, cache_control: str = None):
+    def route(
+        self,
+        rule: str,
+        method: str,
+        cors: Optional[bool] = None,
+        compress: bool = False,
+        cache_control: Optional[str] = None,
+    ):
         """Route decorator includes parameter `method`"""
 
         def register_resolver(func: Callable):

aws_lambda_powertools/event_handler/appsync.py

@@ -1,5 +1,5 @@
 import logging
-from typing import Any, Callable
+from typing import Any, Callable, Optional
 
 from aws_lambda_powertools.utilities.data_classes import AppSyncResolverEvent
 from aws_lambda_powertools.utilities.typing import LambdaContext
@@ -44,7 +44,7 @@ def common_field() -> str:
     def __init__(self):
         self._resolvers: dict = {}
 
-    def resolver(self, type_name: str = "*", field_name: str = None):
+    def resolver(self, type_name: str = "*", field_name: Optional[str] = None):
         """Registers the resolver for field_name
 
         Parameters

aws_lambda_powertools/logging/formatter.py

@@ -60,8 +60,8 @@ def __init__(
         json_serializer: Optional[Callable[[Dict], str]] = None,
         json_deserializer: Optional[Callable[[Dict], str]] = None,
         json_default: Optional[Callable[[Any], Any]] = None,
-        datefmt: str = None,
-        log_record_order: List[str] = None,
+        datefmt: Optional[str] = None,
+        log_record_order: Optional[List[str]] = None,
         utc: bool = False,
         **kwargs
     ):

aws_lambda_powertools/logging/logger.py

@@ -167,11 +167,11 @@ class Logger(logging.Logger):  # lgtm [py/missing-call-to-init]
 
     def __init__(
         self,
-        service: str = None,
-        level: Union[str, int] = None,
+        service: Optional[str] = None,
+        level: Union[str, int, None] = None,
         child: bool = False,
-        sampling_rate: float = None,
-        stream: IO[str] = None,
+        sampling_rate: Optional[float] = None,
+        stream: Optional[IO[str]] = None,
         logger_formatter: Optional[PowertoolsFormatter] = None,
         logger_handler: Optional[logging.Handler] = None,
         **kwargs,
@@ -261,10 +261,10 @@ def _configure_sampling(self):
 
     def inject_lambda_context(
         self,
-        lambda_handler: Callable[[Dict, Any], Any] = None,
-        log_event: bool = None,
-        correlation_id_path: str = None,
-        clear_state: bool = False,
+        lambda_handler: Optional[Callable[[Dict, Any], Any]] = None,
+        log_event: Optional[bool] = None,
+        correlation_id_path: Optional[str] = None,
+        clear_state: Optional[bool] = False,
     ):
         """Decorator to capture Lambda contextual info and inject into logger
 
@@ -324,7 +324,7 @@ def handler(event, context):
             )
 
         log_event = resolve_truthy_env_var_choice(
-            choice=log_event, env=os.getenv(constants.LOGGER_LOG_EVENT_ENV, "false")
+            env=os.getenv(constants.LOGGER_LOG_EVENT_ENV, "false"), choice=log_event
         )
 
         @functools.wraps(lambda_handler)
@@ -421,7 +421,9 @@ def _get_caller_filename():
 
 
 def set_package_logger(
-    level: Union[str, int] = logging.DEBUG, stream: IO[str] = None, formatter: logging.Formatter = None
+    level: Union[str, int] = logging.DEBUG,
+    stream: Optional[IO[str]] = None,
+    formatter: Optional[logging.Formatter] = None,
 ):
     """Set an additional stream handler, formatter, and log level for aws_lambda_powertools package logger.
 

aws_lambda_powertools/metrics/base.py

@@ -5,7 +5,7 @@
 import os
 from collections import defaultdict
 from enum import Enum
-from typing import Any, Dict, List, Union
+from typing import Any, Dict, List, Optional, Union
 
 from ..shared import constants
 from ..shared.functions import resolve_env_var_choice
@@ -76,11 +76,11 @@ class MetricManager:
 
     def __init__(
         self,
-        metric_set: Dict[str, Any] = None,
-        dimension_set: Dict = None,
-        namespace: str = None,
-        metadata_set: Dict[str, Any] = None,
-        service: str = None,
+        metric_set: Optional[Dict[str, Any]] = None,
+        dimension_set: Optional[Dict] = None,
+        namespace: Optional[str] = None,
+        metadata_set: Optional[Dict[str, Any]] = None,
+        service: Optional[str] = None,
     ):
         self.metric_set = metric_set if metric_set is not None else {}
         self.dimension_set = dimension_set if dimension_set is not None else {}
@@ -136,7 +136,9 @@ def add_metric(self, name: str, unit: Union[MetricUnit, str], value: float):
             # since we could have more than 100 metrics
             self.metric_set.clear()
 
-    def serialize_metric_set(self, metrics: Dict = None, dimensions: Dict = None, metadata: Dict = None) -> Dict:
+    def serialize_metric_set(
+        self, metrics: Optional[Dict] = None, dimensions: Optional[Dict] = None, metadata: Optional[Dict] = None
+    ) -> Dict:
         """Serializes metric and dimensions set
 
         Parameters

aws_lambda_powertools/metrics/metric.py

@@ -61,7 +61,7 @@ def add_metric(self, name: str, unit: Union[MetricUnit, str], value: float):
 
 
 @contextmanager
-def single_metric(name: str, unit: MetricUnit, value: float, namespace: str = None):
+def single_metric(name: str, unit: MetricUnit, value: float, namespace: Optional[str] = None):
     """Context manager to simplify creation of a single metric
 
     Example

aws_lambda_powertools/metrics/metrics.py

@@ -71,7 +71,7 @@ def lambda_handler():
     _metadata: Dict[str, Any] = {}
     _default_dimensions: Dict[str, Any] = {}
 
-    def __init__(self, service: str = None, namespace: str = None):
+    def __init__(self, service: Optional[str] = None, namespace: Optional[str] = None):
         self.metric_set = self._metrics
         self.service = service
         self.namespace: Optional[str] = namespace
@@ -125,10 +125,10 @@ def clear_metrics(self):
 
     def log_metrics(
         self,
-        lambda_handler: Callable[[Any, Any], Any] = None,
+        lambda_handler: Optional[Callable[[Any, Any], Any]] = None,
         capture_cold_start_metric: bool = False,
         raise_on_empty_metrics: bool = False,
-        default_dimensions: Dict[str, str] = None,
+        default_dimensions: Optional[Dict[str, str]] = None,
     ):
         """Decorator to serialize and publish metrics at the end of a function execution.
 

aws_lambda_powertools/middleware_factory/factory.py

@@ -2,7 +2,7 @@
 import inspect
 import logging
 import os
-from typing import Callable
+from typing import Callable, Optional
 
 from ..shared import constants
 from ..shared.functions import resolve_truthy_env_var_choice
@@ -12,7 +12,7 @@
 logger = logging.getLogger(__name__)
 
 
-def lambda_handler_decorator(decorator: Callable = None, trace_execution: bool = None):
+def lambda_handler_decorator(decorator: Optional[Callable] = None, trace_execution: Optional[bool] = None):
     """Decorator factory for decorating Lambda handlers.
 
     You can use lambda_handler_decorator to create your own middlewares,
@@ -106,11 +106,11 @@ def lambda_handler(event, context):
         return functools.partial(lambda_handler_decorator, trace_execution=trace_execution)
 
     trace_execution = resolve_truthy_env_var_choice(
-        choice=trace_execution, env=os.getenv(constants.MIDDLEWARE_FACTORY_TRACE_ENV, "false")
+        env=os.getenv(constants.MIDDLEWARE_FACTORY_TRACE_ENV, "false"), choice=trace_execution
     )
 
     @functools.wraps(decorator)
-    def final_decorator(func: Callable = None, **kwargs):
+    def final_decorator(func: Optional[Callable] = None, **kwargs):
         # If called with kwargs return new func with kwargs
         if func is None:
             return functools.partial(final_decorator, **kwargs)

aws_lambda_powertools/shared/functions.py

@@ -2,14 +2,14 @@
 from typing import Any, Optional, Union
 
 
-def resolve_truthy_env_var_choice(env: Any, choice: bool = None) -> bool:
+def resolve_truthy_env_var_choice(env: str, choice: Optional[bool] = None) -> bool:
     """Pick explicit choice over truthy env value, if available, otherwise return truthy env value
 
     NOTE: Environment variable should be resolved by the caller.
 
     Parameters
     ----------
-    env : Any
+    env : str
         environment variable actual value
     choice : bool
         explicit choice

aws_lambda_powertools/tracing/base.py

@@ -2,11 +2,11 @@
 import numbers
 import traceback
 from contextlib import contextmanager
-from typing import Any, AsyncContextManager, ContextManager, List, NoReturn, Set, Union
+from typing import Any, AsyncContextManager, ContextManager, List, NoReturn, Optional, Set, Union
 
 
 class BaseProvider(abc.ABC):
-    @abc.abstractmethod
+    @abc.abstractmethod  # type: ignore
     @contextmanager
     def in_subsegment(self, name=None, **kwargs) -> ContextManager:
         """Return a subsegment context manger.
@@ -19,7 +19,7 @@ def in_subsegment(self, name=None, **kwargs) -> ContextManager:
             Optional parameters to be propagated to segment
         """
 
-    @abc.abstractmethod
+    @abc.abstractmethod  # type: ignore
     @contextmanager
     def in_subsegment_async(self, name=None, **kwargs) -> AsyncContextManager:
         """Return a subsegment async context manger.
@@ -81,7 +81,7 @@ class BaseSegment(abc.ABC):
     """Holds common properties and methods on segment and subsegment."""
 
     @abc.abstractmethod
-    def close(self, end_time: int = None):
+    def close(self, end_time: Optional[int] = None):
         """Close the trace entity by setting `end_time`
         and flip the in progress flag to False.