diff --git a/aws_lambda_powertools/logging/formatter.py b/aws_lambda_powertools/logging/formatter.py
index de9254a3371..e35c9a7a327 100644
--- a/aws_lambda_powertools/logging/formatter.py
+++ b/aws_lambda_powertools/logging/formatter.py
@@ -58,7 +58,7 @@ class LambdaPowertoolsFormatter(BasePowertoolsFormatter):
     def __init__(
         self,
         json_serializer: Optional[Callable[[Dict], str]] = None,
-        json_deserializer: Optional[Callable[[Dict], str]] = None,
+        json_deserializer: Optional[Callable[[Union[Dict, str, bool, int, float]], str]] = None,
         json_default: Optional[Callable[[Any], Any]] = None,
         datefmt: Optional[str] = None,
         log_record_order: Optional[List[str]] = None,
@@ -106,7 +106,7 @@ def __init__(
         self.update_formatter = self.append_keys  # alias to old method
 
         if self.utc:
-            self.converter = time.gmtime
+            self.converter = time.gmtime  # type: ignore
 
         super(LambdaPowertoolsFormatter, self).__init__(datefmt=self.datefmt)
 
@@ -128,7 +128,7 @@ def format(self, record: logging.LogRecord) -> str:  # noqa: A003
         return self.serialize(log=formatted_log)
 
     def formatTime(self, record: logging.LogRecord, datefmt: Optional[str] = None) -> str:
-        record_ts = self.converter(record.created)
+        record_ts = self.converter(record.created)  # type: ignore
         if datefmt:
             return time.strftime(datefmt, record_ts)
 
@@ -201,7 +201,7 @@ def _extract_log_exception(self, log_record: logging.LogRecord) -> Union[Tuple[s
             Log record with constant traceback info and exception name
         """
         if log_record.exc_info:
-            return self.formatException(log_record.exc_info), log_record.exc_info[0].__name__
+            return self.formatException(log_record.exc_info), log_record.exc_info[0].__name__  # type: ignore
 
         return None, None
 
diff --git a/aws_lambda_powertools/logging/logger.py b/aws_lambda_powertools/logging/logger.py
index 35054f86137..e8b67a2ca7e 100644
--- a/aws_lambda_powertools/logging/logger.py
+++ b/aws_lambda_powertools/logging/logger.py
@@ -361,7 +361,7 @@ def registered_handler(self) -> logging.Handler:
         return handlers[0]
 
     @property
-    def registered_formatter(self) -> Optional[PowertoolsFormatter]:
+    def registered_formatter(self) -> PowertoolsFormatter:
         """Convenience property to access logger formatter"""
         return self.registered_handler.formatter  # type: ignore
 
@@ -405,7 +405,9 @@ def get_correlation_id(self) -> Optional[str]:
         str, optional
             Value for the correlation id
         """
-        return self.registered_formatter.log_format.get("correlation_id")
+        if isinstance(self.registered_formatter, LambdaPowertoolsFormatter):
+            return self.registered_formatter.log_format.get("correlation_id")
+        return None
 
     @staticmethod
     def _get_log_level(level: Union[str, int, None]) -> Union[str, int]:
diff --git a/aws_lambda_powertools/metrics/base.py b/aws_lambda_powertools/metrics/base.py
index 853f06f210b..25e502d0887 100644
--- a/aws_lambda_powertools/metrics/base.py
+++ b/aws_lambda_powertools/metrics/base.py
@@ -90,7 +90,7 @@ def __init__(
         self._metric_unit_options = list(MetricUnit.__members__)
         self.metadata_set = metadata_set if metadata_set is not None else {}
 
-    def add_metric(self, name: str, unit: Union[MetricUnit, str], value: float):
+    def add_metric(self, name: str, unit: Union[MetricUnit, str], value: float) -> None:
         """Adds given metric
 
         Example
@@ -215,7 +215,7 @@ def serialize_metric_set(
             **metric_names_and_values,  # "single_metric": 1.0
         }
 
-    def add_dimension(self, name: str, value: str):
+    def add_dimension(self, name: str, value: str) -> None:
         """Adds given dimension to all metrics
 
         Example
@@ -241,7 +241,7 @@ def add_dimension(self, name: str, value: str):
         # checking before casting improves performance in most cases
         self.dimension_set[name] = value if isinstance(value, str) else str(value)
 
-    def add_metadata(self, key: str, value: Any):
+    def add_metadata(self, key: str, value: Any) -> None:
         """Adds high cardinal metadata for metrics object
 
         This will not be available during metrics visualization.
diff --git a/aws_lambda_powertools/metrics/metric.py b/aws_lambda_powertools/metrics/metric.py
index 1ac2bd9450e..a30f428e38e 100644
--- a/aws_lambda_powertools/metrics/metric.py
+++ b/aws_lambda_powertools/metrics/metric.py
@@ -42,7 +42,7 @@ class SingleMetric(MetricManager):
         Inherits from `aws_lambda_powertools.metrics.base.MetricManager`
     """
 
-    def add_metric(self, name: str, unit: Union[MetricUnit, str], value: float):
+    def add_metric(self, name: str, unit: Union[MetricUnit, str], value: float) -> None:
         """Method to prevent more than one metric being created
 
         Parameters
diff --git a/aws_lambda_powertools/metrics/metrics.py b/aws_lambda_powertools/metrics/metrics.py
index fafc604b505..23e9f542eea 100644
--- a/aws_lambda_powertools/metrics/metrics.py
+++ b/aws_lambda_powertools/metrics/metrics.py
@@ -2,8 +2,9 @@
 import json
 import logging
 import warnings
-from typing import Any, Callable, Dict, Optional
+from typing import Any, Callable, Dict, Optional, Union, cast
 
+from ..shared.types import AnyCallableT
 from .base import MetricManager, MetricUnit
 from .metric import single_metric
 
@@ -87,7 +88,7 @@ def __init__(self, service: Optional[str] = None, namespace: Optional[str] = Non
             service=self.service,
         )
 
-    def set_default_dimensions(self, **dimensions):
+    def set_default_dimensions(self, **dimensions) -> None:
         """Persist dimensions across Lambda invocations
 
         Parameters
@@ -113,10 +114,10 @@ def lambda_handler():
 
         self.default_dimensions.update(**dimensions)
 
-    def clear_default_dimensions(self):
+    def clear_default_dimensions(self) -> None:
         self.default_dimensions.clear()
 
-    def clear_metrics(self):
+    def clear_metrics(self) -> None:
         logger.debug("Clearing out existing metric set from memory")
         self.metric_set.clear()
         self.dimension_set.clear()
@@ -125,11 +126,11 @@ def clear_metrics(self):
 
     def log_metrics(
         self,
-        lambda_handler: Optional[Callable[[Any, Any], Any]] = None,
+        lambda_handler: Union[Callable[[Dict, Any], Any], Optional[Callable[[Dict, Any, Optional[Dict]], Any]]] = None,
         capture_cold_start_metric: bool = False,
         raise_on_empty_metrics: bool = False,
         default_dimensions: Optional[Dict[str, str]] = None,
-    ):
+    ) -> AnyCallableT:
         """Decorator to serialize and publish metrics at the end of a function execution.
 
         Be aware that the log_metrics **does call* the decorated function (e.g. lambda_handler).
@@ -169,11 +170,14 @@ def handler(event, context):
         # Return a partial function with args filled
         if lambda_handler is None:
             logger.debug("Decorator called with parameters")
-            return functools.partial(
-                self.log_metrics,
-                capture_cold_start_metric=capture_cold_start_metric,
-                raise_on_empty_metrics=raise_on_empty_metrics,
-                default_dimensions=default_dimensions,
+            return cast(
+                AnyCallableT,
+                functools.partial(
+                    self.log_metrics,
+                    capture_cold_start_metric=capture_cold_start_metric,
+                    raise_on_empty_metrics=raise_on_empty_metrics,
+                    default_dimensions=default_dimensions,
+                ),
             )
 
         @functools.wraps(lambda_handler)
@@ -194,9 +198,9 @@ def decorate(event, context):
 
             return response
 
-        return decorate
+        return cast(AnyCallableT, decorate)
 
-    def __add_cold_start_metric(self, context: Any):
+    def __add_cold_start_metric(self, context: Any) -> None:
         """Add cold start metric and function_name dimension
 
         Parameters
diff --git a/aws_lambda_powertools/middleware_factory/factory.py b/aws_lambda_powertools/middleware_factory/factory.py
index 74858bf6709..8ab16c5e8b7 100644
--- a/aws_lambda_powertools/middleware_factory/factory.py
+++ b/aws_lambda_powertools/middleware_factory/factory.py
@@ -118,7 +118,7 @@ def final_decorator(func: Optional[Callable] = None, **kwargs):
         if not inspect.isfunction(func):
             # @custom_middleware(True) vs @custom_middleware(log_event=True)
             raise MiddlewareInvalidArgumentError(
-                f"Only keyword arguments is supported for middlewares: {decorator.__qualname__} received {func}"
+                f"Only keyword arguments is supported for middlewares: {decorator.__qualname__} received {func}"  # type: ignore # noqa: E501
             )
 
         @functools.wraps(func)
diff --git a/aws_lambda_powertools/shared/constants.py b/aws_lambda_powertools/shared/constants.py
index 622ffbce47b..45b46d236f9 100644
--- a/aws_lambda_powertools/shared/constants.py
+++ b/aws_lambda_powertools/shared/constants.py
@@ -21,3 +21,5 @@
 
 XRAY_SDK_MODULE: str = "aws_xray_sdk"
 XRAY_SDK_CORE_MODULE: str = "aws_xray_sdk.core"
+
+IDEMPOTENCY_DISABLED_ENV: str = "POWERTOOLS_IDEMPOTENCY_DISABLED"
diff --git a/aws_lambda_powertools/shared/jmespath_utils.py b/aws_lambda_powertools/shared/jmespath_utils.py
index 9cc736aedfb..bbb3b38fe04 100644
--- a/aws_lambda_powertools/shared/jmespath_utils.py
+++ b/aws_lambda_powertools/shared/jmespath_utils.py
@@ -6,22 +6,23 @@
 
 import jmespath
 from jmespath.exceptions import LexerError
+from jmespath.functions import Functions, signature
 
 from aws_lambda_powertools.exceptions import InvalidEnvelopeExpressionError
 
 logger = logging.getLogger(__name__)
 
 
-class PowertoolsFunctions(jmespath.functions.Functions):
-    @jmespath.functions.signature({"types": ["string"]})
+class PowertoolsFunctions(Functions):
+    @signature({"types": ["string"]})
     def _func_powertools_json(self, value):
         return json.loads(value)
 
-    @jmespath.functions.signature({"types": ["string"]})
+    @signature({"types": ["string"]})
     def _func_powertools_base64(self, value):
         return base64.b64decode(value).decode()
 
-    @jmespath.functions.signature({"types": ["string"]})
+    @signature({"types": ["string"]})
     def _func_powertools_base64_gzip(self, value):
         encoded = base64.b64decode(value)
         uncompressed = gzip.decompress(encoded)
diff --git a/aws_lambda_powertools/tracing/tracer.py b/aws_lambda_powertools/tracing/tracer.py
index dc010a3712f..2626793304c 100644
--- a/aws_lambda_powertools/tracing/tracer.py
+++ b/aws_lambda_powertools/tracing/tracer.py
@@ -17,7 +17,7 @@
 logger = logging.getLogger(__name__)
 
 aws_xray_sdk = LazyLoader(constants.XRAY_SDK_MODULE, globals(), constants.XRAY_SDK_MODULE)
-aws_xray_sdk.core = LazyLoader(constants.XRAY_SDK_CORE_MODULE, globals(), constants.XRAY_SDK_CORE_MODULE)
+aws_xray_sdk.core = LazyLoader(constants.XRAY_SDK_CORE_MODULE, globals(), constants.XRAY_SDK_CORE_MODULE)  # type: ignore # noqa: E501
 
 
 class Tracer:
diff --git a/aws_lambda_powertools/utilities/batch/sqs.py b/aws_lambda_powertools/utilities/batch/sqs.py
index e37fdbd3fb5..38773a399dd 100644
--- a/aws_lambda_powertools/utilities/batch/sqs.py
+++ b/aws_lambda_powertools/utilities/batch/sqs.py
@@ -31,6 +31,8 @@ class PartialSQSProcessor(BasePartialProcessor):
         botocore config object
     suppress_exception: bool, optional
         Supress exception raised if any messages fail processing, by default False
+    boto3_session : boto3.session.Session, optional
+            Boto3 session to use for AWS API communication
 
 
     Example
@@ -56,12 +58,18 @@ class PartialSQSProcessor(BasePartialProcessor):
 
     """
 
-    def __init__(self, config: Optional[Config] = None, suppress_exception: bool = False):
+    def __init__(
+        self,
+        config: Optional[Config] = None,
+        suppress_exception: bool = False,
+        boto3_session: Optional[boto3.session.Session] = None,
+    ):
         """
         Initializes sqs client.
         """
         config = config or Config()
-        self.client = boto3.client("sqs", config=config)
+        session = boto3_session or boto3.session.Session()
+        self.client = session.client("sqs", config=config)
         self.suppress_exception = suppress_exception
 
         super().__init__()
@@ -142,6 +150,7 @@ def sqs_batch_processor(
     record_handler: Callable,
     config: Optional[Config] = None,
     suppress_exception: bool = False,
+    boto3_session: Optional[boto3.session.Session] = None,
 ):
     """
     Middleware to handle SQS batch event processing
@@ -160,6 +169,8 @@ def sqs_batch_processor(
             botocore config object
     suppress_exception: bool, optional
         Supress exception raised if any messages fail processing, by default False
+    boto3_session : boto3.session.Session, optional
+            Boto3 session to use for AWS API communication
 
     Examples
     --------
@@ -180,7 +191,9 @@ def sqs_batch_processor(
 
     """
     config = config or Config()
-    processor = PartialSQSProcessor(config=config, suppress_exception=suppress_exception)
+    session = boto3_session or boto3.session.Session()
+
+    processor = PartialSQSProcessor(config=config, suppress_exception=suppress_exception, boto3_session=session)
 
     records = event["Records"]
 
diff --git a/aws_lambda_powertools/utilities/data_classes/sqs_event.py b/aws_lambda_powertools/utilities/data_classes/sqs_event.py
index 0e70684cc3f..57caeea4cc2 100644
--- a/aws_lambda_powertools/utilities/data_classes/sqs_event.py
+++ b/aws_lambda_powertools/utilities/data_classes/sqs_event.py
@@ -75,9 +75,9 @@ def data_type(self) -> str:
 
 
 class SQSMessageAttributes(Dict[str, SQSMessageAttribute]):
-    def __getitem__(self, key: str) -> Optional[SQSMessageAttribute]:
+    def __getitem__(self, key: str) -> Optional[SQSMessageAttribute]:  # type: ignore
         item = super(SQSMessageAttributes, self).get(key)
-        return None if item is None else SQSMessageAttribute(item)
+        return None if item is None else SQSMessageAttribute(item)  # type: ignore
 
 
 class SQSRecord(DictWrapper):
diff --git a/aws_lambda_powertools/utilities/feature_flags/appconfig.py b/aws_lambda_powertools/utilities/feature_flags/appconfig.py
index 103d4d87726..e9d9e29a1fa 100644
--- a/aws_lambda_powertools/utilities/feature_flags/appconfig.py
+++ b/aws_lambda_powertools/utilities/feature_flags/appconfig.py
@@ -47,10 +47,7 @@ def __init__(
             Used to log messages. If None is supplied, one will be created.
         """
         super().__init__()
-        if logger == None:
-            self.logger = logging.getLogger(__name__)
-        else:
-            self.logger = logger
+        self.logger = logger or logging.getLogger(__name__)
         self.environment = environment
         self.application = application
         self.name = name
@@ -60,9 +57,31 @@ def __init__(
         self.jmespath_options = jmespath_options
         self._conf_store = AppConfigProvider(environment=environment, application=application, config=sdk_config)
 
+    @property
+    def get_raw_configuration(self) -> Dict[str, Any]:
+        """Fetch feature schema configuration from AWS AppConfig"""
+        try:
+            # parse result conf as JSON, keep in cache for self.max_age seconds
+            return cast(
+                dict,
+                self._conf_store.get(
+                    name=self.name,
+                    transform=TRANSFORM_TYPE,
+                    max_age=self.cache_seconds,
+                ),
+            )
+        except (GetParameterError, TransformParameterError) as exc:
+            err_msg = traceback.format_exc()
+            if "AccessDenied" in err_msg:
+                raise StoreClientError(err_msg) from exc
+            raise ConfigurationStoreError("Unable to get AWS AppConfig configuration file") from exc
+
     def get_configuration(self) -> Dict[str, Any]:
         """Fetch feature schema configuration from AWS AppConfig
 
+        If envelope is set, it'll extract and return feature flags from configuration,
+        otherwise it'll return the entire configuration fetched from AWS AppConfig.
+
         Raises
         ------
         ConfigurationStoreError
@@ -73,25 +92,11 @@ def get_configuration(self) -> Dict[str, Any]:
         Dict[str, Any]
             parsed JSON dictionary
         """
-        try:
-            # parse result conf as JSON, keep in cache for self.max_age seconds
-            config = cast(
-                dict,
-                self._conf_store.get(
-                    name=self.name,
-                    transform=TRANSFORM_TYPE,
-                    max_age=self.cache_seconds,
-                ),
-            )
+        config = self.get_raw_configuration
 
-            if self.envelope:
-                config = jmespath_utils.extract_data_from_envelope(
-                    data=config, envelope=self.envelope, jmespath_options=self.jmespath_options
-                )
+        if self.envelope:
+            config = jmespath_utils.extract_data_from_envelope(
+                data=config, envelope=self.envelope, jmespath_options=self.jmespath_options
+            )
 
-            return config
-        except (GetParameterError, TransformParameterError) as exc:
-            err_msg = traceback.format_exc()
-            if "AccessDenied" in err_msg:
-                raise StoreClientError(err_msg) from exc
-            raise ConfigurationStoreError("Unable to get AWS AppConfig configuration file") from exc
+        return config
diff --git a/aws_lambda_powertools/utilities/feature_flags/base.py b/aws_lambda_powertools/utilities/feature_flags/base.py
index edb94c4f45d..e323f32d8b1 100644
--- a/aws_lambda_powertools/utilities/feature_flags/base.py
+++ b/aws_lambda_powertools/utilities/feature_flags/base.py
@@ -3,10 +3,19 @@
 
 
 class StoreProvider(ABC):
+    @property
+    @abstractmethod
+    def get_raw_configuration(self) -> Dict[str, Any]:
+        """Get configuration from any store and return the parsed JSON dictionary"""
+        raise NotImplementedError()  # pragma: no cover
+
     @abstractmethod
     def get_configuration(self) -> Dict[str, Any]:
         """Get configuration from any store and return the parsed JSON dictionary
 
+        If envelope is set, it'll extract and return feature flags from configuration,
+        otherwise it'll return the entire configuration fetched from the store.
+
         Raises
         ------
         ConfigurationStoreError
@@ -42,10 +51,10 @@ def get_configuration(self) -> Dict[str, Any]:
         }
         ```
         """
-        return NotImplemented  # pragma: no cover
+        raise NotImplementedError()  # pragma: no cover
 
 
 class BaseValidator(ABC):
     @abstractmethod
     def validate(self):
-        return NotImplemented  # pragma: no cover
+        raise NotImplementedError()  # pragma: no cover
diff --git a/aws_lambda_powertools/utilities/feature_flags/feature_flags.py b/aws_lambda_powertools/utilities/feature_flags/feature_flags.py
index 346a114a518..5bf5f33f3d5 100644
--- a/aws_lambda_powertools/utilities/feature_flags/feature_flags.py
+++ b/aws_lambda_powertools/utilities/feature_flags/feature_flags.py
@@ -1,5 +1,5 @@
 import logging
-from typing import Any, Dict, List, Optional, Union, cast
+from typing import Any, Dict, List, Optional, cast
 
 from . import schema
 from .base import StoreProvider
@@ -36,11 +36,8 @@ def __init__(self, store: StoreProvider, logger=None):
         logger: A logging object
             Used to log messages. If None is supplied, one will be created.
         """
-        self._store = store
-        if logger == None:
-            self.logger = logging.getLogger(__name__)
-        else:
-            self.logger = logger
+        self.store = store
+        self.logger = logger or logging.getLogger(__name__)
 
     def _match_by_action(self, action: str, condition_value: Any, context_value: Any) -> bool:
         if not context_value:
@@ -51,6 +48,10 @@ def _match_by_action(self, action: str, condition_value: Any, context_value: Any
             schema.RuleAction.ENDSWITH.value: lambda a, b: a.endswith(b),
             schema.RuleAction.IN.value: lambda a, b: a in b,
             schema.RuleAction.NOT_IN.value: lambda a, b: a not in b,
+            schema.RuleAction.KEY_IN_VALUE.value: lambda a, b: a in b,
+            schema.RuleAction.KEY_NOT_IN_VALUE.value: lambda a, b: a not in b,
+            schema.RuleAction.VALUE_IN_KEY.value: lambda a, b: b in a,
+            schema.RuleAction.VALUE_NOT_IN_KEY.value: lambda a, b: b not in a,
         }
 
         try:
@@ -97,16 +98,20 @@ def _evaluate_rules(
             rule_match_value = rule.get(schema.RULE_MATCH_VALUE)
 
             # Context might contain PII data; do not log its value
-            self.logger.debug(f"Evaluating rule matching, rule={rule_name}, feature={feature_name}, default={feat_default}")
+            self.logger.debug(
+                f"Evaluating rule matching, rule={rule_name}, feature={feature_name}, default={feat_default}"
+            )
             if self._evaluate_conditions(rule_name=rule_name, feature_name=feature_name, rule=rule, context=context):
                 return bool(rule_match_value)
 
             # no rule matched, return default value of feature
-            self.logger.debug(f"no rule matched, returning feature default, default={feat_default}, name={feature_name}")
+            self.logger.debug(
+                f"no rule matched, returning feature default, default={feat_default}, name={feature_name}"
+            )
             return feat_default
         return False
 
-    def get_configuration(self) -> Union[Dict[str, Dict], Dict]:
+    def get_configuration(self) -> Dict:
         """Get validated feature flag schema from configured store.
 
         Largely used to aid testing, since it's called by `evaluate` and `get_enabled_features` methods.
@@ -149,8 +154,8 @@ def get_configuration(self) -> Union[Dict[str, Dict], Dict]:
         ```
         """
         # parse result conf as JSON, keep in cache for max age defined in store
-        self.logger.debug(f"Fetching schema from registered store, store={self._store}")
-        config = self._store.get_configuration()
+        self.logger.debug(f"Fetching schema from registered store, store={self.store}")
+        config: Dict = self.store.get_configuration()
         validator = schema.SchemaValidator(schema=config)
         validator.validate()
 
diff --git a/aws_lambda_powertools/utilities/feature_flags/schema.py b/aws_lambda_powertools/utilities/feature_flags/schema.py
index 8f64db0dfce..f5111a02cd6 100644
--- a/aws_lambda_powertools/utilities/feature_flags/schema.py
+++ b/aws_lambda_powertools/utilities/feature_flags/schema.py
@@ -20,6 +20,10 @@ class RuleAction(str, Enum):
     ENDSWITH = "ENDSWITH"
     IN = "IN"
     NOT_IN = "NOT_IN"
+    KEY_IN_VALUE = "KEY_IN_VALUE"
+    KEY_NOT_IN_VALUE = "KEY_NOT_IN_VALUE"
+    VALUE_IN_KEY = "VALUE_IN_KEY"
+    VALUE_NOT_IN_KEY = "VALUE_NOT_IN_KEY"
 
 
 class SchemaValidator(BaseValidator):
@@ -78,7 +82,9 @@ class SchemaValidator(BaseValidator):
     The value MUST contain the following members:
 
     * **action**: `str`. Operation to perform to match a key and value.
-    The value MUST be either EQUALS, STARTSWITH, ENDSWITH, IN, NOT_IN
+    The value MUST be either EQUALS, STARTSWITH, ENDSWITH,
+    KEY_IN_VALUE KEY_NOT_IN_VALUE VALUE_IN_KEY VALUE_NOT_IN_KEY
+
     * **key**: `str`. Key in given context to perform operation
     * **value**: `Any`. Value in given context that should match action operation.
 
@@ -105,10 +111,7 @@ class SchemaValidator(BaseValidator):
 
     def __init__(self, schema: Dict[str, Any], logger=None):
         self.schema = schema
-        if logger == None:
-            self.logger = logging.getLogger(__name__)
-        else:
-            self.logger = logger
+        self.logger = logger or logging.getLogger(__name__)
 
     def validate(self) -> None:
         self.logger.debug("Validating schema")
@@ -196,7 +199,7 @@ def validate(self):
         for condition in self.conditions:
             self.validate_condition(rule_name=self.rule_name, condition=condition)
 
-    def validate_condition(self,rule_name: str, condition: Dict[str, str]) -> None:
+    def validate_condition(self, rule_name: str, condition: Dict[str, str]) -> None:
         if not condition or not isinstance(condition, dict):
             raise SchemaValidationError(f"Feature rule condition must be a dictionary, rule={rule_name}")
 
diff --git a/aws_lambda_powertools/utilities/idempotency/idempotency.py b/aws_lambda_powertools/utilities/idempotency/idempotency.py
index 06c9a578aa2..6984cfbbd8e 100644
--- a/aws_lambda_powertools/utilities/idempotency/idempotency.py
+++ b/aws_lambda_powertools/utilities/idempotency/idempotency.py
@@ -3,9 +3,11 @@
 """
 import functools
 import logging
+import os
 from typing import Any, Callable, Dict, Optional, cast
 
 from aws_lambda_powertools.middleware_factory import lambda_handler_decorator
+from aws_lambda_powertools.shared.constants import IDEMPOTENCY_DISABLED_ENV
 from aws_lambda_powertools.shared.types import AnyCallableT
 from aws_lambda_powertools.utilities.idempotency.base import IdempotencyHandler
 from aws_lambda_powertools.utilities.idempotency.config import IdempotencyConfig
@@ -56,6 +58,9 @@ def idempotent(
         >>>     return {"StatusCode": 200}
     """
 
+    if os.getenv(IDEMPOTENCY_DISABLED_ENV):
+        return handler(event, context)
+
     config = config or IdempotencyConfig()
     args = event, context
     idempotency_handler = IdempotencyHandler(
@@ -122,6 +127,9 @@ def process_order(customer_id: str, order: dict, **kwargs):
 
     @functools.wraps(function)
     def decorate(*args, **kwargs):
+        if os.getenv(IDEMPOTENCY_DISABLED_ENV):
+            return function(*args, **kwargs)
+
         payload = kwargs.get(data_keyword_argument)
 
         if payload is None:
diff --git a/aws_lambda_powertools/utilities/idempotency/persistence/dynamodb.py b/aws_lambda_powertools/utilities/idempotency/persistence/dynamodb.py
index 73f241bd613..0ce307ab503 100644
--- a/aws_lambda_powertools/utilities/idempotency/persistence/dynamodb.py
+++ b/aws_lambda_powertools/utilities/idempotency/persistence/dynamodb.py
@@ -62,11 +62,11 @@ def __init__(
             >>>     return {"StatusCode": 200}
         """
 
-        boto_config = boto_config or Config()
-        session = boto3_session or boto3.session.Session()
-        self._ddb_resource = session.resource("dynamodb", config=boto_config)
+        self._boto_config = boto_config or Config()
+        self._boto3_session = boto3_session or boto3.session.Session()
+
+        self._table = None
         self.table_name = table_name
-        self.table = self._ddb_resource.Table(self.table_name)
         self.key_attr = key_attr
         self.expiry_attr = expiry_attr
         self.status_attr = status_attr
@@ -74,6 +74,25 @@ def __init__(
         self.validation_key_attr = validation_key_attr
         super(DynamoDBPersistenceLayer, self).__init__()
 
+    @property
+    def table(self):
+        """
+        Caching property to store boto3 dynamodb Table resource
+
+        """
+        if self._table:
+            return self._table
+        ddb_resource = self._boto3_session.resource("dynamodb", config=self._boto_config)
+        self._table = ddb_resource.Table(self.table_name)
+        return self._table
+
+    @table.setter
+    def table(self, table):
+        """
+        Allow table instance variable to be set directly, primarily for use in tests
+        """
+        self._table = table
+
     def _item_to_data_record(self, item: Dict[str, Any]) -> DataRecord:
         """
         Translate raw item records from DynamoDB to DataRecord
@@ -125,7 +144,7 @@ def _put_record(self, data_record: DataRecord) -> None:
                 ExpressionAttributeNames={"#id": self.key_attr, "#now": self.expiry_attr},
                 ExpressionAttributeValues={":now": int(now.timestamp())},
             )
-        except self._ddb_resource.meta.client.exceptions.ConditionalCheckFailedException:
+        except self.table.meta.client.exceptions.ConditionalCheckFailedException:
             logger.debug(f"Failed to put record for already existing idempotency key: {data_record.idempotency_key}")
             raise IdempotencyItemAlreadyExistsError
 
@@ -155,7 +174,7 @@ def _update_record(self, data_record: DataRecord):
             "ExpressionAttributeNames": expression_attr_names,
         }
 
-        self.table.update_item(**kwargs)  # type: ignore
+        self.table.update_item(**kwargs)
 
     def _delete_record(self, data_record: DataRecord) -> None:
         logger.debug(f"Deleting record for idempotency key: {data_record.idempotency_key}")
diff --git a/aws_lambda_powertools/utilities/parameters/appconfig.py b/aws_lambda_powertools/utilities/parameters/appconfig.py
index 4a400aa7789..d1613c14513 100644
--- a/aws_lambda_powertools/utilities/parameters/appconfig.py
+++ b/aws_lambda_powertools/utilities/parameters/appconfig.py
@@ -29,6 +29,8 @@ class AppConfigProvider(BaseProvider):
         Application of the configuration to pass during client initialization
     config: botocore.config.Config, optional
         Botocore configuration to pass during client initialization
+    boto3_session : boto3.session.Session, optional
+            Boto3 session to use for AWS API communication
 
     Example
     -------
@@ -60,13 +62,20 @@ class AppConfigProvider(BaseProvider):
 
     client: Any = None
 
-    def __init__(self, environment: str, application: Optional[str] = None, config: Optional[Config] = None):
+    def __init__(
+        self,
+        environment: str,
+        application: Optional[str] = None,
+        config: Optional[Config] = None,
+        boto3_session: Optional[boto3.session.Session] = None,
+    ):
         """
         Initialize the App Config client
         """
 
         config = config or Config()
-        self.client = boto3.client("appconfig", config=config)
+        session = boto3_session or boto3.session.Session()
+        self.client = session.client("appconfig", config=config)
         self.application = resolve_env_var_choice(
             choice=application, env=os.getenv(constants.SERVICE_NAME_ENV, "service_undefined")
         )
diff --git a/aws_lambda_powertools/utilities/parameters/dynamodb.py b/aws_lambda_powertools/utilities/parameters/dynamodb.py
index 39bd1a8d6b7..9220edf3b05 100644
--- a/aws_lambda_powertools/utilities/parameters/dynamodb.py
+++ b/aws_lambda_powertools/utilities/parameters/dynamodb.py
@@ -30,6 +30,8 @@ class DynamoDBProvider(BaseProvider):
         Complete url to reference local DynamoDB instance, e.g. http://localhost:8080
     config: botocore.config.Config, optional
         Botocore configuration to pass during client initialization
+    boto3_session : boto3.session.Session, optional
+            Boto3 session to use for AWS API communication
 
     Example
     -------
@@ -149,13 +151,16 @@ def __init__(
         value_attr: str = "value",
         endpoint_url: Optional[str] = None,
         config: Optional[Config] = None,
+        boto3_session: Optional[boto3.session.Session] = None,
     ):
         """
         Initialize the DynamoDB client
         """
 
         config = config or Config()
-        self.table = boto3.resource("dynamodb", endpoint_url=endpoint_url, config=config).Table(table_name)
+        session = boto3_session or boto3.session.Session()
+
+        self.table = session.resource("dynamodb", endpoint_url=endpoint_url, config=config).Table(table_name)
 
         self.key_attr = key_attr
         self.sort_attr = sort_attr
diff --git a/aws_lambda_powertools/utilities/parameters/secrets.py b/aws_lambda_powertools/utilities/parameters/secrets.py
index 5699876d90e..b64e70ae184 100644
--- a/aws_lambda_powertools/utilities/parameters/secrets.py
+++ b/aws_lambda_powertools/utilities/parameters/secrets.py
@@ -19,6 +19,8 @@ class SecretsProvider(BaseProvider):
     ----------
     config: botocore.config.Config, optional
         Botocore configuration to pass during client initialization
+    boto3_session : boto3.session.Session, optional
+            Boto3 session to use for AWS API communication
 
     Example
     -------
@@ -58,14 +60,14 @@ class SecretsProvider(BaseProvider):
 
     client: Any = None
 
-    def __init__(self, config: Optional[Config] = None):
+    def __init__(self, config: Optional[Config] = None, boto3_session: Optional[boto3.session.Session] = None):
         """
         Initialize the Secrets Manager client
         """
 
         config = config or Config()
-
-        self.client = boto3.client("secretsmanager", config=config)
+        session = boto3_session or boto3.session.Session()
+        self.client = session.client("secretsmanager", config=config)
 
         super().__init__()
 
diff --git a/aws_lambda_powertools/utilities/parameters/ssm.py b/aws_lambda_powertools/utilities/parameters/ssm.py
index 2a16ad91f08..4cbb16354c7 100644
--- a/aws_lambda_powertools/utilities/parameters/ssm.py
+++ b/aws_lambda_powertools/utilities/parameters/ssm.py
@@ -19,6 +19,8 @@ class SSMProvider(BaseProvider):
     ----------
     config: botocore.config.Config, optional
         Botocore configuration to pass during client initialization
+    boto3_session : boto3.session.Session, optional
+            Boto3 session to use for AWS API communication
 
     Example
     -------
@@ -74,13 +76,14 @@ class SSMProvider(BaseProvider):
 
     client: Any = None
 
-    def __init__(self, config: Optional[Config] = None):
+    def __init__(self, config: Optional[Config] = None, boto3_session: Optional[boto3.session.Session] = None):
         """
         Initialize the SSM Parameter Store client
         """
 
         config = config or Config()
-        self.client = boto3.client("ssm", config=config)
+        session = boto3_session or boto3.session.Session()
+        self.client = session.client("ssm", config=config)
 
         super().__init__()
 
diff --git a/aws_lambda_powertools/utilities/validation/base.py b/aws_lambda_powertools/utilities/validation/base.py
index 2a337b85971..61d692d7f28 100644
--- a/aws_lambda_powertools/utilities/validation/base.py
+++ b/aws_lambda_powertools/utilities/validation/base.py
@@ -33,10 +33,10 @@ def validate_data_against_schema(data: Union[Dict, str], schema: Dict, formats:
     except (TypeError, AttributeError, fastjsonschema.JsonSchemaDefinitionException) as e:
         raise InvalidSchemaFormatError(f"Schema received: {schema}, Formats: {formats}. Error: {e}")
     except fastjsonschema.JsonSchemaValueException as e:
-        message = f"Failed schema validation. Error: {e.message}, Path: {e.path}, Data: {e.value}"
+        message = f"Failed schema validation. Error: {e.message}, Path: {e.path}, Data: {e.value}"  # noqa: B306
         raise SchemaValidationError(
             message,
-            validation_message=e.message,
+            validation_message=e.message,  # noqa: B306
             name=e.name,
             path=e.path,
             value=e.value,
diff --git a/docs/index.md b/docs/index.md
index 781a96e2eb3..b79a2e7074f 100644
--- a/docs/index.md
+++ b/docs/index.md
@@ -124,6 +124,54 @@ If using SAM, you can include this SAR App as part of your shared Layers stack,
             )
     ```
 
+=== "Terraform"
+
+	> Credits to [Dani Comnea](https://github.com/DanyC97) for providing the Terraform equivalent.
+
+    ```terraform hl_lines="12-13 15-20 23-25 40"
+    terraform {
+      required_version = "~> 0.13"
+      required_providers {
+        aws = "~> 3.50.0"
+      }
+    }
+
+    provider "aws" {
+      region  = "us-east-1"
+    }
+
+    resource "aws_serverlessapplicationrepository_cloudformation_stack" "deploy_sar_stack" {
+      name = "aws-lambda-powertools-python-layer"
+
+      application_id   = data.aws_serverlessapplicationrepository_application.sar_app.application_id
+      semantic_version = data.aws_serverlessapplicationrepository_application.sar_app.semantic_version
+      capabilities = [
+        "CAPABILITY_IAM",
+        "CAPABILITY_NAMED_IAM"
+      ]
+    }
+
+    data "aws_serverlessapplicationrepository_application" "sar_app" {
+      application_id   = "arn:aws:serverlessrepo:eu-west-1:057560766410:applications/aws-lambda-powertools-python-layer"
+      semantic_version = var.aws_powertools_version
+    }
+
+    variable "aws_powertools_version" {
+      type        = string
+      default     = "1.20.2"
+      description = "The AWS Powertools release version"
+    }
+
+    output "deployed_powertools_sar_version" {
+      value = data.aws_serverlessapplicationrepository_application.sar_app.semantic_version
+    }
+
+	# Fetch Lambda Powertools Layer ARN from deployed SAR App
+	output "aws_lambda_powertools_layer_arn" {
+	  value = aws_serverlessapplicationrepository_cloudformation_stack.deploy_sar_stack.outputs.LayerVersionArn
+	}
+    ```
+
 ??? tip "Example of least-privileged IAM permissions to deploy Layer"
 
     > Credits to [mwarkentin](https://github.com/mwarkentin) for providing the scoped down IAM permissions.
diff --git a/docs/utilities/batch.md b/docs/utilities/batch.md
index 96770fb1849..56ab160e9f9 100644
--- a/docs/utilities/batch.md
+++ b/docs/utilities/batch.md
@@ -143,10 +143,13 @@ Use `PartialSQSProcessor` context manager to access a list of all return values
         return result
     ```
 
-### Passing custom boto3 config
+### Customizing boto configuration
 
-If you need to pass custom configuration such as region to the SDK, you can pass your own [botocore config object](https://botocore.amazonaws.com/v1/documentation/api/latest/reference/config.html) to
-the `sqs_batch_processor` decorator:
+The **`config`** and **`boto3_session`** parameters enable you to pass in a custom [botocore config object](https://botocore.amazonaws.com/v1/documentation/api/latest/reference/config.html)
+or a custom [boto3 session](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/core/session.html) when using the `sqs_batch_processor`
+decorator or `PartialSQSProcessor` class.
+
+> Custom config example
 
 === "Decorator"
 
@@ -193,6 +196,53 @@ the `sqs_batch_processor` decorator:
         return result
     ```
 
+> Custom boto3 session example
+
+=== "Decorator"
+
+    ```python  hl_lines="4 12"
+    from aws_lambda_powertools.utilities.batch import sqs_batch_processor
+    from botocore.config import Config
+
+    session = boto3.session.Session()
+
+    def record_handler(record):
+        # This will be called for each individual message from a batch
+        # It should raise an exception if the message was not processed successfully
+        return_value = do_something_with(record["body"])
+        return return_value
+
+    @sqs_batch_processor(record_handler=record_handler, boto3_session=session)
+    def lambda_handler(event, context):
+        return {"statusCode": 200}
+    ```
+
+=== "Context manager"
+
+    ```python  hl_lines="4 16"
+    from aws_lambda_powertools.utilities.batch import PartialSQSProcessor
+    import boto3
+
+    session = boto3.session.Session()
+
+    def record_handler(record):
+        # This will be called for each individual message from a batch
+        # It should raise an exception if the message was not processed successfully
+        return_value = do_something_with(record["body"])
+        return return_value
+
+
+    def lambda_handler(event, context):
+        records = event["Records"]
+
+        processor = PartialSQSProcessor(boto3_session=session)
+
+        with processor(records, record_handler):
+            result = processor.process()
+
+        return result
+    ```
+
 ### Suppressing exceptions
 
 If you want to disable the default behavior where `SQSBatchProcessingError` is raised if there are any errors, you can pass the `suppress_exception` boolean argument.
diff --git a/docs/utilities/feature_flags.md b/docs/utilities/feature_flags.md
index f5f899e3d11..37a50ae293c 100644
--- a/docs/utilities/feature_flags.md
+++ b/docs/utilities/feature_flags.md
@@ -366,7 +366,7 @@ You can use `get_enabled_features` method for scenarios where you need a list of
                     "when_match": true,
                     "conditions": [
                         {
-                            "action": "IN",
+                            "action": "KEY_IN_VALUE",
                             "key": "CloudFront-Viewer-Country",
                             "value": ["NL", "IE", "UK", "PL", "PT"]
                         }
@@ -450,9 +450,20 @@ The `conditions` block is a list of conditions that contain `action`, `key`, and
     }
     ```
 
-The `action` configuration can have 5 different values: `EQUALS`, `STARTSWITH`, `ENDSWITH`, `IN`, `NOT_IN`.
+The `action` configuration can have the following values, where the expressions **`a`** is the `key` and **`b`** is the `value` above:
 
-The `key` and `value` will be compared to the input from the context parameter.
+Action | Equivalent expression
+------------------------------------------------- | ---------------------------------------------------------------------------------
+**EQUALS** | `lambda a, b: a == b`
+**STARTSWITH** | `lambda a, b: a.startswith(b)`
+**ENDSWITH** | `lambda a, b: a.endswith(b)`
+**KEY_IN_VALUE** | `lambda a, b: a in b`
+**KEY_NOT_IN_VALUE** | `lambda a, b: a not in b`
+**VALUE_IN_KEY** | `lambda a, b: b in a`
+**VALUE_NOT_IN_KEY** | `lambda a, b: b not in a`
+
+
+!!! info "The `**key**` and `**value**` will be compared to the input from the `**context**` parameter."
 
 **For multiple conditions**, we will evaluate the list of conditions as a logical `AND`, so all conditions needs to match to return `when_match` value.
 
@@ -529,6 +540,27 @@ For this to work, you need to use a JMESPath expression via the `envelope` param
     }
     ```
 
+### Getting fetched configuration
+
+You can access the configuration fetched from the store via `get_raw_configuration` property within the store instance.
+
+=== "app.py"
+
+    ```python hl_lines="12"
+    from aws_lambda_powertools.utilities.feature_flags import FeatureFlags, AppConfigStore
+
+    app_config = AppConfigStore(
+        environment="dev",
+        application="product-catalogue",
+        name="configuration",
+        envelope = "feature_flags"
+    )
+
+	feature_flags = FeatureFlags(store=app_config)
+
+	config = app_config.get_raw_configuration
+    ```
+
 ### Built-in store provider
 
 !!! info "For GA, you'll be able to bring your own store."
@@ -651,3 +683,11 @@ Method | When to use | Requires new deployment on changes | Supported services
 **[Environment variables](https://docs.aws.amazon.com/lambda/latest/dg/configuration-envvars.html){target="_blank"}** | Simple configuration that will rarely if ever change, because changing it requires a Lambda function deployment. | Yes | Lambda
 **[Parameters utility](parameters.md)** | Access to secrets, or fetch parameters in different formats from AWS System Manager Parameter Store or Amazon DynamoDB. | No | Parameter Store, DynamoDB, Secrets Manager, AppConfig
 **Feature flags utility** | Rule engine to define when one or multiple features should be enabled depending on the input. | No | AppConfig
+
+
+## Deprecation list when GA
+
+Breaking change | Recommendation
+------------------------------------------------- | ---------------------------------------------------------------------------------
+`IN` RuleAction | Use `KEY_IN_VALUE` instead
+`NOT_IN` RuleAction | Use `KEY_NOT_IN_VALUE` instead
diff --git a/docs/utilities/idempotency.md b/docs/utilities/idempotency.md
index a9a5a129e63..bf06e3292b7 100644
--- a/docs/utilities/idempotency.md
+++ b/docs/utilities/idempotency.md
@@ -548,7 +548,7 @@ This means that we will raise **`IdempotencyKeyError`** if the evaluation of **`
 
 ### Customizing boto configuration
 
-You can provide a custom boto configuration via **`boto_config`**, or an existing boto session via **`boto3_session`** parameters, when constructing the persistence store.
+The **`boto_config`** and **`boto3_session`** parameters enable you to pass in a custom [botocore config object](https://botocore.amazonaws.com/v1/documentation/api/latest/reference/config.html) or a custom [boto3 session](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/core/session.html) when constructing the persistence store.
 
 === "Custom session"
 
@@ -765,6 +765,123 @@ The idempotency utility can be used with the `validator` decorator. Ensure that
 !!! tip "JMESPath Powertools functions are also available"
     Built-in functions known in the validation utility like `powertools_json`, `powertools_base64`, `powertools_base64_gzip` are also available to use in this utility.
 
+
+## Testing your code
+
+The idempotency utility provides several routes to test your code.
+
+### Disabling the idempotency utility
+When testing your code, you may wish to disable the idempotency logic altogether and focus on testing your business logic. To do this, you can set the environment variable `POWERTOOLS_IDEMPOTENCY_DISABLED`
+with a truthy value. If you prefer setting this for specific tests, and are using Pytest, you can use [monkeypatch](https://docs.pytest.org/en/latest/monkeypatch.html) fixture:
+
+=== "tests.py"
+
+    ```python hl_lines="2 3"
+    def test_idempotent_lambda_handler(monkeypatch):
+        # Set POWERTOOLS_IDEMPOTENCY_DISABLED before calling decorated functions
+        monkeypatch.setenv("POWERTOOLS_IDEMPOTENCY_DISABLED", 1)
+
+        result = handler()
+        ...
+    ```
+=== "app.py"
+
+    ```python
+    from aws_lambda_powertools.utilities.idempotency import (
+    DynamoDBPersistenceLayer, idempotent
+    )
+
+    persistence_layer = DynamoDBPersistenceLayer(table_name="idempotency")
+
+    @idempotent(persistence_store=persistence_layer)
+    def handler(event, context):
+        print('expensive operation')
+        return {
+            "payment_id": 12345,
+            "message": "success",
+            "statusCode": 200,
+        }
+    ```
+
+### Testing with DynamoDB Local
+
+To test with [DynamoDB Local](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/DynamoDBLocal.DownloadingAndRunning.html), you can replace the `Table` resource used by the persistence layer with one you create inside your tests. This allows you to set the endpoint_url.
+
+=== "tests.py"
+
+    ```python hl_lines="6 7 8"
+    import boto3
+
+    import app
+
+    def test_idempotent_lambda():
+        # Create our own Table resource using the endpoint for our DynamoDB Local instance
+        resource = boto3.resource("dynamodb", endpoint_url='http://localhost:8000')
+        table = resource.Table(app.persistence_layer.table_name)
+        app.persistence_layer.table = table
+
+        result = app.handler({'testkey': 'testvalue'}, {})
+        assert result['payment_id'] == 12345
+    ```
+
+=== "app.py"
+
+    ```python
+    from aws_lambda_powertools.utilities.idempotency import (
+    DynamoDBPersistenceLayer, idempotent
+    )
+
+    persistence_layer = DynamoDBPersistenceLayer(table_name="idempotency")
+
+    @idempotent(persistence_store=persistence_layer)
+    def handler(event, context):
+        print('expensive operation')
+        return {
+            "payment_id": 12345,
+            "message": "success",
+            "statusCode": 200,
+        }
+    ```
+
+### How do I mock all DynamoDB I/O operations
+
+The idempotency utility lazily creates the dynamodb [Table](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/dynamodb.html#table) which it uses to access DynamoDB.
+This means it is possible to pass a mocked Table resource, or stub various methods.
+
+=== "tests.py"
+
+    ```python hl_lines="6 7 8 9"
+    from unittest.mock import MagicMock
+
+    import app
+
+    def test_idempotent_lambda():
+        table = MagicMock()
+        app.persistence_layer.table = table
+        result = app.handler({'testkey': 'testvalue'}, {})
+        table.put_item.assert_called()
+        ...
+    ```
+
+=== "app.py"
+
+    ```python
+    from aws_lambda_powertools.utilities.idempotency import (
+    DynamoDBPersistenceLayer, idempotent
+    )
+
+    persistence_layer = DynamoDBPersistenceLayer(table_name="idempotency")
+
+    @idempotent(persistence_store=persistence_layer)
+    def handler(event, context):
+        print('expensive operation')
+        return {
+            "payment_id": 12345,
+            "message": "success",
+            "statusCode": 200,
+        }
+    ```
+
 ## Extra resources
 
 If you're interested in a deep dive on how Amazon uses idempotency when building our APIs, check out
diff --git a/docs/utilities/parameters.md b/docs/utilities/parameters.md
index 081d22817ab..51fd0196abd 100644
--- a/docs/utilities/parameters.md
+++ b/docs/utilities/parameters.md
@@ -501,3 +501,40 @@ Here is the mapping between this utility's functions and methods and the underly
 | DynamoDB            | `DynamoDBProvider.get`          | `dynamodb`       | ([Table resource](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/dynamodb.html#table)) | [get_item](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/dynamodb.html#DynamoDB.Table.get_item)
 | DynamoDB            | `DynamoDBProvider.get_multiple` | `dynamodb`       | ([Table resource](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/dynamodb.html#table)) | [query](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/dynamodb.html#DynamoDB.Table.query)
 | App Config          | `get_app_config`                | `appconfig`      | [get_configuration](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/appconfig.html#AppConfig.Client.get_configuration) |
+
+
+### Customizing boto configuration
+
+The **`config`** and **`boto3_session`** parameters enable you to pass in a custom [botocore config object](https://botocore.amazonaws.com/v1/documentation/api/latest/reference/config.html) or a custom [boto3 session](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/core/session.html) when constructing any of the built-in provider classes.
+
+> **Example**
+
+
+=== "Custom session"
+
+    ```python hl_lines="2 4 5"
+    from aws_lambda_powertools.utilities import parameters
+    import boto3
+
+    boto3_session = boto3.session.Session()
+    ssm_provider = parameters.SSMProvider(boto3_session=boto3_session)
+
+    def handler(event, context):
+        # Retrieve a single parameter
+        value = ssm_provider.get("/my/parameter")
+        ...
+    ```
+=== "Custom config"
+
+    ```python hl_lines="2 4 5"
+    from aws_lambda_powertools.utilities import parameters
+    from botocore.config import Config
+
+    boto_config = Config()
+    ssm_provider = parameters.SSMProvider(config=boto_config)
+
+    def handler(event, context):
+        # Retrieve a single parameter
+        value = ssm_provider.get("/my/parameter")
+        ...
+    ```
diff --git a/mypy.ini b/mypy.ini
index 2436d7074d2..faf6014a54d 100644
--- a/mypy.ini
+++ b/mypy.ini
@@ -11,6 +11,12 @@ show_error_context = True
 [mypy-jmespath]
 ignore_missing_imports=True
 
+[mypy-jmespath.exceptions]
+ignore_missing_imports=True
+
+[mypy-jmespath.functions]
+ignore_missing_imports=True
+
 [mypy-boto3]
 ignore_missing_imports = True
 
diff --git a/poetry.lock b/poetry.lock
index 2205938537e..374ef8ad735 100644
--- a/poetry.lock
+++ b/poetry.lock
@@ -81,14 +81,14 @@ d = ["aiohttp (>=3.3.2)", "aiohttp-cors"]
 
 [[package]]
 name = "boto3"
-version = "1.18.49"
+version = "1.18.51"
 description = "The AWS SDK for Python"
 category = "main"
 optional = false
 python-versions = ">= 3.6"
 
 [package.dependencies]
-botocore = ">=1.21.49,<1.22.0"
+botocore = ">=1.21.51,<1.22.0"
 jmespath = ">=0.7.1,<1.0.0"
 s3transfer = ">=0.5.0,<0.6.0"
 
@@ -97,7 +97,7 @@ crt = ["botocore[crt] (>=1.21.0,<2.0a0)"]
 
 [[package]]
 name = "botocore"
-version = "1.21.49"
+version = "1.21.51"
 description = "Low-level, data-driven core of boto 3."
 category = "main"
 optional = false
@@ -244,7 +244,7 @@ toml = "*"
 
 [[package]]
 name = "flake8-bugbear"
-version = "21.9.1"
+version = "21.9.2"
 description = "A plugin for flake8 finding likely bugs and design problems in your program. Contains warnings that don't belong in pyflakes and pycodestyle."
 category = "dev"
 optional = false
@@ -1047,7 +1047,7 @@ pydantic = ["pydantic", "email-validator"]
 [metadata]
 lock-version = "1.1"
 python-versions = "^3.6.1"
-content-hash = "ddec4b961b63336adb798a590c98df41a874da5ee4b0e37993c3c3a0437dc839"
+content-hash = "c55d59c37d135eeba1ae059cbfe702c6167dd68645d1001bc90482dddfde8ab4"
 
 [metadata.files]
 appdirs = [
@@ -1074,12 +1074,12 @@ black = [
     {file = "black-20.8b1.tar.gz", hash = "sha256:1c02557aa099101b9d21496f8a914e9ed2222ef70336404eeeac8edba836fbea"},
 ]
 boto3 = [
-    {file = "boto3-1.18.49-py3-none-any.whl", hash = "sha256:1a2908d2829268f1b2355bad3a96bfdc8e41523629b5d958bcedfc35d2d232dd"},
-    {file = "boto3-1.18.49.tar.gz", hash = "sha256:e520655c9caf2f21853644d88b59b1c32bc44ccd58b20574883b25eb6256d938"},
+    {file = "boto3-1.18.51-py3-none-any.whl", hash = "sha256:83d6f539e0f0e0f0c532bb2b11d1e9c5055d1d806d64a61aff4f49399c294ee7"},
+    {file = "boto3-1.18.51.tar.gz", hash = "sha256:98279095b1d08ee6d8d587f2c66fda6d560ad3046e98cd140c1aa8e1ed018c70"},
 ]
 botocore = [
-    {file = "botocore-1.21.49-py3-none-any.whl", hash = "sha256:eab89183f7d94cabacde79a266060bb9429249e33a39b7ba4c1b15c965095477"},
-    {file = "botocore-1.21.49.tar.gz", hash = "sha256:0161c3b64e34315928aae7fdbce49e684c9c2cfad2435cb22023b7ad87306f12"},
+    {file = "botocore-1.21.51-py3-none-any.whl", hash = "sha256:2089f9fa36a59d8c02435c49d58ccc7b3ceb9c0c054ea4f71631c3c3a1c5245e"},
+    {file = "botocore-1.21.51.tar.gz", hash = "sha256:17a10dd33334e7e3aaa4e12f66317284f96bb53267e20bc877a187c442681772"},
 ]
 certifi = [
     {file = "certifi-2020.12.5-py2.py3-none-any.whl", hash = "sha256:719a74fb9e33b9bd44cc7f3a8d94bc35e4049deebe19ba7d8e108280cfd59830"},
@@ -1179,8 +1179,8 @@ flake8-black = [
     {file = "flake8_black-0.2.3-py3-none-any.whl", hash = "sha256:cc080ba5b3773b69ba102b6617a00cc4ecbad8914109690cfda4d565ea435d96"},
 ]
 flake8-bugbear = [
-    {file = "flake8-bugbear-21.9.1.tar.gz", hash = "sha256:2f60c8ce0dc53d51da119faab2d67dea978227f0f92ed3c44eb7d65fb2e06a96"},
-    {file = "flake8_bugbear-21.9.1-py36.py37.py38-none-any.whl", hash = "sha256:45bfdccfb9f2d8aa140e33cac8f46f1e38215c13d5aa8650e7e188d84e2f94c6"},
+    {file = "flake8-bugbear-21.9.2.tar.gz", hash = "sha256:db9a09893a6c649a197f5350755100bb1dd84f110e60cf532fdfa07e41808ab2"},
+    {file = "flake8_bugbear-21.9.2-py36.py37.py38-none-any.whl", hash = "sha256:4f7eaa6f05b7d7ea4cbbde93f7bcdc5438e79320fa1ec420d860c181af38b769"},
 ]
 flake8-builtins = [
     {file = "flake8-builtins-1.5.3.tar.gz", hash = "sha256:09998853b2405e98e61d2ff3027c47033adbdc17f9fe44ca58443d876eb00f3b"},
diff --git a/pyproject.toml b/pyproject.toml
index c5197c626ab..2f2c2549d9e 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -49,7 +49,7 @@ bandit = "^1.7.0"
 radon = "^5.1.0"
 xenon = "^0.8.0"
 flake8-eradicate = "^1.1.0"
-flake8-bugbear = "^21.9.1"
+flake8-bugbear = "^21.9.2"
 mkdocs-material = "^7.3.0"
 mkdocs-git-revision-date-plugin = "^0.3.1"
 mike = "^0.6.0"
diff --git a/tests/functional/feature_flags/test_feature_flags.py b/tests/functional/feature_flags/test_feature_flags.py
index 5342105da3d..f6ce93abaa9 100644
--- a/tests/functional/feature_flags/test_feature_flags.py
+++ b/tests/functional/feature_flags/test_feature_flags.py
@@ -301,6 +301,8 @@ def test_flags_conditions_rule_match_multiple_actions_multiple_rules_multiple_co
 
 
 # check a case where the feature exists but the rule doesn't match so we revert to the default value of the feature
+
+# Check IN/NOT_IN/KEY_IN_VALUE/KEY_NOT_IN_VALUE/VALUE_IN_KEY/VALUE_NOT_IN_KEY conditions
 def test_flags_match_rule_with_in_action(mocker, config):
     expected_value = True
     mocked_app_config_schema = {
@@ -397,6 +399,207 @@ def test_flags_no_match_rule_with_not_in_action(mocker, config):
     assert toggle == expected_value
 
 
+def test_flags_match_rule_with_key_in_value_action(mocker, config):
+    expected_value = True
+    mocked_app_config_schema = {
+        "my_feature": {
+            "default": False,
+            "rules": {
+                "tenant id is contained in [6, 2]": {
+                    "when_match": expected_value,
+                    "conditions": [
+                        {
+                            "action": RuleAction.KEY_IN_VALUE.value,
+                            "key": "tenant_id",
+                            "value": ["6", "2"],
+                        }
+                    ],
+                }
+            },
+        }
+    }
+    feature_flags = init_feature_flags(mocker, mocked_app_config_schema, config)
+    toggle = feature_flags.evaluate(name="my_feature", context={"tenant_id": "6", "username": "a"}, default=False)
+    assert toggle == expected_value
+
+
+def test_flags_no_match_rule_with_key_in_value_action(mocker, config):
+    expected_value = False
+    mocked_app_config_schema = {
+        "my_feature": {
+            "default": expected_value,
+            "rules": {
+                "tenant id is contained in [8, 2]": {
+                    "when_match": True,
+                    "conditions": [
+                        {
+                            "action": RuleAction.KEY_IN_VALUE.value,
+                            "key": "tenant_id",
+                            "value": ["8", "2"],
+                        }
+                    ],
+                }
+            },
+        }
+    }
+    feature_flags = init_feature_flags(mocker, mocked_app_config_schema, config)
+    toggle = feature_flags.evaluate(name="my_feature", context={"tenant_id": "6", "username": "a"}, default=False)
+    assert toggle == expected_value
+
+
+def test_flags_match_rule_with_key_not_in_value_action(mocker, config):
+    expected_value = True
+    mocked_app_config_schema = {
+        "my_feature": {
+            "default": False,
+            "rules": {
+                "tenant id is contained in [8, 2]": {
+                    "when_match": expected_value,
+                    "conditions": [
+                        {
+                            "action": RuleAction.KEY_NOT_IN_VALUE.value,
+                            "key": "tenant_id",
+                            "value": ["10", "4"],
+                        }
+                    ],
+                }
+            },
+        }
+    }
+    feature_flags = init_feature_flags(mocker, mocked_app_config_schema, config)
+    toggle = feature_flags.evaluate(name="my_feature", context={"tenant_id": "6", "username": "a"}, default=False)
+    assert toggle == expected_value
+
+
+def test_flags_no_match_rule_with_key_not_in_value_action(mocker, config):
+    expected_value = False
+    mocked_app_config_schema = {
+        "my_feature": {
+            "default": expected_value,
+            "rules": {
+                "tenant id is contained in [8, 2]": {
+                    "when_match": True,
+                    "conditions": [
+                        {
+                            "action": RuleAction.KEY_NOT_IN_VALUE.value,
+                            "key": "tenant_id",
+                            "value": ["6", "4"],
+                        }
+                    ],
+                }
+            },
+        }
+    }
+    feature_flags = init_feature_flags(mocker, mocked_app_config_schema, config)
+    toggle = feature_flags.evaluate(name="my_feature", context={"tenant_id": "6", "username": "a"}, default=False)
+    assert toggle == expected_value
+
+
+def test_flags_match_rule_with_value_in_key_action(mocker, config):
+    expected_value = True
+    mocked_app_config_schema = {
+        "my_feature": {
+            "default": False,
+            "rules": {
+                "user is in the SYSADMIN group": {
+                    "when_match": expected_value,
+                    "conditions": [
+                        {
+                            "action": RuleAction.VALUE_IN_KEY.value,
+                            "key": "groups",
+                            "value": "SYSADMIN",
+                        }
+                    ],
+                }
+            },
+        }
+    }
+    feature_flags = init_feature_flags(mocker, mocked_app_config_schema, config)
+    toggle = feature_flags.evaluate(
+        name="my_feature", context={"tenant_id": "6", "username": "a", "groups": ["SYSADMIN", "IT"]}, default=False
+    )
+    assert toggle == expected_value
+
+
+def test_flags_no_match_rule_with_value_in_key_action(mocker, config):
+    expected_value = False
+    mocked_app_config_schema = {
+        "my_feature": {
+            "default": expected_value,
+            "rules": {
+                "tenant id is contained in [8, 2]": {
+                    "when_match": True,
+                    "conditions": [
+                        {
+                            "action": RuleAction.VALUE_IN_KEY.value,
+                            "key": "groups",
+                            "value": "GUEST",
+                        }
+                    ],
+                }
+            },
+        }
+    }
+    feature_flags = init_feature_flags(mocker, mocked_app_config_schema, config)
+    toggle = feature_flags.evaluate(
+        name="my_feature", context={"tenant_id": "6", "username": "a", "groups": ["SYSADMIN", "IT"]}, default=False
+    )
+    assert toggle == expected_value
+
+
+def test_flags_match_rule_with_value_not_in_key_action(mocker, config):
+    expected_value = True
+    mocked_app_config_schema = {
+        "my_feature": {
+            "default": False,
+            "rules": {
+                "user is in the GUEST group": {
+                    "when_match": expected_value,
+                    "conditions": [
+                        {
+                            "action": RuleAction.VALUE_NOT_IN_KEY.value,
+                            "key": "groups",
+                            "value": "GUEST",
+                        }
+                    ],
+                }
+            },
+        }
+    }
+    feature_flags = init_feature_flags(mocker, mocked_app_config_schema, config)
+    toggle = feature_flags.evaluate(
+        name="my_feature", context={"tenant_id": "6", "username": "a", "groups": ["SYSADMIN", "IT"]}, default=False
+    )
+    assert toggle == expected_value
+
+
+def test_flags_no_match_rule_with_value_not_in_key_action(mocker, config):
+    expected_value = False
+    mocked_app_config_schema = {
+        "my_feature": {
+            "default": expected_value,
+            "rules": {
+                "user is in the SYSADMIN group": {
+                    "when_match": True,
+                    "conditions": [
+                        {
+                            "action": RuleAction.VALUE_NOT_IN_KEY.value,
+                            "key": "groups",
+                            "value": "SYSADMIN",
+                        }
+                    ],
+                }
+            },
+        }
+    }
+    feature_flags = init_feature_flags(mocker, mocked_app_config_schema, config)
+    toggle = feature_flags.evaluate(
+        name="my_feature", context={"tenant_id": "6", "username": "a", "groups": ["SYSADMIN", "IT"]}, default=False
+    )
+    assert toggle == expected_value
+
+
+# Check multiple features
 def test_multiple_features_enabled(mocker, config):
     expected_value = ["my_feature", "my_feature2"]
     mocked_app_config_schema = {
@@ -587,3 +790,15 @@ def test_get_feature_toggle_propagates_access_denied_error(mocker, config):
     # THEN raise StoreClientError error
     with pytest.raises(StoreClientError, match="AccessDeniedException") as err:
         feature_flags.evaluate(name="Foo", default=False)
+
+
+def test_get_configuration_with_envelope_and_raw(mocker, config):
+    expected_value = True
+    mocked_app_config_schema = {"log_level": "INFO", "features": {"my_feature": {"default": expected_value}}}
+    feature_flags = init_feature_flags(mocker, mocked_app_config_schema, config, envelope="features")
+
+    features_config = feature_flags.get_configuration()
+    config = feature_flags.store.get_raw_configuration
+
+    assert "log_level" in config
+    assert "log_level" not in features_config
diff --git a/tests/functional/feature_flags/test_schema_validation.py b/tests/functional/feature_flags/test_schema_validation.py
index ce85494afce..1cd14aa4287 100644
--- a/tests/functional/feature_flags/test_schema_validation.py
+++ b/tests/functional/feature_flags/test_schema_validation.py
@@ -220,6 +220,26 @@ def test_valid_condition_all_actions():
                             CONDITION_KEY: "username",
                             CONDITION_VALUE: ["c"],
                         },
+                        {
+                            CONDITION_ACTION: RuleAction.KEY_IN_VALUE.value,
+                            CONDITION_KEY: "username",
+                            CONDITION_VALUE: ["a", "b"],
+                        },
+                        {
+                            CONDITION_ACTION: RuleAction.KEY_NOT_IN_VALUE.value,
+                            CONDITION_KEY: "username",
+                            CONDITION_VALUE: ["c"],
+                        },
+                        {
+                            CONDITION_ACTION: RuleAction.VALUE_IN_KEY.value,
+                            CONDITION_KEY: "groups",
+                            CONDITION_VALUE: "SYSADMIN",
+                        },
+                        {
+                            CONDITION_ACTION: RuleAction.VALUE_NOT_IN_KEY.value,
+                            CONDITION_KEY: "groups",
+                            CONDITION_VALUE: "GUEST",
+                        },
                     ],
                 }
             },
diff --git a/tests/functional/idempotency/test_idempotency.py b/tests/functional/idempotency/test_idempotency.py
index cb0d43ae6fa..b1d0914d181 100644
--- a/tests/functional/idempotency/test_idempotency.py
+++ b/tests/functional/idempotency/test_idempotency.py
@@ -3,6 +3,7 @@
 import json
 import sys
 from hashlib import md5
+from unittest.mock import MagicMock
 
 import jmespath
 import pytest
@@ -994,3 +995,25 @@ def dummy(payload):
 
     # WHEN
     dummy(payload=data_two)
+
+
+def test_idempotency_disabled_envvar(monkeypatch, lambda_context, persistence_store: DynamoDBPersistenceLayer):
+    # Scenario to validate no requests sent to dynamodb table when 'POWERTOOLS_IDEMPOTENCY_DISABLED' is set
+    mock_event = {"data": "value"}
+
+    persistence_store.table = MagicMock()
+
+    monkeypatch.setenv("POWERTOOLS_IDEMPOTENCY_DISABLED", "1")
+
+    @idempotent_function(data_keyword_argument="data", persistence_store=persistence_store)
+    def dummy(data):
+        return {"message": "hello"}
+
+    @idempotent(persistence_store=persistence_store)
+    def dummy_handler(event, context):
+        return {"message": "hi"}
+
+    dummy(data=mock_event)
+    dummy_handler(mock_event, lambda_context)
+
+    assert len(persistence_store.table.method_calls) == 0