chore: general simplifications and cleanup

aws_lambda_powertools/logging/filters.py

@@ -13,4 +13,4 @@ def filter(self, record):  # noqa: A003
         created by loggers who don't have a handler.
         """
         logger = record.name
-        return False if self.logger in logger else True
+        return self.logger not in logger

aws_lambda_powertools/logging/formatter.py

@@ -49,9 +49,7 @@ def _build_root_keys(**kwargs):
     @staticmethod
     def _get_latest_trace_id():
         xray_trace_id = os.getenv("_X_AMZN_TRACE_ID")
-        trace_id = xray_trace_id.split(";")[0].replace("Root=", "") if xray_trace_id else None
-
-        return trace_id
+        return xray_trace_id.split(";")[0].replace("Root=", "") if xray_trace_id else None
 
     def update_formatter(self, **kwargs):
         self.format_dict.update(kwargs)

aws_lambda_powertools/logging/logger.py

@@ -291,9 +291,10 @@ def _get_log_level(level: Union[str, int]) -> Union[str, int]:
             return level
 
         log_level: str = level or os.getenv("LOG_LEVEL")
-        log_level = log_level.upper() if log_level is not None else logging.INFO
+        if log_level is None:
+            return logging.INFO
 
-        return log_level
+        return log_level.upper()
 
     @staticmethod
     def _get_caller_filename():
@@ -303,9 +304,7 @@ def _get_caller_filename():
         # Before previous frame => Caller
         frame = inspect.currentframe()
         caller_frame = frame.f_back.f_back.f_back
-        filename = caller_frame.f_globals["__name__"]
-
-        return filename
+        return caller_frame.f_globals["__name__"]
 
 
 def set_package_logger(

aws_lambda_powertools/metrics/base.py

@@ -240,10 +240,7 @@ def add_dimension(self, name: str, value: str):
         # Cast value to str according to EMF spec
         # Majority of values are expected to be string already, so
         # checking before casting improves performance in most cases
-        if isinstance(value, str):
-            self.dimension_set[name] = value
-        else:
-            self.dimension_set[name] = str(value)
+        self.dimension_set[name] = value if isinstance(value, str) else str(value)
 
     def add_metadata(self, key: str, value: Any):
         """Adds high cardinal metadata for metrics object

aws_lambda_powertools/tracing/tracer.py

@@ -454,24 +454,22 @@ async def async_tasks():
         method_name = f"{method.__name__}"
 
         if inspect.iscoroutinefunction(method):
-            decorate = self._decorate_async_function(
+            return self._decorate_async_function(
                 method=method, capture_response=capture_response, method_name=method_name
             )
         elif inspect.isgeneratorfunction(method):
-            decorate = self._decorate_generator_function(
+            return self._decorate_generator_function(
                 method=method, capture_response=capture_response, method_name=method_name
             )
         elif hasattr(method, "__wrapped__") and inspect.isgeneratorfunction(method.__wrapped__):
-            decorate = self._decorate_generator_function_with_context_manager(
+            return self._decorate_generator_function_with_context_manager(
                 method=method, capture_response=capture_response, method_name=method_name
             )
         else:
-            decorate = self._decorate_sync_function(
+            return self._decorate_sync_function(
                 method=method, capture_response=capture_response, method_name=method_name
             )
 
-        return decorate
-
     def _decorate_async_function(self, method: Callable = None, capture_response: bool = True, method_name: str = None):
         @functools.wraps(method)
         async def decorate(*args, **kwargs):
@@ -650,9 +648,9 @@ def __build_config(
 
         self._config["provider"] = provider if provider is not None else self._config["provider"]
         self._config["auto_patch"] = auto_patch if auto_patch is not None else self._config["auto_patch"]
-        self._config["service"] = is_service if is_service else self._config["service"]
-        self._config["disabled"] = is_disabled if is_disabled else self._config["disabled"]
-        self._config["patch_modules"] = patch_modules if patch_modules else self._config["patch_modules"]
+        self._config["service"] = is_service or self._config["service"]
+        self._config["disabled"] = is_disabled or self._config["disabled"]
+        self._config["patch_modules"] = patch_modules or self._config["patch_modules"]
 
     @classmethod
     def _reset_config(cls):

aws_lambda_powertools/utilities/parameters/dynamodb.py

@@ -206,8 +206,4 @@ def _get_multiple(self, path: str, **sdk_options) -> Dict[str, str]:
             response = self.table.query(**sdk_options)
             items.extend(response.get("Items", []))
 
-        retval = {}
-        for item in items:
-            retval[item[self.sort_attr]] = item[self.value_attr]
-
-        return retval
+        return {item[self.sort_attr]: item[self.value_attr] for item in items}

aws_lambda_powertools/utilities/parser/envelopes/cloudwatch.py

@@ -36,7 +36,6 @@ def parse(self, data: Optional[Union[Dict[str, Any], Any]], model: Model) -> Lis
         logger.debug(f"Parsing incoming data with SNS model {CloudWatchLogsModel}")
         parsed_envelope = CloudWatchLogsModel.parse_obj(data)
         logger.debug(f"Parsing CloudWatch records in `body` with {model}")
-        output = []
-        for record in parsed_envelope.awslogs.decoded_data.logEvents:
-            output.append(self._parse(data=record.message, model=model))
-        return output
+        return [
+            self._parse(data=record.message, model=model) for record in parsed_envelope.awslogs.decoded_data.logEvents
+        ]

aws_lambda_powertools/utilities/parser/envelopes/dynamodb.py

@@ -32,13 +32,11 @@ def parse(self, data: Optional[Union[Dict[str, Any], Any]], model: Model) -> Lis
         """
         logger.debug(f"Parsing incoming data with DynamoDB Stream model {DynamoDBStreamModel}")
         parsed_envelope = DynamoDBStreamModel.parse_obj(data)
-        output = []
         logger.debug(f"Parsing DynamoDB Stream new and old records with {model}")
-        for record in parsed_envelope.Records:
-            output.append(
-                {
-                    "NewImage": self._parse(data=record.dynamodb.NewImage, model=model),
-                    "OldImage": self._parse(data=record.dynamodb.OldImage, model=model),
-                }
-            )
-        return output
+        return [
+            {
+                "NewImage": self._parse(data=record.dynamodb.NewImage, model=model),
+                "OldImage": self._parse(data=record.dynamodb.OldImage, model=model),
+            }
+            for record in parsed_envelope.Records
+        ]

aws_lambda_powertools/utilities/parser/envelopes/kinesis.py

@@ -11,7 +11,7 @@
 class KinesisDataStreamEnvelope(BaseEnvelope):
     """Kinesis Data Stream Envelope to extract array of Records
 
-    The record's data parameter is a base64 encoded string which is parsed into a bytes array, 
+    The record's data parameter is a base64 encoded string which is parsed into a bytes array,
     though it can also be a JSON encoded string.
     Regardless of its type it'll be parsed into a BaseModel object.
 
@@ -36,8 +36,7 @@ def parse(self, data: Optional[Union[Dict[str, Any], Any]], model: Model) -> Lis
         """
         logger.debug(f"Parsing incoming data with Kinesis model {KinesisDataStreamModel}")
         parsed_envelope: KinesisDataStreamModel = KinesisDataStreamModel.parse_obj(data)
-        output = []
         logger.debug(f"Parsing Kinesis records in `body` with {model}")
-        for record in parsed_envelope.Records:
-            output.append(self._parse(data=record.kinesis.data.decode("utf-8"), model=model))
-        return output
+        return [
+            self._parse(data=record.kinesis.data.decode("utf-8"), model=model) for record in parsed_envelope.Records
+        ]

aws_lambda_powertools/utilities/parser/envelopes/sns.py

@@ -35,8 +35,5 @@ def parse(self, data: Optional[Union[Dict[str, Any], Any]], model: Model) -> Lis
         """
         logger.debug(f"Parsing incoming data with SNS model {SnsModel}")
         parsed_envelope = SnsModel.parse_obj(data)
-        output = []
         logger.debug(f"Parsing SNS records in `body` with {model}")
-        for record in parsed_envelope.Records:
-            output.append(self._parse(data=record.Sns.Message, model=model))
-        return output
+        return [self._parse(data=record.Sns.Message, model=model) for record in parsed_envelope.Records]

aws_lambda_powertools/utilities/parser/envelopes/sqs.py

@@ -35,8 +35,5 @@ def parse(self, data: Optional[Union[Dict[str, Any], Any]], model: Model) -> Lis
         """
         logger.debug(f"Parsing incoming data with SQS model {SqsModel}")
         parsed_envelope = SqsModel.parse_obj(data)
-        output = []
         logger.debug(f"Parsing SQS records in `body` with {model}")
-        for record in parsed_envelope.Records:
-            output.append(self._parse(data=record.body, model=model))
-        return output
+        return [self._parse(data=record.body, model=model) for record in parsed_envelope.Records]