Module aws_lambda_powertools.utilities.data_classes.connect_contact_flow_event
Expand source code
from enum import Enum, auto
from typing import Dict, Optional
from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
class ConnectContactFlowChannel(Enum):
VOICE = auto()
CHAT = auto()
class ConnectContactFlowEndpointType(Enum):
TELEPHONE_NUMBER = auto()
class ConnectContactFlowInitiationMethod(Enum):
INBOUND = auto()
OUTBOUND = auto()
TRANSFER = auto()
CALLBACK = auto()
API = auto()
class ConnectContactFlowEndpoint(DictWrapper):
@property
def address(self) -> str:
"""The phone number."""
return self["Address"]
@property
def endpoint_type(self) -> ConnectContactFlowEndpointType:
"""The endpoint type."""
return ConnectContactFlowEndpointType[self["Type"]]
class ConnectContactFlowQueue(DictWrapper):
@property
def arn(self) -> str:
"""The unique queue ARN."""
return self["ARN"]
@property
def name(self) -> str:
"""The queue name."""
return self["Name"]
class ConnectContactFlowMediaStreamAudio(DictWrapper):
@property
def start_fragment_number(self) -> Optional[str]:
"""The number that identifies the Kinesis Video Streams fragment, in the stream used for Live media streaming,
in which the customer audio stream started.
"""
return self["StartFragmentNumber"]
@property
def start_timestamp(self) -> Optional[str]:
"""When the customer audio stream started."""
return self["StartTimestamp"]
@property
def stream_arn(self) -> Optional[str]:
"""The ARN of the Kinesis Video stream used for Live media streaming that includes the customer data to
reference.
"""
return self["StreamARN"]
class ConnectContactFlowMediaStreamCustomer(DictWrapper):
@property
def audio(self) -> ConnectContactFlowMediaStreamAudio:
return ConnectContactFlowMediaStreamAudio(self["Audio"])
class ConnectContactFlowMediaStreams(DictWrapper):
@property
def customer(self) -> ConnectContactFlowMediaStreamCustomer:
return ConnectContactFlowMediaStreamCustomer(self["Customer"])
class ConnectContactFlowData(DictWrapper):
@property
def attributes(self) -> Dict[str, str]:
"""These are attributes that have been previously associated with a contact,
such as when using a Set contact attributes block in a contact flow.
This map may be empty if there aren't any saved attributes.
"""
return self["Attributes"]
@property
def channel(self) -> ConnectContactFlowChannel:
"""The method used to contact your contact center."""
return ConnectContactFlowChannel[self["Channel"]]
@property
def contact_id(self) -> str:
"""The unique identifier of the contact."""
return self["ContactId"]
@property
def customer_endpoint(self) -> Optional[ConnectContactFlowEndpoint]:
"""Contains the customer’s address (number) and type of address."""
if self["CustomerEndpoint"] is not None:
return ConnectContactFlowEndpoint(self["CustomerEndpoint"])
return None
@property
def initial_contact_id(self) -> str:
"""The unique identifier for the contact associated with the first interaction between the customer and your
contact center. Use the initial contact ID to track contacts between contact flows.
"""
return self["InitialContactId"]
@property
def initiation_method(self) -> ConnectContactFlowInitiationMethod:
"""How the contact was initiated."""
return ConnectContactFlowInitiationMethod[self["InitiationMethod"]]
@property
def instance_arn(self) -> str:
"""The ARN for your Amazon Connect instance."""
return self["InstanceARN"]
@property
def previous_contact_id(self) -> str:
"""The unique identifier for the contact before it was transferred.
Use the previous contact ID to trace contacts between contact flows.
"""
return self["PreviousContactId"]
@property
def queue(self) -> Optional[ConnectContactFlowQueue]:
"""The current queue."""
if self["Queue"] is not None:
return ConnectContactFlowQueue(self["Queue"])
return None
@property
def system_endpoint(self) -> Optional[ConnectContactFlowEndpoint]:
"""Contains the address (number) the customer dialed to call your contact center and type of address."""
if self["SystemEndpoint"] is not None:
return ConnectContactFlowEndpoint(self["SystemEndpoint"])
return None
@property
def media_streams(self) -> ConnectContactFlowMediaStreams:
return ConnectContactFlowMediaStreams(self["MediaStreams"])
class ConnectContactFlowEvent(DictWrapper):
"""Amazon Connect contact flow event
Documentation:
-------------
- https://docs.aws.amazon.com/connect/latest/adminguide/connect-lambda-functions.html
"""
@property
def contact_data(self) -> ConnectContactFlowData:
"""This is always passed by Amazon Connect for every contact. Some parameters are optional."""
return ConnectContactFlowData(self["Details"]["ContactData"])
@property
def parameters(self) -> Dict[str, str]:
"""These are parameters specific to this call that were defined when you created the Lambda function."""
return self["Details"]["Parameters"]
Classes
class ConnectContactFlowChannel (*args, **kwds)
-
Create a collection of name/value pairs.
Example enumeration:
>>> class Color(Enum): ... RED = 1 ... BLUE = 2 ... GREEN = 3
Access them by:
- attribute access:
Color.RED
- value lookup:
Color(1)
- name lookup:
Color['RED']
Enumerations can be iterated over, and know how many members they have:
>>> len(Color) 3
>>> list(Color) [<Color.RED: 1>, <Color.BLUE: 2>, <Color.GREEN: 3>]
Methods can be added to enumerations, and members can have their own attributes – see the documentation for details.
Expand source code
class ConnectContactFlowChannel(Enum): VOICE = auto() CHAT = auto()
Ancestors
- enum.Enum
Class variables
var CHAT
var VOICE
class ConnectContactFlowData (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 Pythonobj
, by default json.loads
Expand source code
class ConnectContactFlowData(DictWrapper): @property def attributes(self) -> Dict[str, str]: """These are attributes that have been previously associated with a contact, such as when using a Set contact attributes block in a contact flow. This map may be empty if there aren't any saved attributes. """ return self["Attributes"] @property def channel(self) -> ConnectContactFlowChannel: """The method used to contact your contact center.""" return ConnectContactFlowChannel[self["Channel"]] @property def contact_id(self) -> str: """The unique identifier of the contact.""" return self["ContactId"] @property def customer_endpoint(self) -> Optional[ConnectContactFlowEndpoint]: """Contains the customer’s address (number) and type of address.""" if self["CustomerEndpoint"] is not None: return ConnectContactFlowEndpoint(self["CustomerEndpoint"]) return None @property def initial_contact_id(self) -> str: """The unique identifier for the contact associated with the first interaction between the customer and your contact center. Use the initial contact ID to track contacts between contact flows. """ return self["InitialContactId"] @property def initiation_method(self) -> ConnectContactFlowInitiationMethod: """How the contact was initiated.""" return ConnectContactFlowInitiationMethod[self["InitiationMethod"]] @property def instance_arn(self) -> str: """The ARN for your Amazon Connect instance.""" return self["InstanceARN"] @property def previous_contact_id(self) -> str: """The unique identifier for the contact before it was transferred. Use the previous contact ID to trace contacts between contact flows. """ return self["PreviousContactId"] @property def queue(self) -> Optional[ConnectContactFlowQueue]: """The current queue.""" if self["Queue"] is not None: return ConnectContactFlowQueue(self["Queue"]) return None @property def system_endpoint(self) -> Optional[ConnectContactFlowEndpoint]: """Contains the address (number) the customer dialed to call your contact center and type of address.""" if self["SystemEndpoint"] is not None: return ConnectContactFlowEndpoint(self["SystemEndpoint"]) return None @property def media_streams(self) -> ConnectContactFlowMediaStreams: return ConnectContactFlowMediaStreams(self["MediaStreams"])
Ancestors
- DictWrapper
- collections.abc.Mapping
- collections.abc.Collection
- collections.abc.Sized
- collections.abc.Iterable
- collections.abc.Container
Instance variables
var attributes : Dict[str, str]
-
These are attributes that have been previously associated with a contact, such as when using a Set contact attributes block in a contact flow. This map may be empty if there aren't any saved attributes.
Expand source code
@property def attributes(self) -> Dict[str, str]: """These are attributes that have been previously associated with a contact, such as when using a Set contact attributes block in a contact flow. This map may be empty if there aren't any saved attributes. """ return self["Attributes"]
var channel : ConnectContactFlowChannel
-
The method used to contact your contact center.
Expand source code
@property def channel(self) -> ConnectContactFlowChannel: """The method used to contact your contact center.""" return ConnectContactFlowChannel[self["Channel"]]
var contact_id : str
-
The unique identifier of the contact.
Expand source code
@property def contact_id(self) -> str: """The unique identifier of the contact.""" return self["ContactId"]
var customer_endpoint : Optional[ConnectContactFlowEndpoint]
-
Contains the customer’s address (number) and type of address.
Expand source code
@property def customer_endpoint(self) -> Optional[ConnectContactFlowEndpoint]: """Contains the customer’s address (number) and type of address.""" if self["CustomerEndpoint"] is not None: return ConnectContactFlowEndpoint(self["CustomerEndpoint"]) return None
var initial_contact_id : str
-
The unique identifier for the contact associated with the first interaction between the customer and your contact center. Use the initial contact ID to track contacts between contact flows.
Expand source code
@property def initial_contact_id(self) -> str: """The unique identifier for the contact associated with the first interaction between the customer and your contact center. Use the initial contact ID to track contacts between contact flows. """ return self["InitialContactId"]
var initiation_method : ConnectContactFlowInitiationMethod
-
How the contact was initiated.
Expand source code
@property def initiation_method(self) -> ConnectContactFlowInitiationMethod: """How the contact was initiated.""" return ConnectContactFlowInitiationMethod[self["InitiationMethod"]]
var instance_arn : str
-
The ARN for your Amazon Connect instance.
Expand source code
@property def instance_arn(self) -> str: """The ARN for your Amazon Connect instance.""" return self["InstanceARN"]
var media_streams : ConnectContactFlowMediaStreams
-
Expand source code
@property def media_streams(self) -> ConnectContactFlowMediaStreams: return ConnectContactFlowMediaStreams(self["MediaStreams"])
var previous_contact_id : str
-
The unique identifier for the contact before it was transferred. Use the previous contact ID to trace contacts between contact flows.
Expand source code
@property def previous_contact_id(self) -> str: """The unique identifier for the contact before it was transferred. Use the previous contact ID to trace contacts between contact flows. """ return self["PreviousContactId"]
var queue : Optional[ConnectContactFlowQueue]
-
The current queue.
Expand source code
@property def queue(self) -> Optional[ConnectContactFlowQueue]: """The current queue.""" if self["Queue"] is not None: return ConnectContactFlowQueue(self["Queue"]) return None
var system_endpoint : Optional[ConnectContactFlowEndpoint]
-
Contains the address (number) the customer dialed to call your contact center and type of address.
Expand source code
@property def system_endpoint(self) -> Optional[ConnectContactFlowEndpoint]: """Contains the address (number) the customer dialed to call your contact center and type of address.""" if self["SystemEndpoint"] is not None: return ConnectContactFlowEndpoint(self["SystemEndpoint"]) return None
Inherited members
class ConnectContactFlowEndpoint (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 Pythonobj
, by default json.loads
Expand source code
class ConnectContactFlowEndpoint(DictWrapper): @property def address(self) -> str: """The phone number.""" return self["Address"] @property def endpoint_type(self) -> ConnectContactFlowEndpointType: """The endpoint type.""" return ConnectContactFlowEndpointType[self["Type"]]
Ancestors
- DictWrapper
- collections.abc.Mapping
- collections.abc.Collection
- collections.abc.Sized
- collections.abc.Iterable
- collections.abc.Container
Instance variables
var address : str
-
The phone number.
Expand source code
@property def address(self) -> str: """The phone number.""" return self["Address"]
var endpoint_type : ConnectContactFlowEndpointType
-
The endpoint type.
Expand source code
@property def endpoint_type(self) -> ConnectContactFlowEndpointType: """The endpoint type.""" return ConnectContactFlowEndpointType[self["Type"]]
Inherited members
class ConnectContactFlowEndpointType (*args, **kwds)
-
Create a collection of name/value pairs.
Example enumeration:
>>> class Color(Enum): ... RED = 1 ... BLUE = 2 ... GREEN = 3
Access them by:
- attribute access:
Color.RED
- value lookup:
Color(1)
- name lookup:
Color['RED']
Enumerations can be iterated over, and know how many members they have:
>>> len(Color) 3
>>> list(Color) [<Color.RED: 1>, <Color.BLUE: 2>, <Color.GREEN: 3>]
Methods can be added to enumerations, and members can have their own attributes – see the documentation for details.
Expand source code
class ConnectContactFlowEndpointType(Enum): TELEPHONE_NUMBER = auto()
Ancestors
- enum.Enum
Class variables
var TELEPHONE_NUMBER
class ConnectContactFlowEvent (data: Dict[str, Any], json_deserializer: Optional[Callable] = None)
-
Amazon Connect contact flow event
Documentation:
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 Pythonobj
, by default json.loads
Expand source code
class ConnectContactFlowEvent(DictWrapper): """Amazon Connect contact flow event Documentation: ------------- - https://docs.aws.amazon.com/connect/latest/adminguide/connect-lambda-functions.html """ @property def contact_data(self) -> ConnectContactFlowData: """This is always passed by Amazon Connect for every contact. Some parameters are optional.""" return ConnectContactFlowData(self["Details"]["ContactData"]) @property def parameters(self) -> Dict[str, str]: """These are parameters specific to this call that were defined when you created the Lambda function.""" return self["Details"]["Parameters"]
Ancestors
- DictWrapper
- collections.abc.Mapping
- collections.abc.Collection
- collections.abc.Sized
- collections.abc.Iterable
- collections.abc.Container
Instance variables
var contact_data : ConnectContactFlowData
-
This is always passed by Amazon Connect for every contact. Some parameters are optional.
Expand source code
@property def contact_data(self) -> ConnectContactFlowData: """This is always passed by Amazon Connect for every contact. Some parameters are optional.""" return ConnectContactFlowData(self["Details"]["ContactData"])
var parameters : Dict[str, str]
-
These are parameters specific to this call that were defined when you created the Lambda function.
Expand source code
@property def parameters(self) -> Dict[str, str]: """These are parameters specific to this call that were defined when you created the Lambda function.""" return self["Details"]["Parameters"]
Inherited members
class ConnectContactFlowInitiationMethod (*args, **kwds)
-
Create a collection of name/value pairs.
Example enumeration:
>>> class Color(Enum): ... RED = 1 ... BLUE = 2 ... GREEN = 3
Access them by:
- attribute access:
Color.RED
- value lookup:
Color(1)
- name lookup:
Color['RED']
Enumerations can be iterated over, and know how many members they have:
>>> len(Color) 3
>>> list(Color) [<Color.RED: 1>, <Color.BLUE: 2>, <Color.GREEN: 3>]
Methods can be added to enumerations, and members can have their own attributes – see the documentation for details.
Expand source code
class ConnectContactFlowInitiationMethod(Enum): INBOUND = auto() OUTBOUND = auto() TRANSFER = auto() CALLBACK = auto() API = auto()
Ancestors
- enum.Enum
Class variables
var API
var CALLBACK
var INBOUND
var OUTBOUND
var TRANSFER
class ConnectContactFlowMediaStreamAudio (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 Pythonobj
, by default json.loads
Expand source code
class ConnectContactFlowMediaStreamAudio(DictWrapper): @property def start_fragment_number(self) -> Optional[str]: """The number that identifies the Kinesis Video Streams fragment, in the stream used for Live media streaming, in which the customer audio stream started. """ return self["StartFragmentNumber"] @property def start_timestamp(self) -> Optional[str]: """When the customer audio stream started.""" return self["StartTimestamp"] @property def stream_arn(self) -> Optional[str]: """The ARN of the Kinesis Video stream used for Live media streaming that includes the customer data to reference. """ return self["StreamARN"]
Ancestors
- DictWrapper
- collections.abc.Mapping
- collections.abc.Collection
- collections.abc.Sized
- collections.abc.Iterable
- collections.abc.Container
Instance variables
var start_fragment_number : Optional[str]
-
The number that identifies the Kinesis Video Streams fragment, in the stream used for Live media streaming, in which the customer audio stream started.
Expand source code
@property def start_fragment_number(self) -> Optional[str]: """The number that identifies the Kinesis Video Streams fragment, in the stream used for Live media streaming, in which the customer audio stream started. """ return self["StartFragmentNumber"]
var start_timestamp : Optional[str]
-
When the customer audio stream started.
Expand source code
@property def start_timestamp(self) -> Optional[str]: """When the customer audio stream started.""" return self["StartTimestamp"]
var stream_arn : Optional[str]
-
The ARN of the Kinesis Video stream used for Live media streaming that includes the customer data to reference.
Expand source code
@property def stream_arn(self) -> Optional[str]: """The ARN of the Kinesis Video stream used for Live media streaming that includes the customer data to reference. """ return self["StreamARN"]
Inherited members
class ConnectContactFlowMediaStreamCustomer (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 Pythonobj
, by default json.loads
Expand source code
class ConnectContactFlowMediaStreamCustomer(DictWrapper): @property def audio(self) -> ConnectContactFlowMediaStreamAudio: return ConnectContactFlowMediaStreamAudio(self["Audio"])
Ancestors
- DictWrapper
- collections.abc.Mapping
- collections.abc.Collection
- collections.abc.Sized
- collections.abc.Iterable
- collections.abc.Container
Instance variables
var audio : ConnectContactFlowMediaStreamAudio
-
Expand source code
@property def audio(self) -> ConnectContactFlowMediaStreamAudio: return ConnectContactFlowMediaStreamAudio(self["Audio"])
Inherited members
class ConnectContactFlowMediaStreams (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 Pythonobj
, by default json.loads
Expand source code
class ConnectContactFlowMediaStreams(DictWrapper): @property def customer(self) -> ConnectContactFlowMediaStreamCustomer: return ConnectContactFlowMediaStreamCustomer(self["Customer"])
Ancestors
- DictWrapper
- collections.abc.Mapping
- collections.abc.Collection
- collections.abc.Sized
- collections.abc.Iterable
- collections.abc.Container
Instance variables
var customer : ConnectContactFlowMediaStreamCustomer
-
Expand source code
@property def customer(self) -> ConnectContactFlowMediaStreamCustomer: return ConnectContactFlowMediaStreamCustomer(self["Customer"])
Inherited members
class ConnectContactFlowQueue (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 Pythonobj
, by default json.loads
Expand source code
class ConnectContactFlowQueue(DictWrapper): @property def arn(self) -> str: """The unique queue ARN.""" return self["ARN"] @property def name(self) -> str: """The queue name.""" return self["Name"]
Ancestors
- DictWrapper
- collections.abc.Mapping
- collections.abc.Collection
- collections.abc.Sized
- collections.abc.Iterable
- collections.abc.Container
Instance variables
var arn : str
-
The unique queue ARN.
Expand source code
@property def arn(self) -> str: """The unique queue ARN.""" return self["ARN"]
var name : str
-
The queue name.
Expand source code
@property def name(self) -> str: """The queue name.""" return self["Name"]
Inherited members