Module aws_lambda_powertools.utilities.data_classes.vpc_lattice

Expand source code 
from typing import Any, Dict, Optional, overload

from aws_lambda_powertools.shared.headers_serializer import (
    BaseHeadersSerializer,
    HttpApiHeadersSerializer,
)
from aws_lambda_powertools.utilities.data_classes.common import BaseProxyEvent, DictWrapper
from aws_lambda_powertools.utilities.data_classes.shared_functions import (
    base64_decode,
    get_header_value,
    get_query_string_value,
)


class VPCLatticeEventBase(BaseProxyEvent):
    @property
    def body(self) -> str:
        """The VPC Lattice body."""
        return self["body"]

    @property
    def json_body(self) -> Any:
        """Parses the submitted body as json"""
        if self._json_data is None:
            self._json_data = self._json_deserializer(self.decoded_body)
        return self._json_data

    @property
    def headers(self) -> Dict[str, str]:
        """The VPC Lattice event headers."""
        return self["headers"]

    @property
    def decoded_body(self) -> str:
        """Dynamically base64 decode body as a str"""
        body: str = self["body"]
        if self.is_base64_encoded:
            return base64_decode(body)
        return body

    @property
    def method(self) -> str:
        """The VPC Lattice method used. Valid values include: DELETE, GET, HEAD, OPTIONS, PATCH, POST, and PUT."""
        return self["method"]

    @property
    def http_method(self) -> str:
        """The HTTP method used. Valid values include: DELETE, GET, HEAD, OPTIONS, PATCH, POST, and PUT."""
        return self["method"]

    def get_query_string_value(self, name: str, default_value: Optional[str] = None) -> Optional[str]:
        """Get query string value by name

        Parameters
        ----------
        name: str
            Query string parameter name
        default_value: str, optional
            Default value if no value was found by name
        Returns
        -------
        str, optional
            Query string parameter value
        """
        return get_query_string_value(
            query_string_parameters=self.query_string_parameters,
            name=name,
            default_value=default_value,
        )

    @overload
    def get_header_value(
        self,
        name: str,
        default_value: str,
        case_sensitive: Optional[bool] = False,
    ) -> str:
        ...

    @overload
    def get_header_value(
        self,
        name: str,
        default_value: Optional[str] = None,
        case_sensitive: Optional[bool] = False,
    ) -> Optional[str]:
        ...

    def get_header_value(
        self,
        name: str,
        default_value: Optional[str] = None,
        case_sensitive: Optional[bool] = False,
    ) -> Optional[str]:
        """Get header value by name

        Parameters
        ----------
        name: str
            Header name
        default_value: str, optional
            Default value if no value was found by name
        case_sensitive: bool
            Whether to use a case-sensitive look up
        Returns
        -------
        str, optional
            Header value
        """
        return get_header_value(
            headers=self.headers,
            name=name,
            default_value=default_value,
            case_sensitive=case_sensitive,
        )

    def header_serializer(self) -> BaseHeadersSerializer:
        # When using the VPC Lattice integration, we have multiple HTTP Headers.
        return HttpApiHeadersSerializer()


class VPCLatticeEvent(VPCLatticeEventBase):
    @property
    def raw_path(self) -> str:
        """The raw VPC Lattice request path."""
        return self["raw_path"]

    @property
    def is_base64_encoded(self) -> bool:
        """A boolean flag to indicate if the applicable request payload is Base64-encode"""
        return self["is_base64_encoded"]

    # VPCLattice event has no path field
    # Added here for consistency with the BaseProxyEvent class
    @property
    def path(self) -> str:
        return self["raw_path"]

    @property
    def query_string_parameters(self) -> Dict[str, str]:
        """The request query string parameters."""
        return self["query_string_parameters"]


class vpcLatticeEventV2Identity(DictWrapper):
    @property
    def source_vpc_arn(self) -> Optional[str]:
        """The VPC Lattice v2 Event requestContext Identity sourceVpcArn"""
        return self.get("sourceVpcArn")

    @property
    def get_type(self) -> Optional[str]:
        """The VPC Lattice v2 Event requestContext Identity type"""
        return self.get("type")

    @property
    def principal(self) -> Optional[str]:
        """The VPC Lattice v2 Event requestContext principal"""
        return self.get("principal")

    @property
    def principal_org_id(self) -> Optional[str]:
        """The VPC Lattice v2 Event requestContext principalOrgID"""
        return self.get("principalOrgID")

    @property
    def session_name(self) -> Optional[str]:
        """The VPC Lattice v2 Event requestContext sessionName"""
        return self.get("sessionName")

    @property
    def x509_subject_cn(self) -> Optional[str]:
        """The VPC Lattice v2 Event requestContext X509SubjectCn"""
        return self.get("X509SubjectCn")

    @property
    def x509_issuer_ou(self) -> Optional[str]:
        """The VPC Lattice v2 Event requestContext X509IssuerOu"""
        return self.get("X509IssuerOu")

    @property
    def x509_san_dns(self) -> Optional[str]:
        """The VPC Lattice v2 Event requestContext X509SanDns"""
        return self.get("x509SanDns")

    @property
    def x509_san_uri(self) -> Optional[str]:
        """The VPC Lattice v2 Event requestContext X509SanUri"""
        return self.get("X509SanUri")

    @property
    def x509_san_name_cn(self) -> Optional[str]:
        """The VPC Lattice v2 Event requestContext X509SanNameCn"""
        return self.get("X509SanNameCn")


class vpcLatticeEventV2RequestContext(DictWrapper):
    @property
    def service_network_arn(self) -> str:
        """The VPC Lattice v2 Event requestContext serviceNetworkArn"""
        return self["serviceNetworkArn"]

    @property
    def service_arn(self) -> str:
        """The VPC Lattice v2 Event requestContext serviceArn"""
        return self["serviceArn"]

    @property
    def target_group_arn(self) -> str:
        """The VPC Lattice v2 Event requestContext targetGroupArn"""
        return self["targetGroupArn"]

    @property
    def identity(self) -> vpcLatticeEventV2Identity:
        """The VPC Lattice v2 Event requestContext identity"""
        return vpcLatticeEventV2Identity(self["identity"])

    @property
    def region(self) -> str:
        """The VPC Lattice v2 Event requestContext serviceNetworkArn"""
        return self["region"]

    @property
    def time_epoch(self) -> float:
        """The VPC Lattice v2 Event requestContext timeEpoch"""
        return self["timeEpoch"]


class VPCLatticeEventV2(VPCLatticeEventBase):
    @property
    def version(self) -> str:
        """The VPC Lattice v2 Event version"""
        return self["version"]

    @property
    def is_base64_encoded(self) -> Optional[bool]:
        """A boolean flag to indicate if the applicable request payload is Base64-encode"""
        return self.get("isBase64Encoded")

    @property
    def path(self) -> str:
        """The VPC Lattice v2 Event path"""
        return self["path"]

    @property
    def request_context(self) -> vpcLatticeEventV2RequestContext:
        """he VPC Lattice v2 Event request context."""
        return vpcLatticeEventV2RequestContext(self["requestContext"])

    @property
    def query_string_parameters(self) -> Optional[Dict[str, str]]:
        """The request query string parameters."""
        return self.get("queryStringParameters")

Classes

class VPCLatticeEvent (data: Dict[str, Any], json_deserializer: Optional[Callable] = None)

Provides a single read only access to a wrapper dict

Parameters

data : Dict[str, Any]
Lambda Event Source Event payload
json_deserializer : Callable, optional
function to deserialize str, bytes, bytearray containing a JSON document to a Python obj, by default json.loads
Expand source code 
class VPCLatticeEvent(VPCLatticeEventBase):
    @property
    def raw_path(self) -> str:
        """The raw VPC Lattice request path."""
        return self["raw_path"]

    @property
    def is_base64_encoded(self) -> bool:
        """A boolean flag to indicate if the applicable request payload is Base64-encode"""
        return self["is_base64_encoded"]

    # VPCLattice event has no path field
    # Added here for consistency with the BaseProxyEvent class
    @property
    def path(self) -> str:
        return self["raw_path"]

    @property
    def query_string_parameters(self) -> Dict[str, str]:
        """The request query string parameters."""
        return self["query_string_parameters"]

Ancestors

Instance variables

var is_base64_encoded : bool

A boolean flag to indicate if the applicable request payload is Base64-encode

Expand source code 
@property
def is_base64_encoded(self) -> bool:
    """A boolean flag to indicate if the applicable request payload is Base64-encode"""
    return self["is_base64_encoded"]
var path : str
Expand source code 
@property
def path(self) -> str:
    return self["raw_path"]
var query_string_parameters : Dict[str, str]

The request query string parameters.

Expand source code 
@property
def query_string_parameters(self) -> Dict[str, str]:
    """The request query string parameters."""
    return self["query_string_parameters"]
var raw_path : str

The raw VPC Lattice request path.

Expand source code 
@property
def raw_path(self) -> str:
    """The raw VPC Lattice request path."""
    return self["raw_path"]

Inherited members

class VPCLatticeEventBase (data: Dict[str, Any], json_deserializer: Optional[Callable] = None)

Provides a single read only access to a wrapper dict

Parameters

data : Dict[str, Any]
Lambda Event Source Event payload
json_deserializer : Callable, optional
function to deserialize str, bytes, bytearray containing a JSON document to a Python obj, by default json.loads
Expand source code 
class VPCLatticeEventBase(BaseProxyEvent):
    @property
    def body(self) -> str:
        """The VPC Lattice body."""
        return self["body"]

    @property
    def json_body(self) -> Any:
        """Parses the submitted body as json"""
        if self._json_data is None:
            self._json_data = self._json_deserializer(self.decoded_body)
        return self._json_data

    @property
    def headers(self) -> Dict[str, str]:
        """The VPC Lattice event headers."""
        return self["headers"]

    @property
    def decoded_body(self) -> str:
        """Dynamically base64 decode body as a str"""
        body: str = self["body"]
        if self.is_base64_encoded:
            return base64_decode(body)
        return body

    @property
    def method(self) -> str:
        """The VPC Lattice method used. Valid values include: DELETE, GET, HEAD, OPTIONS, PATCH, POST, and PUT."""
        return self["method"]

    @property
    def http_method(self) -> str:
        """The HTTP method used. Valid values include: DELETE, GET, HEAD, OPTIONS, PATCH, POST, and PUT."""
        return self["method"]

    def get_query_string_value(self, name: str, default_value: Optional[str] = None) -> Optional[str]:
        """Get query string value by name

        Parameters
        ----------
        name: str
            Query string parameter name
        default_value: str, optional
            Default value if no value was found by name
        Returns
        -------
        str, optional
            Query string parameter value
        """
        return get_query_string_value(
            query_string_parameters=self.query_string_parameters,
            name=name,
            default_value=default_value,
        )

    @overload
    def get_header_value(
        self,
        name: str,
        default_value: str,
        case_sensitive: Optional[bool] = False,
    ) -> str:
        ...

    @overload
    def get_header_value(
        self,
        name: str,
        default_value: Optional[str] = None,
        case_sensitive: Optional[bool] = False,
    ) -> Optional[str]:
        ...

    def get_header_value(
        self,
        name: str,
        default_value: Optional[str] = None,
        case_sensitive: Optional[bool] = False,
    ) -> Optional[str]:
        """Get header value by name

        Parameters
        ----------
        name: str
            Header name
        default_value: str, optional
            Default value if no value was found by name
        case_sensitive: bool
            Whether to use a case-sensitive look up
        Returns
        -------
        str, optional
            Header value
        """
        return get_header_value(
            headers=self.headers,
            name=name,
            default_value=default_value,
            case_sensitive=case_sensitive,
        )

    def header_serializer(self) -> BaseHeadersSerializer:
        # When using the VPC Lattice integration, we have multiple HTTP Headers.
        return HttpApiHeadersSerializer()

Ancestors

  • BaseProxyEvent
  • DictWrapper
  • collections.abc.Mapping
  • collections.abc.Collection
  • collections.abc.Sized
  • collections.abc.Iterable
  • collections.abc.Container

Subclasses

Instance variables

var body : str

The VPC Lattice body.

Expand source code 
@property
def body(self) -> str:
    """The VPC Lattice body."""
    return self["body"]
var headers : Dict[str, str]

The VPC Lattice event headers.

Expand source code 
@property
def headers(self) -> Dict[str, str]:
    """The VPC Lattice event headers."""
    return self["headers"]
var method : str

The VPC Lattice method used. Valid values include: DELETE, GET, HEAD, OPTIONS, PATCH, POST, and PUT.

Expand source code 
@property
def method(self) -> str:
    """The VPC Lattice method used. Valid values include: DELETE, GET, HEAD, OPTIONS, PATCH, POST, and PUT."""
    return self["method"]

Methods

def get_header_value(self, name: str, default_value: Optional[str] = None, case_sensitive: Optional[bool] = False) ‑> Optional[str]

Get header value by name

Parameters

name : str
Header name
default_value : str, optional
Default value if no value was found by name
case_sensitive : bool
Whether to use a case-sensitive look up

Returns

str, optional
Header value
Expand source code 
def get_header_value(
    self,
    name: str,
    default_value: Optional[str] = None,
    case_sensitive: Optional[bool] = False,
) -> Optional[str]:
    """Get header value by name

    Parameters
    ----------
    name: str
        Header name
    default_value: str, optional
        Default value if no value was found by name
    case_sensitive: bool
        Whether to use a case-sensitive look up
    Returns
    -------
    str, optional
        Header value
    """
    return get_header_value(
        headers=self.headers,
        name=name,
        default_value=default_value,
        case_sensitive=case_sensitive,
    )
def header_serializer(self) ‑> BaseHeadersSerializer
Expand source code 
def header_serializer(self) -> BaseHeadersSerializer:
    # When using the VPC Lattice integration, we have multiple HTTP Headers.
    return HttpApiHeadersSerializer()

Inherited members

class VPCLatticeEventV2 (data: Dict[str, Any], json_deserializer: Optional[Callable] = None)

Provides a single read only access to a wrapper dict

Parameters

data : Dict[str, Any]
Lambda Event Source Event payload
json_deserializer : Callable, optional
function to deserialize str, bytes, bytearray containing a JSON document to a Python obj, by default json.loads
Expand source code 
class VPCLatticeEventV2(VPCLatticeEventBase):
    @property
    def version(self) -> str:
        """The VPC Lattice v2 Event version"""
        return self["version"]

    @property
    def is_base64_encoded(self) -> Optional[bool]:
        """A boolean flag to indicate if the applicable request payload is Base64-encode"""
        return self.get("isBase64Encoded")

    @property
    def path(self) -> str:
        """The VPC Lattice v2 Event path"""
        return self["path"]

    @property
    def request_context(self) -> vpcLatticeEventV2RequestContext:
        """he VPC Lattice v2 Event request context."""
        return vpcLatticeEventV2RequestContext(self["requestContext"])

    @property
    def query_string_parameters(self) -> Optional[Dict[str, str]]:
        """The request query string parameters."""
        return self.get("queryStringParameters")

Ancestors

Instance variables

var is_base64_encoded : Optional[bool]

A boolean flag to indicate if the applicable request payload is Base64-encode

Expand source code 
@property
def is_base64_encoded(self) -> Optional[bool]:
    """A boolean flag to indicate if the applicable request payload is Base64-encode"""
    return self.get("isBase64Encoded")
var path : str

The VPC Lattice v2 Event path

Expand source code 
@property
def path(self) -> str:
    """The VPC Lattice v2 Event path"""
    return self["path"]
var query_string_parameters : Optional[Dict[str, str]]

The request query string parameters.

Expand source code 
@property
def query_string_parameters(self) -> Optional[Dict[str, str]]:
    """The request query string parameters."""
    return self.get("queryStringParameters")
var request_context : vpcLatticeEventV2RequestContext

he VPC Lattice v2 Event request context.

Expand source code 
@property
def request_context(self) -> vpcLatticeEventV2RequestContext:
    """he VPC Lattice v2 Event request context."""
    return vpcLatticeEventV2RequestContext(self["requestContext"])
var version : str

The VPC Lattice v2 Event version

Expand source code 
@property
def version(self) -> str:
    """The VPC Lattice v2 Event version"""
    return self["version"]

Inherited members

class vpcLatticeEventV2Identity (data: Dict[str, Any], json_deserializer: Optional[Callable] = None)

Provides a single read only access to a wrapper dict

Parameters

data : Dict[str, Any]
Lambda Event Source Event payload
json_deserializer : Callable, optional
function to deserialize str, bytes, bytearray containing a JSON document to a Python obj, by default json.loads
Expand source code 
class vpcLatticeEventV2Identity(DictWrapper):
    @property
    def source_vpc_arn(self) -> Optional[str]:
        """The VPC Lattice v2 Event requestContext Identity sourceVpcArn"""
        return self.get("sourceVpcArn")

    @property
    def get_type(self) -> Optional[str]:
        """The VPC Lattice v2 Event requestContext Identity type"""
        return self.get("type")

    @property
    def principal(self) -> Optional[str]:
        """The VPC Lattice v2 Event requestContext principal"""
        return self.get("principal")

    @property
    def principal_org_id(self) -> Optional[str]:
        """The VPC Lattice v2 Event requestContext principalOrgID"""
        return self.get("principalOrgID")

    @property
    def session_name(self) -> Optional[str]:
        """The VPC Lattice v2 Event requestContext sessionName"""
        return self.get("sessionName")

    @property
    def x509_subject_cn(self) -> Optional[str]:
        """The VPC Lattice v2 Event requestContext X509SubjectCn"""
        return self.get("X509SubjectCn")

    @property
    def x509_issuer_ou(self) -> Optional[str]:
        """The VPC Lattice v2 Event requestContext X509IssuerOu"""
        return self.get("X509IssuerOu")

    @property
    def x509_san_dns(self) -> Optional[str]:
        """The VPC Lattice v2 Event requestContext X509SanDns"""
        return self.get("x509SanDns")

    @property
    def x509_san_uri(self) -> Optional[str]:
        """The VPC Lattice v2 Event requestContext X509SanUri"""
        return self.get("X509SanUri")

    @property
    def x509_san_name_cn(self) -> Optional[str]:
        """The VPC Lattice v2 Event requestContext X509SanNameCn"""
        return self.get("X509SanNameCn")

Ancestors

  • DictWrapper
  • collections.abc.Mapping
  • collections.abc.Collection
  • collections.abc.Sized
  • collections.abc.Iterable
  • collections.abc.Container

Instance variables

var get_type : Optional[str]

The VPC Lattice v2 Event requestContext Identity type

Expand source code 
@property
def get_type(self) -> Optional[str]:
    """The VPC Lattice v2 Event requestContext Identity type"""
    return self.get("type")
var principal : Optional[str]

The VPC Lattice v2 Event requestContext principal

Expand source code 
@property
def principal(self) -> Optional[str]:
    """The VPC Lattice v2 Event requestContext principal"""
    return self.get("principal")
var principal_org_id : Optional[str]

The VPC Lattice v2 Event requestContext principalOrgID

Expand source code 
@property
def principal_org_id(self) -> Optional[str]:
    """The VPC Lattice v2 Event requestContext principalOrgID"""
    return self.get("principalOrgID")
var session_name : Optional[str]

The VPC Lattice v2 Event requestContext sessionName

Expand source code 
@property
def session_name(self) -> Optional[str]:
    """The VPC Lattice v2 Event requestContext sessionName"""
    return self.get("sessionName")
var source_vpc_arn : Optional[str]

The VPC Lattice v2 Event requestContext Identity sourceVpcArn

Expand source code 
@property
def source_vpc_arn(self) -> Optional[str]:
    """The VPC Lattice v2 Event requestContext Identity sourceVpcArn"""
    return self.get("sourceVpcArn")
var x509_issuer_ou : Optional[str]

The VPC Lattice v2 Event requestContext X509IssuerOu

Expand source code 
@property
def x509_issuer_ou(self) -> Optional[str]:
    """The VPC Lattice v2 Event requestContext X509IssuerOu"""
    return self.get("X509IssuerOu")
var x509_san_dns : Optional[str]

The VPC Lattice v2 Event requestContext X509SanDns

Expand source code 
@property
def x509_san_dns(self) -> Optional[str]:
    """The VPC Lattice v2 Event requestContext X509SanDns"""
    return self.get("x509SanDns")
var x509_san_name_cn : Optional[str]

The VPC Lattice v2 Event requestContext X509SanNameCn

Expand source code 
@property
def x509_san_name_cn(self) -> Optional[str]:
    """The VPC Lattice v2 Event requestContext X509SanNameCn"""
    return self.get("X509SanNameCn")
var x509_san_uri : Optional[str]

The VPC Lattice v2 Event requestContext X509SanUri

Expand source code 
@property
def x509_san_uri(self) -> Optional[str]:
    """The VPC Lattice v2 Event requestContext X509SanUri"""
    return self.get("X509SanUri")
var x509_subject_cn : Optional[str]

The VPC Lattice v2 Event requestContext X509SubjectCn

Expand source code 
@property
def x509_subject_cn(self) -> Optional[str]:
    """The VPC Lattice v2 Event requestContext X509SubjectCn"""
    return self.get("X509SubjectCn")

Inherited members

class vpcLatticeEventV2RequestContext (data: Dict[str, Any], json_deserializer: Optional[Callable] = None)

Provides a single read only access to a wrapper dict

Parameters

data : Dict[str, Any]
Lambda Event Source Event payload
json_deserializer : Callable, optional
function to deserialize str, bytes, bytearray containing a JSON document to a Python obj, by default json.loads
Expand source code 
class vpcLatticeEventV2RequestContext(DictWrapper):
    @property
    def service_network_arn(self) -> str:
        """The VPC Lattice v2 Event requestContext serviceNetworkArn"""
        return self["serviceNetworkArn"]

    @property
    def service_arn(self) -> str:
        """The VPC Lattice v2 Event requestContext serviceArn"""
        return self["serviceArn"]

    @property
    def target_group_arn(self) -> str:
        """The VPC Lattice v2 Event requestContext targetGroupArn"""
        return self["targetGroupArn"]

    @property
    def identity(self) -> vpcLatticeEventV2Identity:
        """The VPC Lattice v2 Event requestContext identity"""
        return vpcLatticeEventV2Identity(self["identity"])

    @property
    def region(self) -> str:
        """The VPC Lattice v2 Event requestContext serviceNetworkArn"""
        return self["region"]

    @property
    def time_epoch(self) -> float:
        """The VPC Lattice v2 Event requestContext timeEpoch"""
        return self["timeEpoch"]

Ancestors

  • DictWrapper
  • collections.abc.Mapping
  • collections.abc.Collection
  • collections.abc.Sized
  • collections.abc.Iterable
  • collections.abc.Container

Instance variables

var identity : vpcLatticeEventV2Identity

The VPC Lattice v2 Event requestContext identity

Expand source code 
@property
def identity(self) -> vpcLatticeEventV2Identity:
    """The VPC Lattice v2 Event requestContext identity"""
    return vpcLatticeEventV2Identity(self["identity"])
var region : str

The VPC Lattice v2 Event requestContext serviceNetworkArn

Expand source code 
@property
def region(self) -> str:
    """The VPC Lattice v2 Event requestContext serviceNetworkArn"""
    return self["region"]
var service_arn : str

The VPC Lattice v2 Event requestContext serviceArn

Expand source code 
@property
def service_arn(self) -> str:
    """The VPC Lattice v2 Event requestContext serviceArn"""
    return self["serviceArn"]
var service_network_arn : str

The VPC Lattice v2 Event requestContext serviceNetworkArn

Expand source code 
@property
def service_network_arn(self) -> str:
    """The VPC Lattice v2 Event requestContext serviceNetworkArn"""
    return self["serviceNetworkArn"]
var target_group_arn : str

The VPC Lattice v2 Event requestContext targetGroupArn

Expand source code 
@property
def target_group_arn(self) -> str:
    """The VPC Lattice v2 Event requestContext targetGroupArn"""
    return self["targetGroupArn"]
var time_epoch : float

The VPC Lattice v2 Event requestContext timeEpoch

Expand source code 
@property
def time_epoch(self) -> float:
    """The VPC Lattice v2 Event requestContext timeEpoch"""
    return self["timeEpoch"]

Inherited members