Module aws_lambda_powertools.shared.functions

Functions

def abs_lambda_path(relative_path: str = '') ‑> str

Expand source code

def abs_lambda_path(relative_path: str = "") -> str:
    """Return the absolute path from the given relative path to lambda handler.

    Parameters
    ----------
    relative_path : str, optional
        The relative path to the lambda handler, by default an empty string.

    Returns
    -------
    str
        The absolute path generated from the given relative path.
        If the environment variable LAMBDA_TASK_ROOT is set, it will use that value.
        Otherwise, it will use the current working directory.
        If the path is empty, it will return the current working directory.
    """
    # Retrieve the LAMBDA_TASK_ROOT environment variable or default to an empty string
    current_working_directory = os.environ.get("LAMBDA_TASK_ROOT", "") or str(Path.cwd())

    return str(Path(current_working_directory, relative_path))

Return the absolute path from the given relative path to lambda handler.

Parameters

relative_path : str, optional

The relative path to the lambda handler, by default an empty string.

Returns

str

The absolute path generated from the given relative path. If the environment variable LAMBDA_TASK_ROOT is set, it will use that value. Otherwise, it will use the current working directory. If the path is empty, it will return the current working directory.

def base64_decode(value: str) ‑> bytes

Expand source code

def base64_decode(value: str) -> bytes:
    try:
        logger.debug("Decoding base64 item to bytes")
        return base64.b64decode(value)
    except (BinAsciiError, TypeError):
        raise ValueError("base64 decode failed - is this base64 encoded string?")

def bytes_to_base64_string(value: bytes) ‑> str

Expand source code

def bytes_to_base64_string(value: bytes) -> str:
    try:
        logger.debug("Encoding bytes to base64 string")
        return base64.b64encode(value).decode()
    except TypeError:
        raise ValueError(f"base64 encoding failed - is this bytes data? type: {type(value)}")

def bytes_to_string(value: bytes) ‑> str

Expand source code

def bytes_to_string(value: bytes) -> str:
    try:
        return value.decode("utf-8")
    except (BinAsciiError, TypeError):
        raise ValueError("base64 UTF-8 decode failed")

def dataclass_to_dict(data) ‑> dict

Expand source code

def dataclass_to_dict(data) -> dict:
    """Dump standard dataclass as dict.

    Note we use lazy import to prevent bloating other code parts.

    Parameters
    ----------
    data: dataclass
        Dataclass

    Returns
    -------

    dict:
        Pydantic model serialized to dict
    """
    import dataclasses

    return dataclasses.asdict(data)

Dump standard dataclass as dict.

Note we use lazy import to prevent bloating other code parts.

Parameters

data : dataclass

Dataclass

Returns

dict:

Pydantic model serialized to dict

def extract_event_from_common_models(data: Any) ‑> dict | typing.Any

Expand source code

def extract_event_from_common_models(data: Any) -> dict | Any:
    """Extract raw event from common types used in Powertools

    If event cannot be extracted, return received data as is.

    Common models:

        - Event Source Data Classes (DictWrapper)
        - Python Dataclasses
        - Pydantic Models (BaseModel)

    Parameters
    ----------
    data : Any
        Original event, a potential instance of DictWrapper/BaseModel/Dataclass

    Notes
    -----

    Why not using static type for function argument?

    DictWrapper would cause a circular import. Pydantic BaseModel could
    cause a ModuleNotFound or trigger init reflection worsening cold start.
    """
    # Short-circuit most common type first for perf
    if isinstance(data, dict):
        return data

    # Is it an Event Source Data Class?
    if getattr(data, "raw_event", None):
        return data.raw_event

    # Is it a Pydantic Model?
    if is_pydantic(data):
        return pydantic_to_dict(data)

    # Is it a Dataclass?
    if is_dataclass(data):
        return dataclass_to_dict(data)

    # Return as is
    return data

Extract raw event from common types used in Powertools

If event cannot be extracted, return received data as is.

Common models:

- Event Source Data Classes (DictWrapper)
- Python Dataclasses
- Pydantic Models (BaseModel)

Parameters

data : Any

Original event, a potential instance of DictWrapper/BaseModel/Dataclass

Notes

Why not using static type for function argument?

DictWrapper would cause a circular import. Pydantic BaseModel could cause a ModuleNotFound or trigger init reflection worsening cold start.

def is_dataclass(data) ‑> bool

Expand source code

def is_dataclass(data) -> bool:
    """Whether data is a dataclass

    Parameters
    ----------
    data: dataclass
        Dataclass obj

    Returns
    -------
    bool
        Whether it's a Dataclass
    """
    return getattr(data, "__dataclass_fields__", False)

Whether data is a dataclass

Parameters

data : dataclass

Dataclass obj

Returns

bool

Whether it's a Dataclass

def is_pydantic(data) ‑> bool

Expand source code

def is_pydantic(data) -> bool:
    """Whether data is a Pydantic model by checking common field available in v1/v2

    Parameters
    ----------
    data: BaseModel
        Pydantic model

    Returns
    -------
    bool
        Whether it's a Pydantic model
    """
    return getattr(data, "json", False)

Whether data is a Pydantic model by checking common field available in v1/v2

Parameters

data : BaseModel

Pydantic model

Returns

bool

Whether it's a Pydantic model

def powertools_debug_is_set() ‑> bool

Expand source code

def powertools_debug_is_set() -> bool:
    is_on = strtobool(os.getenv(constants.POWERTOOLS_DEBUG_ENV, "0"))
    if is_on:
        warnings.warn("POWERTOOLS_DEBUG environment variable is enabled. Setting logging level to DEBUG.", stacklevel=2)
        return True

    return False

def powertools_dev_is_set() ‑> bool

Expand source code

def powertools_dev_is_set() -> bool:
    is_on = strtobool(os.getenv(constants.POWERTOOLS_DEV_ENV, "0"))
    if is_on:
        warnings.warn(
            "POWERTOOLS_DEV environment variable is enabled. Increasing verbosity across utilities.",
            stacklevel=2,
        )
        return True

    return False

def pydantic_to_dict(data) ‑> dict

Expand source code

def pydantic_to_dict(data) -> dict:
    """Dump Pydantic model v1 and v2 as dict.

    Note we use lazy import since Pydantic is an optional dependency.

    Parameters
    ----------
    data: BaseModel
        Pydantic model

    Returns
    -------

    dict:
        Pydantic model serialized to dict
    """
    from aws_lambda_powertools.event_handler.openapi.compat import _model_dump

    return _model_dump(data)

Dump Pydantic model v1 and v2 as dict.

Note we use lazy import since Pydantic is an optional dependency.

Parameters

data : BaseModel

Pydantic model

Returns

dict:

Pydantic model serialized to dict

def resolve_env_var_choice(env: str | None = None, choice: str | float | None = None) ‑> str | float | None

Expand source code

def resolve_env_var_choice(
    env: str | None = None,
    choice: str | float | None = None,
) -> str | float | None:
    """Pick explicit choice over env, if available, otherwise return env value received

    NOTE: Environment variable should be resolved by the caller.

    Parameters
    ----------
    env : str, Optional
        environment variable actual value
    choice : str|float, optional
        explicit choice

    Returns
    -------
    choice : str, Optional
        resolved choice as either bool or environment value
    """
    return choice if choice is not None else env

Pick explicit choice over env, if available, otherwise return env value received

NOTE: Environment variable should be resolved by the caller.

Parameters

env : str, Optional

environment variable actual value

choice : str|float, optional

explicit choice

Returns

choice : str, Optional

resolved choice as either bool or environment value

def resolve_max_age(env: str, choice: int | None) ‑> int

Expand source code

def resolve_max_age(env: str, choice: int | None) -> int:
    """Resolve max age value"""
    return choice if choice is not None else int(env)

Resolve max age value

def resolve_truthy_env_var_choice(env: str, choice: bool | None = None) ‑> bool

Expand source code

def resolve_truthy_env_var_choice(env: str, choice: bool | None = 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 : str
        environment variable actual value
    choice : bool
        explicit choice

    Returns
    -------
    choice : str
        resolved choice as either bool or environment value
    """
    return choice if choice is not None else strtobool(env)

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 : str

environment variable actual value

choice : bool

explicit choice

Returns

choice : str

resolved choice as either bool or environment value

def sanitize_xray_segment_name(name: str) ‑> str

Expand source code

def sanitize_xray_segment_name(name: str) -> str:
    return re.sub(constants.INVALID_XRAY_NAME_CHARACTERS, "", name)

def slice_dictionary(data: dict, chunk_size: int) ‑> Generator[dict, None, None]

Expand source code

def slice_dictionary(data: dict, chunk_size: int) -> Generator[dict, None, None]:
    for _ in range(0, len(data), chunk_size):
        yield {dict_key: data[dict_key] for dict_key in itertools.islice(data, chunk_size)}

def strtobool(value: str) ‑> bool

Expand source code

def strtobool(value: str) -> bool:
    """Convert a string representation of truth to True or False.

    True values are 'y', 'yes', 't', 'true', 'on', and '1'; false values
    are 'n', 'no', 'f', 'false', 'off', and '0'.  Raises ValueError if
    'value' is anything else.

    > note:: Copied from distutils.util.
    """
    value = value.lower()
    if value in ("1", "y", "yes", "t", "true", "on"):
        return True
    if value in ("0", "n", "no", "f", "false", "off"):
        return False
    raise ValueError(f"invalid truth value {value!r}")

Convert a string representation of truth to True or False.

True values are 'y', 'yes', 't', 'true', 'on', and '1'; false values are 'n', 'no', 'f', 'false', 'off', and '0'. Raises ValueError if 'value' is anything else.

note:: Copied from distutils.util.