Module aws_lambda_powertools.utilities.data_classes.vpc_lattice

Classes

class VPCLatticeEvent (data: dict[str, Any], json_deserializer: Callable | None = 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"]

    @cached_property
    def resolved_headers_field(self) -> dict[str, Any]:
        return CaseInsensitiveDict((k, v.split(",") if "," in v else v) for k, v in self.headers.items())

Ancestors

• VPCLatticeEventBase
• BaseProxyEvent
• DictWrapper
• collections.abc.Mapping
• collections.abc.Collection
• collections.abc.Sized
• collections.abc.Iterable
• collections.abc.Container
• typing.Generic

Instance variables

prop 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"]

prop path : str

Expand source code

@property
def path(self) -> str:
    return self["raw_path"]

prop 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"]

prop 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

• VPCLatticeEventBase:
  • body
  • decoded_body
  • get
  • get_header_value
  • get_multi_value_query_string_values
  • get_query_string_value
  • headers
  • http_method
  • json_body
  • method
  • raw_event
  • resolved_headers_field
  • resolved_query_string_parameters

class VPCLatticeEventBase (data: dict[str, Any], json_deserializer: Callable | None = 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"]

    @cached_property
    def json_body(self) -> Any:
        """Parses the submitted body as json"""
        return self._json_deserializer(self.decoded_body)

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

    @property
    def decoded_body(self) -> str:
        """Dynamically base64 decode body as a str"""
        body: str = self["body"]
        return base64_decode(body) if self.is_base64_encoded else 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 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
• typing.Generic

Subclasses

• VPCLatticeEvent
• VPCLatticeEventV2

Instance variables

prop body : str

The VPC Lattice body.

Expand source code

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

prop decoded_body : str

Dynamically base64 decode body as a str

Expand source code

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

prop 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 CaseInsensitiveDict(self["headers"])

prop 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 header_serializer(self) ‑> BaseHeadersSerializer

Inherited members

• BaseProxyEvent:
  • get
  • get_header_value
  • get_multi_value_query_string_values
  • get_query_string_value
  • http_method
  • json_body
  • raw_event
  • resolved_headers_field
  • resolved_query_string_parameters

class VPCLatticeEventV2 (data: dict[str, Any], json_deserializer: Callable | None = 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) -> bool | None:
        """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:
        """The VPC Lattice v2 Event request context."""
        return vpcLatticeEventV2RequestContext(self["requestContext"])

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

        For VPC Lattice V2, the queryStringParameters will contain a dict[str, list[str]]
        so to keep compatibility with existing utilities, we merge all the values with a comma.
        """
        params = self.get("queryStringParameters") or {}
        return {k: ",".join(v) for k, v in params.items()}

    @property
    def resolved_headers_field(self) -> dict[str, str]:
        if self.headers is not None:
            return {key.lower(): value for key, value in self.headers.items()}

        return {}

Ancestors

• VPCLatticeEventBase
• BaseProxyEvent
• DictWrapper
• collections.abc.Mapping
• collections.abc.Collection
• collections.abc.Sized
• collections.abc.Iterable
• collections.abc.Container
• typing.Generic

Instance variables

prop is_base64_encoded : bool | None

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

Expand source code

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

prop 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

The request query string parameters.

For VPC Lattice V2, the queryStringParameters will contain a dict[str, list[str]] so to keep compatibility with existing utilities, we merge all the values with a comma.

Expand source code

def __get__(self, instance, owner=None):
    if instance is None:
        return self
    if self.attrname is None:
        raise TypeError(
            "Cannot use cached_property instance without calling __set_name__ on it.")
    try:
        cache = instance.__dict__
    except AttributeError:  # not all objects have __dict__ (e.g. class defines slots)
        msg = (
            f"No '__dict__' attribute on {type(instance).__name__!r} "
            f"instance to cache {self.attrname!r} property."
        )
        raise TypeError(msg) from None
    val = cache.get(self.attrname, _NOT_FOUND)
    if val is _NOT_FOUND:
        val = self.func(instance)
        try:
            cache[self.attrname] = val
        except TypeError:
            msg = (
                f"The '__dict__' attribute on {type(instance).__name__!r} instance "
                f"does not support item assignment for caching {self.attrname!r} property."
            )
            raise TypeError(msg) from None
    return val

prop request_context : vpcLatticeEventV2RequestContext

The VPC Lattice v2 Event request context.

Expand source code

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

prop 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

• VPCLatticeEventBase:
  • body
  • decoded_body
  • get
  • get_header_value
  • get_multi_value_query_string_values
  • get_query_string_value
  • headers
  • http_method
  • json_body
  • method
  • raw_event
  • resolved_headers_field
  • resolved_query_string_parameters

class vpcLatticeEventV2Identity (data: dict[str, Any], json_deserializer: Callable | None = 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) -> str | None:
        """The VPC Lattice v2 Event requestContext Identity sourceVpcArn"""
        return self.get("sourceVpcArn")

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

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

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

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

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

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

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

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

    @property
    def x509_san_name_cn(self) -> str | None:
        """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
• typing.Generic

Instance variables

prop get_type : str | None

The VPC Lattice v2 Event requestContext Identity type

Expand source code

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

prop principal : str | None

The VPC Lattice v2 Event requestContext principal

Expand source code

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

prop principal_org_id : str | None

The VPC Lattice v2 Event requestContext principalOrgID

Expand source code

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

prop session_name : str | None

The VPC Lattice v2 Event requestContext sessionName

Expand source code

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

prop source_vpc_arn : str | None

The VPC Lattice v2 Event requestContext Identity sourceVpcArn

Expand source code

@property
def source_vpc_arn(self) -> str | None:
    """The VPC Lattice v2 Event requestContext Identity sourceVpcArn"""
    return self.get("sourceVpcArn")

prop x509_issuer_ou : str | None

The VPC Lattice v2 Event requestContext X509IssuerOu

Expand source code

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

prop x509_san_dns : str | None

The VPC Lattice v2 Event requestContext X509SanDns

Expand source code

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

prop x509_san_name_cn : str | None

The VPC Lattice v2 Event requestContext X509SanNameCn

Expand source code

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

prop x509_san_uri : str | None

The VPC Lattice v2 Event requestContext X509SanUri

Expand source code

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

prop x509_subject_cn : str | None

The VPC Lattice v2 Event requestContext X509SubjectCn

Expand source code

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

Inherited members

• DictWrapper:
  • get
  • raw_event

class vpcLatticeEventV2RequestContext (data: dict[str, Any], json_deserializer: Callable | None = 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
• typing.Generic

Instance variables

prop 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"])

prop 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"]

prop 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"]

prop 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"]

prop 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"]

prop 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

• DictWrapper:
  • get
  • raw_event