Source code for idem_aws.states.aws.lambda_aws.function_event_invoke_config

"""State module for managing Lambda function asynchronous invocation configuration"""
import copy
from dataclasses import field
from dataclasses import make_dataclass
from typing import Any

__contracts__ = ["resource"]

from typing import Dict


[docs]async def present( hub, ctx, name: str, function_name: str, resource_id: str = None, qualifier: str = None, maximum_retry_attempts: int = None, maximum_event_age_in_seconds: int = None, destination_config: make_dataclass( """A destination configuration for events after they have been sent to a function.""" "DestinationConfigRequest", [ ("OnSuccess", Dict, field(default=None)), ("OnFailure", Dict, field(default=None)), ], ) = None, ) -> Dict[str, Any]: """ Configures options for asynchronous invocation on a function, version, or alias. If a configuration already exists for a function, version, or alias, this operation overwrites it. By default, Lambda retries an asynchronous invocation twice if the function returns an error. It retains events in a queue for up to six hours. When an event fails all processing attempts or stays in the asynchronous invocation queue for too long, Lambda discards it. To retain discarded events, configure a dead-letter queue. To send an invocation record to a queue, topic, function, or event bus, specify a destination. You can configure separate destinations for successful invocations (on-success) and events that fail all processing attempts (on-failure). You can configure destinations in addition to or instead of a dead-letter queue. Args: name(str): An Idem name of the resource. resource_id(str, Optional): The Amazon Resource Name (ARN) of the function. Defaults to None. function_name(str): The name of the Lambda function. The length constraint applies only to the full ARN. If you specify only the function name, it is limited to 64 characters in length. Name formats * Function name - my-function * Function ARN - ``arn:aws:lambda:us-west-2:123456789012:function:my-function`` * Partial ARN - ``123456789012:function:my-function`` qualifier(str, Optional): A version number or alias name. Defaults to None. maximum_retry_attempts(int, Optional): The maximum number of times to retry when the function returns an error. Defaults to None. maximum_event_age_in_seconds(int, Optional): The maximum age of a request that Lambda sends to a function for processing. Defaults to None. destination_config(Dict, Optional): A destination for events after they have been sent to a function for processing. * (OnSuccess, Dict, Optional): The destination configuration for successful invocations. * (OnFailure, Dict, Optional): The destination configuration for failed invocations. Destinations * Function - The Amazon Resource Name (ARN) of a Lambda function. * Queue - The ARN of an SQS queue. * Topic - The ARN of an SNS topic. * Event Bus - The ARN of an Amazon EventBridge event bus. Defaults to None. Returns: Dict[str, Any] Examples: .. code-block:: sls function_event_invoke_config-01234672f3336db8: aws.lambda_aws.function_event_invoke_config.present: - name: value - function_name: value """ result = dict(comment=[], old_state=None, new_state=None, name=name, result=True) before = None resource_updated = False if resource_id: before = await hub.exec.aws.lambda_aws.function_event_invoke_config.get( ctx, name=name, resource_id=resource_id ) if not (before["result"] and before["ret"]): result["comment"] = before["comment"] result["result"] = False return result result["old_state"] = copy.deepcopy(before["ret"]) result["new_state"] = copy.deepcopy(result["old_state"]) desired_resource_parameters = { "qualifier": qualifier, "maximum_retry_attempts": maximum_retry_attempts, "maximum_event_age_in_seconds": maximum_event_age_in_seconds, "destination_config": destination_config, } update_function_event_invoke_config_ret = await hub.tool.aws.lambda_aws.function_event_invoke_config.update_function_event_invoke_config( ctx, function_name=function_name, current_state=result["old_state"], desired_state=desired_resource_parameters, ) if not update_function_event_invoke_config_ret["result"]: result["result"] = False result["comment"] = update_function_event_invoke_config_ret["comment"] return result result["comment"] = ( result["comment"] + update_function_event_invoke_config_ret["comment"] ) resource_updated = bool(update_function_event_invoke_config_ret["ret"]) if ctx.get("test", False) and resource_updated: result["new_state"].update(update_function_event_invoke_config_ret["ret"]) return result else: if ctx.get("test", False): result["new_state"] = hub.tool.aws.test_state_utils.generate_test_state( enforced_state={}, desired_state={ "name": name, "function_name": function_name, "qualifier": qualifier, "maximum_retry_attempts": maximum_retry_attempts, "maximum_event_age_in_seconds": maximum_event_age_in_seconds, "destination_config": destination_config, }, ) result["comment"] = hub.tool.aws.comment_utils.would_create_comment( resource_type="aws.lambda.function_event_invoke_config", name=name ) return result create_ret = await hub.exec.boto3.client[ "lambda" ].put_function_event_invoke_config( ctx, FunctionName=function_name, Qualifier=qualifier, MaximumRetryAttempts=maximum_retry_attempts, MaximumEventAgeInSeconds=maximum_event_age_in_seconds, DestinationConfig=destination_config, ) result["result"] = create_ret["result"] if not result["result"]: result["comment"] = create_ret["comment"] return result result["comment"] = hub.tool.aws.comment_utils.create_comment( resource_type="aws.lambda.function_event_invoke_config", name=name ) if (not before) or resource_updated: after = await hub.exec.aws.lambda_aws.function_event_invoke_config.get( ctx, name=name, function_name=function_name, qualifier=qualifier ) if not (after["result"] and after["ret"]): result["result"] = False result["comment"] += after["comment"] return result result["new_state"] = copy.deepcopy(after["ret"]) return result
[docs]async def absent( hub, ctx, name: str, resource_id: str = None, ) -> Dict[str, Any]: """Deletes the configuration for asynchronous invocation for a function, version, or alias. Args: name(str): An Idem name of the resource. resource_id(str, Optional): The Amazon Resource Name (ARN) of the function. Idem automatically considers this resource being absent if this field is not specified. Returns: Dict[str, Any] Examples: .. code-block:: sls function_event_invoke_config-01234672f3336db8: aws.lambda_aws.function_event_invoke_config.absent: - name: value - resource_id: value """ already_absent_msg = hub.tool.aws.comment_utils.already_absent_comment( resource_type="aws.lambda.function_event_invoke_config", name=name ) result = dict( comment=already_absent_msg, old_state=None, new_state=None, name=name, result=True, ) if not resource_id: return result before = await hub.exec.aws.lambda_aws.function_event_invoke_config.get( ctx, name=name, resource_id=resource_id ) if not before["result"]: result["comment"] = before["comment"] result["result"] = False return result if before["ret"]: result["old_state"] = before["ret"] if ctx.get("test", False): result["comment"] = hub.tool.aws.comment_utils.would_delete_comment( resource_type="aws.lambda.function_event_invoke_config", name=name ) else: function_name = hub.tool.aws.arn_utils.get_resource_name(resource_id) qualifier = hub.tool.aws.arn_utils.get_qualifier(resource_id) ret = await hub.exec.boto3.client[ "lambda" ].delete_function_event_invoke_config( ctx, **{"FunctionName": function_name, "Qualifier": qualifier} ) result["result"] = ret["result"] if not result["result"]: result["comment"] = ret["comment"] return result result["comment"] = hub.tool.aws.comment_utils.delete_comment( resource_type="aws.lambda.function_event_invoke_config", name=name ) return result
[docs]async def describe(hub, ctx) -> Dict[str, Dict[str, Any]]: """Retrieves a list of configurations for asynchronous invocation for a function. Returns: Dict[str, Dict[str, Any]] Examples: .. code-block:: bash $ idem describe aws.lambda_aws.function_event_invoke_config """ result = {} get_functions_ret = await hub.exec.boto3.client["lambda"].list_functions(ctx) if not get_functions_ret["result"]: hub.log.warning(f"Could not list functions {get_functions_ret['comment']}") return result for function in get_functions_ret["ret"]["Functions"]: function_name = function.get("FunctionName") get_event_invoke_configs_ret = await hub.exec.boto3.client[ "lambda" ].list_function_event_invoke_configs(ctx, FunctionName=function_name) if not get_event_invoke_configs_ret["result"]: hub.log.debug( f"Could not get event invoke config for functionName '{function_name}': " f"{get_event_invoke_configs_ret['comment']}. Describe will skip this function and continue." ) continue for function_event_invoke_config in get_event_invoke_configs_ret["ret"][ "FunctionEventInvokeConfigs" ]: resource_arn = function_event_invoke_config.get("FunctionArn") # Including fields to match the 'present' function parameters resource_translated = hub.tool.aws.lambda_aws.function_event_invoke_config.convert_raw_function_event_invoke_config_to_present( function_event_invoke_config, function_name=function_name, ) result[resource_arn] = { "aws.lambda.function_event_invoke_config.present": [ {parameter_key: parameter_value} for parameter_key, parameter_value in resource_translated.items() ] } return result