Ejemplos de Lambda que se utilizan para SDK Python (Boto3) - Ejemplos de código de AWS SDK

Hay más AWS SDK ejemplos disponibles en el GitHub repositorio de AWS Doc SDK Examples.

Las traducciones son generadas a través de traducción automática. En caso de conflicto entre la traducción y la version original de inglés, prevalecerá la version en inglés.

Ejemplos de Lambda que se utilizan para SDK Python (Boto3)

Los siguientes ejemplos de código muestran cómo realizar acciones e implementar escenarios comunes AWS SDK for Python (Boto3) mediante Lambda.

Las acciones son extractos de código de programas más grandes y deben ejecutarse en contexto. Mientras las acciones muestran cómo llamar a las funciones de servicio individuales, es posible ver las acciones en contexto en los escenarios relacionados y en los ejemplos entre servicios.

Los escenarios son ejemplos de código que muestran cómo llevar a cabo una tarea específica llamando a varias funciones dentro del mismo servicio.

Cada ejemplo incluye un enlace a GitHub, donde puede encontrar instrucciones sobre cómo configurar y ejecutar el código en su contexto.

Introducción

En los siguientes ejemplos de código se muestra cómo empezar a utilizar Lambda.

SDKpara Python (Boto3)
nota

Hay más información. GitHub Busque el ejemplo completo y aprenda a configurar y ejecutar en el Repositorio de ejemplos de código de AWS.

import boto3 def main(): """ List the Lambda functions in your AWS account. """ # Create the Lambda client lambda_client = boto3.client("lambda") # Use the paginator to list the functions paginator = lambda_client.get_paginator("list_functions") response_iterator = paginator.paginate() print("Here are the Lambda functions in your account:") for page in response_iterator: for function in page["Functions"]: print(f" {function['FunctionName']}") if __name__ == "__main__": main()
  • Para API obtener más información, consulte ListFunctionsla AWS SDKreferencia de Python (Boto3). API

Acciones

En el siguiente ejemplo de código, se muestra cómo usar CreateFunction.

SDKpara Python (Boto3)
nota

Hay más información. GitHub Busque el ejemplo completo y aprenda a configurar y ejecutar en el Repositorio de ejemplos de código de AWS.

class LambdaWrapper: def __init__(self, lambda_client, iam_resource): self.lambda_client = lambda_client self.iam_resource = iam_resource def create_function( self, function_name, handler_name, iam_role, deployment_package ): """ Deploys a Lambda function. :param function_name: The name of the Lambda function. :param handler_name: The fully qualified name of the handler function. This must include the file name and the function name. :param iam_role: The IAM role to use for the function. :param deployment_package: The deployment package that contains the function code in .zip format. :return: The Amazon Resource Name (ARN) of the newly created function. """ try: response = self.lambda_client.create_function( FunctionName=function_name, Description="AWS Lambda doc example", Runtime="python3.8", Role=iam_role.arn, Handler=handler_name, Code={"ZipFile": deployment_package}, Publish=True, ) function_arn = response["FunctionArn"] waiter = self.lambda_client.get_waiter("function_active_v2") waiter.wait(FunctionName=function_name) logger.info( "Created function '%s' with ARN: '%s'.", function_name, response["FunctionArn"], ) except ClientError: logger.error("Couldn't create function %s.", function_name) raise else: return function_arn
  • Para API obtener más información, consulte CreateFunctionla AWS SDKreferencia de Python (Boto3). API

En el siguiente ejemplo de código, se muestra cómo usar DeleteFunction.

SDKpara Python (Boto3)
nota

Hay más información. GitHub Busque el ejemplo completo y aprenda a configurar y ejecutar en el Repositorio de ejemplos de código de AWS.

class LambdaWrapper: def __init__(self, lambda_client, iam_resource): self.lambda_client = lambda_client self.iam_resource = iam_resource def delete_function(self, function_name): """ Deletes a Lambda function. :param function_name: The name of the function to delete. """ try: self.lambda_client.delete_function(FunctionName=function_name) except ClientError: logger.exception("Couldn't delete function %s.", function_name) raise
  • Para API obtener más información, consulte DeleteFunctionla AWS SDKreferencia de Python (Boto3). API

En el siguiente ejemplo de código, se muestra cómo usar GetFunction.

SDKpara Python (Boto3)
nota

Hay más información. GitHub Busque el ejemplo completo y aprenda a configurar y ejecutar en el Repositorio de ejemplos de código de AWS.

class LambdaWrapper: def __init__(self, lambda_client, iam_resource): self.lambda_client = lambda_client self.iam_resource = iam_resource def get_function(self, function_name): """ Gets data about a Lambda function. :param function_name: The name of the function. :return: The function data. """ response = None try: response = self.lambda_client.get_function(FunctionName=function_name) except ClientError as err: if err.response["Error"]["Code"] == "ResourceNotFoundException": logger.info("Function %s does not exist.", function_name) else: logger.error( "Couldn't get function %s. Here's why: %s: %s", function_name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise return response
  • Para API obtener más información, consulte GetFunctionla AWS SDKreferencia de Python (Boto3). API

En el siguiente ejemplo de código, se muestra cómo usar Invoke.

SDKpara Python (Boto3)
nota

Hay más información. GitHub Busque el ejemplo completo y aprenda a configurar y ejecutar en el Repositorio de ejemplos de código de AWS.

class LambdaWrapper: def __init__(self, lambda_client, iam_resource): self.lambda_client = lambda_client self.iam_resource = iam_resource def invoke_function(self, function_name, function_params, get_log=False): """ Invokes a Lambda function. :param function_name: The name of the function to invoke. :param function_params: The parameters of the function as a dict. This dict is serialized to JSON before it is sent to Lambda. :param get_log: When true, the last 4 KB of the execution log are included in the response. :return: The response from the function invocation. """ try: response = self.lambda_client.invoke( FunctionName=function_name, Payload=json.dumps(function_params), LogType="Tail" if get_log else "None", ) logger.info("Invoked function %s.", function_name) except ClientError: logger.exception("Couldn't invoke function %s.", function_name) raise return response
  • Para API obtener más información, consulte Invoke in AWS SDKfor Python (Boto3) Reference. API

En el siguiente ejemplo de código, se muestra cómo usar ListFunctions.

SDKpara Python (Boto3)
nota

Hay más información. GitHub Busque el ejemplo completo y aprenda a configurar y ejecutar en el Repositorio de ejemplos de código de AWS.

class LambdaWrapper: def __init__(self, lambda_client, iam_resource): self.lambda_client = lambda_client self.iam_resource = iam_resource def list_functions(self): """ Lists the Lambda functions for the current account. """ try: func_paginator = self.lambda_client.get_paginator("list_functions") for func_page in func_paginator.paginate(): for func in func_page["Functions"]: print(func["FunctionName"]) desc = func.get("Description") if desc: print(f"\t{desc}") print(f"\t{func['Runtime']}: {func['Handler']}") except ClientError as err: logger.error( "Couldn't list functions. Here's why: %s: %s", err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise
  • Para API obtener más información, consulte ListFunctionsla AWS SDKreferencia de Python (Boto3). API

En el siguiente ejemplo de código, se muestra cómo usar UpdateFunctionCode.

SDKpara Python (Boto3)
nota

Hay más información. GitHub Busque el ejemplo completo y aprenda a configurar y ejecutar en el Repositorio de ejemplos de código de AWS.

class LambdaWrapper: def __init__(self, lambda_client, iam_resource): self.lambda_client = lambda_client self.iam_resource = iam_resource def update_function_code(self, function_name, deployment_package): """ Updates the code for a Lambda function by submitting a .zip archive that contains the code for the function. :param function_name: The name of the function to update. :param deployment_package: The function code to update, packaged as bytes in .zip format. :return: Data about the update, including the status. """ try: response = self.lambda_client.update_function_code( FunctionName=function_name, ZipFile=deployment_package ) except ClientError as err: logger.error( "Couldn't update function %s. Here's why: %s: %s", function_name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return response
  • Para API obtener más información, consulte UpdateFunctionCodela AWS SDKreferencia de Python (Boto3). API

En el siguiente ejemplo de código, se muestra cómo usar UpdateFunctionConfiguration.

SDKpara Python (Boto3)
nota

Hay más información. GitHub Busque el ejemplo completo y aprenda a configurar y ejecutar en el Repositorio de ejemplos de código de AWS.

class LambdaWrapper: def __init__(self, lambda_client, iam_resource): self.lambda_client = lambda_client self.iam_resource = iam_resource def update_function_configuration(self, function_name, env_vars): """ Updates the environment variables for a Lambda function. :param function_name: The name of the function to update. :param env_vars: A dict of environment variables to update. :return: Data about the update, including the status. """ try: response = self.lambda_client.update_function_configuration( FunctionName=function_name, Environment={"Variables": env_vars} ) except ClientError as err: logger.error( "Couldn't update function configuration %s. Here's why: %s: %s", function_name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return response

Escenarios

En el siguiente ejemplo de código, se muestra cómo:

  • Cree un IAM rol y una función Lambda y, a continuación, cargue el código del controlador.

  • Invocar la función con un único parámetro y obtener resultados.

  • Actualizar el código de la función y configurar con una variable de entorno.

  • Invocar la función con un nuevo parámetro y obtener resultados. Mostrar el registro de ejecución devuelto.

  • Enumerar las funciones de su cuenta y, luego, limpiar los recursos.

Para obtener información, consulte Crear una función de Lambda con la consola.

SDKpara Python (Boto3)
nota

Hay más información. GitHub Busque el ejemplo completo y aprenda a configurar y ejecutar en el Repositorio de ejemplos de código de AWS.

Defina un controlador de Lambda que aumente un número.

import logging logger = logging.getLogger() logger.setLevel(logging.INFO) def lambda_handler(event, context): """ Accepts an action and a single number, performs the specified action on the number, and returns the result. The only allowable action is 'increment'. :param event: The event dict that contains the parameters sent when the function is invoked. :param context: The context in which the function is called. :return: The result of the action. """ result = None action = event.get("action") if action == "increment": result = event.get("number", 0) + 1 logger.info("Calculated result of %s", result) else: logger.error("%s is not a valid action.", action) response = {"result": result} return response

Defina un segundo controlador Lambda que realice operaciones aritméticas.

import logging import os logger = logging.getLogger() # Define a list of Python lambda functions that are called by this AWS Lambda function. ACTIONS = { "plus": lambda x, y: x + y, "minus": lambda x, y: x - y, "times": lambda x, y: x * y, "divided-by": lambda x, y: x / y, } def lambda_handler(event, context): """ Accepts an action and two numbers, performs the specified action on the numbers, and returns the result. :param event: The event dict that contains the parameters sent when the function is invoked. :param context: The context in which the function is called. :return: The result of the specified action. """ # Set the log level based on a variable configured in the Lambda environment. logger.setLevel(os.environ.get("LOG_LEVEL", logging.INFO)) logger.debug("Event: %s", event) action = event.get("action") func = ACTIONS.get(action) x = event.get("x") y = event.get("y") result = None try: if func is not None and x is not None and y is not None: result = func(x, y) logger.info("%s %s %s is %s", x, action, y, result) else: logger.error("I can't calculate %s %s %s.", x, action, y) except ZeroDivisionError: logger.warning("I can't divide %s by 0!", x) response = {"result": result} return response

Cree funciones que ajusten las acciones de Lambda.

class LambdaWrapper: def __init__(self, lambda_client, iam_resource): self.lambda_client = lambda_client self.iam_resource = iam_resource @staticmethod def create_deployment_package(source_file, destination_file): """ Creates a Lambda deployment package in .zip format in an in-memory buffer. This buffer can be passed directly to Lambda when creating the function. :param source_file: The name of the file that contains the Lambda handler function. :param destination_file: The name to give the file when it's deployed to Lambda. :return: The deployment package. """ buffer = io.BytesIO() with zipfile.ZipFile(buffer, "w") as zipped: zipped.write(source_file, destination_file) buffer.seek(0) return buffer.read() def get_iam_role(self, iam_role_name): """ Get an AWS Identity and Access Management (IAM) role. :param iam_role_name: The name of the role to retrieve. :return: The IAM role. """ role = None try: temp_role = self.iam_resource.Role(iam_role_name) temp_role.load() role = temp_role logger.info("Got IAM role %s", role.name) except ClientError as err: if err.response["Error"]["Code"] == "NoSuchEntity": logger.info("IAM role %s does not exist.", iam_role_name) else: logger.error( "Couldn't get IAM role %s. Here's why: %s: %s", iam_role_name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise return role def create_iam_role_for_lambda(self, iam_role_name): """ Creates an IAM role that grants the Lambda function basic permissions. If a role with the specified name already exists, it is used for the demo. :param iam_role_name: The name of the role to create. :return: The role and a value that indicates whether the role is newly created. """ role = self.get_iam_role(iam_role_name) if role is not None: return role, False lambda_assume_role_policy = { "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Principal": {"Service": "lambda.amazonaws.com"}, "Action": "sts:AssumeRole", } ], } policy_arn = "arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole" try: role = self.iam_resource.create_role( RoleName=iam_role_name, AssumeRolePolicyDocument=json.dumps(lambda_assume_role_policy), ) logger.info("Created role %s.", role.name) role.attach_policy(PolicyArn=policy_arn) logger.info("Attached basic execution policy to role %s.", role.name) except ClientError as error: if error.response["Error"]["Code"] == "EntityAlreadyExists": role = self.iam_resource.Role(iam_role_name) logger.warning("The role %s already exists. Using it.", iam_role_name) else: logger.exception( "Couldn't create role %s or attach policy %s.", iam_role_name, policy_arn, ) raise return role, True def get_function(self, function_name): """ Gets data about a Lambda function. :param function_name: The name of the function. :return: The function data. """ response = None try: response = self.lambda_client.get_function(FunctionName=function_name) except ClientError as err: if err.response["Error"]["Code"] == "ResourceNotFoundException": logger.info("Function %s does not exist.", function_name) else: logger.error( "Couldn't get function %s. Here's why: %s: %s", function_name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise return response def create_function( self, function_name, handler_name, iam_role, deployment_package ): """ Deploys a Lambda function. :param function_name: The name of the Lambda function. :param handler_name: The fully qualified name of the handler function. This must include the file name and the function name. :param iam_role: The IAM role to use for the function. :param deployment_package: The deployment package that contains the function code in .zip format. :return: The Amazon Resource Name (ARN) of the newly created function. """ try: response = self.lambda_client.create_function( FunctionName=function_name, Description="AWS Lambda doc example", Runtime="python3.8", Role=iam_role.arn, Handler=handler_name, Code={"ZipFile": deployment_package}, Publish=True, ) function_arn = response["FunctionArn"] waiter = self.lambda_client.get_waiter("function_active_v2") waiter.wait(FunctionName=function_name) logger.info( "Created function '%s' with ARN: '%s'.", function_name, response["FunctionArn"], ) except ClientError: logger.error("Couldn't create function %s.", function_name) raise else: return function_arn def delete_function(self, function_name): """ Deletes a Lambda function. :param function_name: The name of the function to delete. """ try: self.lambda_client.delete_function(FunctionName=function_name) except ClientError: logger.exception("Couldn't delete function %s.", function_name) raise def invoke_function(self, function_name, function_params, get_log=False): """ Invokes a Lambda function. :param function_name: The name of the function to invoke. :param function_params: The parameters of the function as a dict. This dict is serialized to JSON before it is sent to Lambda. :param get_log: When true, the last 4 KB of the execution log are included in the response. :return: The response from the function invocation. """ try: response = self.lambda_client.invoke( FunctionName=function_name, Payload=json.dumps(function_params), LogType="Tail" if get_log else "None", ) logger.info("Invoked function %s.", function_name) except ClientError: logger.exception("Couldn't invoke function %s.", function_name) raise return response def update_function_code(self, function_name, deployment_package): """ Updates the code for a Lambda function by submitting a .zip archive that contains the code for the function. :param function_name: The name of the function to update. :param deployment_package: The function code to update, packaged as bytes in .zip format. :return: Data about the update, including the status. """ try: response = self.lambda_client.update_function_code( FunctionName=function_name, ZipFile=deployment_package ) except ClientError as err: logger.error( "Couldn't update function %s. Here's why: %s: %s", function_name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return response def update_function_configuration(self, function_name, env_vars): """ Updates the environment variables for a Lambda function. :param function_name: The name of the function to update. :param env_vars: A dict of environment variables to update. :return: Data about the update, including the status. """ try: response = self.lambda_client.update_function_configuration( FunctionName=function_name, Environment={"Variables": env_vars} ) except ClientError as err: logger.error( "Couldn't update function configuration %s. Here's why: %s: %s", function_name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return response def list_functions(self): """ Lists the Lambda functions for the current account. """ try: func_paginator = self.lambda_client.get_paginator("list_functions") for func_page in func_paginator.paginate(): for func in func_page["Functions"]: print(func["FunctionName"]) desc = func.get("Description") if desc: print(f"\t{desc}") print(f"\t{func['Runtime']}: {func['Handler']}") except ClientError as err: logger.error( "Couldn't list functions. Here's why: %s: %s", err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise

Cree una función que ejecute el escenario.

class UpdateFunctionWaiter(CustomWaiter): """A custom waiter that waits until a function is successfully updated.""" def __init__(self, client): super().__init__( "UpdateSuccess", "GetFunction", "Configuration.LastUpdateStatus", {"Successful": WaitState.SUCCESS, "Failed": WaitState.FAILURE}, client, ) def wait(self, function_name): self._wait(FunctionName=function_name) def run_scenario(lambda_client, iam_resource, basic_file, calculator_file, lambda_name): """ Runs the scenario. :param lambda_client: A Boto3 Lambda client. :param iam_resource: A Boto3 IAM resource. :param basic_file: The name of the file that contains the basic Lambda handler. :param calculator_file: The name of the file that contains the calculator Lambda handler. :param lambda_name: The name to give resources created for the scenario, such as the IAM role and the Lambda function. """ logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s") print("-" * 88) print("Welcome to the AWS Lambda getting started with functions demo.") print("-" * 88) wrapper = LambdaWrapper(lambda_client, iam_resource) print("Checking for IAM role for Lambda...") iam_role, should_wait = wrapper.create_iam_role_for_lambda(lambda_name) if should_wait: logger.info("Giving AWS time to create resources...") wait(10) print(f"Looking for function {lambda_name}...") function = wrapper.get_function(lambda_name) if function is None: print("Zipping the Python script into a deployment package...") deployment_package = wrapper.create_deployment_package( basic_file, f"{lambda_name}.py" ) print(f"...and creating the {lambda_name} Lambda function.") wrapper.create_function( lambda_name, f"{lambda_name}.lambda_handler", iam_role, deployment_package ) else: print(f"Function {lambda_name} already exists.") print("-" * 88) print(f"Let's invoke {lambda_name}. This function increments a number.") action_params = { "action": "increment", "number": q.ask("Give me a number to increment: ", q.is_int), } print(f"Invoking {lambda_name}...") response = wrapper.invoke_function(lambda_name, action_params) print( f"Incrementing {action_params['number']} resulted in " f"{json.load(response['Payload'])}" ) print("-" * 88) print(f"Let's update the function to an arithmetic calculator.") q.ask("Press Enter when you're ready.") print("Creating a new deployment package...") deployment_package = wrapper.create_deployment_package( calculator_file, f"{lambda_name}.py" ) print(f"...and updating the {lambda_name} Lambda function.") update_waiter = UpdateFunctionWaiter(lambda_client) wrapper.update_function_code(lambda_name, deployment_package) update_waiter.wait(lambda_name) print(f"This function uses an environment variable to control logging level.") print(f"Let's set it to DEBUG to get the most logging.") wrapper.update_function_configuration( lambda_name, {"LOG_LEVEL": logging.getLevelName(logging.DEBUG)} ) actions = ["plus", "minus", "times", "divided-by"] want_invoke = True while want_invoke: print(f"Let's invoke {lambda_name}. You can invoke these actions:") for index, action in enumerate(actions): print(f"{index + 1}: {action}") action_params = {} action_index = q.ask( "Enter the number of the action you want to take: ", q.is_int, q.in_range(1, len(actions)), ) action_params["action"] = actions[action_index - 1] print(f"You've chosen to invoke 'x {action_params['action']} y'.") action_params["x"] = q.ask("Enter a value for x: ", q.is_int) action_params["y"] = q.ask("Enter a value for y: ", q.is_int) print(f"Invoking {lambda_name}...") response = wrapper.invoke_function(lambda_name, action_params, True) print( f"Calculating {action_params['x']} {action_params['action']} {action_params['y']} " f"resulted in {json.load(response['Payload'])}" ) q.ask("Press Enter to see the logs from the call.") print(base64.b64decode(response["LogResult"]).decode()) want_invoke = q.ask("That was fun. Shall we do it again? (y/n) ", q.is_yesno) print("-" * 88) if q.ask( "Do you want to list all of the functions in your account? (y/n) ", q.is_yesno ): wrapper.list_functions() print("-" * 88) if q.ask("Ready to delete the function and role? (y/n) ", q.is_yesno): for policy in iam_role.attached_policies.all(): policy.detach_role(RoleName=iam_role.name) iam_role.delete() print(f"Deleted role {lambda_name}.") wrapper.delete_function(lambda_name) print(f"Deleted function {lambda_name}.") print("\nThanks for watching!") print("-" * 88) if __name__ == "__main__": try: run_scenario( boto3.client("lambda"), boto3.resource("iam"), "lambda_handler_basic.py", "lambda_handler_calculator.py", "doc_example_lambda_calculator", ) except Exception: logging.exception("Something went wrong with the demo!")

Ejemplos sin servidor

El siguiente ejemplo de código muestra cómo implementar una función Lambda que se conecta a una RDS base de datos. La función realiza una solicitud sencilla a la base de datos y devuelve el resultado.

SDKpara Python (Boto3)
nota

Hay más información. GitHub Busque el ejemplo completo y aprenda a configurar y ejecutar en el repositorio de ejemplos sin servidor.

Conexión a una RDS base de datos de Amazon en una función Lambda mediante Python.

import json import os import boto3 import pymysql # RDS settings proxy_host_name = os.environ['PROXY_HOST_NAME'] port = int(os.environ['PORT']) db_name = os.environ['DB_NAME'] db_user_name = os.environ['DB_USER_NAME'] aws_region = os.environ['AWS_REGION'] # Fetch RDS Auth Token def get_auth_token(): client = boto3.client('rds') token = client.generate_db_auth_token( DBHostname=proxy_host_name, Port=port DBUsername=db_user_name Region=aws_region ) return token def lambda_handler(event, context): token = get_auth_token() try: connection = pymysql.connect( host=proxy_host_name, user=db_user_name, password=token, db=db_name, port=port, ssl={'ca': 'Amazon RDS'} # Ensure you have the CA bundle for SSL connection ) with connection.cursor() as cursor: cursor.execute('SELECT %s + %s AS sum', (3, 2)) result = cursor.fetchone() return result except Exception as e: return (f"Error: {str(e)}") # Return an error message if an exception occurs

En el siguiente ejemplo de código se muestra cómo implementar una función de Lambda que recibe un evento activado al recibir registros de un flujo de Kinesis. La función recupera la carga útil de Kinesis, la decodifica desde Base64 y registra el contenido del registro.

SDKpara Python (Boto3)
nota

Hay más información. GitHub Busque el ejemplo completo y aprenda a configurar y ejecutar en el repositorio de ejemplos sin servidor.

Uso de un evento de Kinesis con Lambda mediante Python.

# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. # SPDX-License-Identifier: Apache-2.0 import base64 def lambda_handler(event, context): for record in event['Records']: try: print(f"Processed Kinesis Event - EventID: {record['eventID']}") record_data = base64.b64decode(record['kinesis']['data']).decode('utf-8') print(f"Record Data: {record_data}") # TODO: Do interesting work based on the new data except Exception as e: print(f"An error occurred {e}") raise e print(f"Successfully processed {len(event['Records'])} records.")

El siguiente ejemplo de código muestra cómo implementar una función de Lambda que recibe un evento desencadenado al recibir registros de una transmisión de DynamoDB. La función recupera la carga útil de DynamoDB y registra el contenido del registro.

SDKpara Python (Boto3)
nota

Hay más información. GitHub Busque el ejemplo completo y aprenda a configurar y ejecutar en el repositorio de ejemplos sin servidor.

Consumo de un evento de DynamoDB con Lambda mediante Python.

# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. # SPDX-License-Identifier: Apache-2.0 import json def lambda_handler(event, context): print(json.dumps(event, indent=2)) for record in event['Records']: log_dynamodb_record(record) def log_dynamodb_record(record): print(record['eventID']) print(record['eventName']) print(f"DynamoDB Record: {json.dumps(record['dynamodb'])}")

El siguiente ejemplo de código muestra cómo implementar una función de Lambda que recibe un evento que se desencadena al recibir registros de un flujo de cambios de DocumentDB. La función recupera la carga útil de DocumentDB y registra el contenido del registro.

SDKpara Python (Boto3)
nota

Hay más información. GitHub Busque el ejemplo completo y aprenda a configurar y ejecutar en el repositorio de ejemplos sin servidor.

Consumo de un evento de Amazon DocumentDB con Lambda mediante Python.

import json def lambda_handler(event, context): for record in event.get('events', []): log_document_db_event(record) return 'OK' def log_document_db_event(record): event_data = record.get('event', {}) operation_type = event_data.get('operationType', 'Unknown') db = event_data.get('ns', {}).get('db', 'Unknown') collection = event_data.get('ns', {}).get('coll', 'Unknown') full_document = event_data.get('fullDocument', {}) print(f"Operation type: {operation_type}") print(f"db: {db}") print(f"collection: {collection}") print("Full document:", json.dumps(full_document, indent=2))

En el siguiente ejemplo de código se muestra cómo implementar una función de Lambda que recibe un evento activado al cargar un objeto en un bucket de S3. La función recupera el nombre del bucket de S3 y la clave del objeto del parámetro de evento y llama a Amazon S3 API para recuperar y registrar el tipo de contenido del objeto.

SDKpara Python (Boto3)
nota

Hay más información. GitHub Busque el ejemplo completo y aprenda a configurar y ejecutar en el repositorio de ejemplos sin servidor.

Uso de un evento de S3 con Lambda mediante Python.

# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. # SPDX-License-Identifier: Apache-2.0 import json import urllib.parse import boto3 print('Loading function') s3 = boto3.client('s3') def lambda_handler(event, context): #print("Received event: " + json.dumps(event, indent=2)) # Get the object from the event and show its content type bucket = event['Records'][0]['s3']['bucket']['name'] key = urllib.parse.unquote_plus(event['Records'][0]['s3']['object']['key'], encoding='utf-8') try: response = s3.get_object(Bucket=bucket, Key=key) print("CONTENT TYPE: " + response['ContentType']) return response['ContentType'] except Exception as e: print(e) print('Error getting object {} from bucket {}. Make sure they exist and your bucket is in the same region as this function.'.format(key, bucket)) raise e

El siguiente ejemplo de código muestra cómo implementar una función Lambda que recibe un evento desencadenado por la recepción de mensajes de un SNS tema. La función recupera los mensajes del parámetro de eventos y registra el contenido de cada mensaje.

SDKpara Python (Boto3)
nota

Hay más información. GitHub Busque el ejemplo completo y aprenda a configurar y ejecutar en el repositorio de ejemplos sin servidor.

Consumir un SNS evento con Lambda mediante Python.

# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. # SPDX-License-Identifier: Apache-2.0 def lambda_handler(event, context): for record in event['Records']: process_message(record) print("done") def process_message(record): try: message = record['Sns']['Message'] print(f"Processed message {message}") # TODO; Process your record here except Exception as e: print("An error occurred") raise e

El siguiente ejemplo de código muestra cómo implementar una función Lambda que recibe un evento desencadenado por la recepción de mensajes de una SQS cola. La función recupera los mensajes del parámetro de eventos y registra el contenido de cada mensaje.

SDKpara Python (Boto3)
nota

Hay más información. GitHub Busque el ejemplo completo y aprenda a configurar y ejecutar en el repositorio de ejemplos sin servidor.

Consumir un SQS evento con Lambda mediante Python.

# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. # SPDX-License-Identifier: Apache-2.0 def lambda_handler(event, context): for message in event['Records']: process_message(message) print("done") def process_message(message): try: print(f"Processed message {message['body']}") # TODO: Do interesting work based on the new message except Exception as err: print("An error occurred") raise err

En el siguiente ejemplo de código se muestra cómo implementar una respuesta por lotes parcial para funciones de Lambda que reciben eventos de un flujo de Kinesis. La función informa los errores de los elementos del lote en la respuesta y le indica a Lambda que vuelva a intentar esos mensajes más adelante.

SDKpara Python (Boto3)
nota

Hay más información. GitHub Busque el ejemplo completo y aprenda a configurar y ejecutar en el repositorio de ejemplos sin servidor.

Notificación de los errores de los elementos del lote de Kinesis con Lambda mediante Python.

# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. # SPDX-License-Identifier: Apache-2.0 def handler(event, context): records = event.get("Records") curRecordSequenceNumber = "" for record in records: try: # Process your record curRecordSequenceNumber = record["kinesis"]["sequenceNumber"] except Exception as e: # Return failed record's sequence number return {"batchItemFailures":[{"itemIdentifier": curRecordSequenceNumber}]} return {"batchItemFailures":[]}

El siguiente ejemplo de código muestra cómo implementar una respuesta por lotes parcial para las funciones de Lambda que reciben eventos de una transmisión de DynamoDB. La función informa los errores de los elementos del lote en la respuesta y le indica a Lambda que vuelva a intentar esos mensajes más adelante.

SDKpara Python (Boto3)
nota

Hay más información. GitHub Busque el ejemplo completo y aprenda a configurar y ejecutar en el repositorio de ejemplos sin servidor.

Notificación de los errores de los elementos del lote de DynamoDB con Lambda mediante Python.

# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. # SPDX-License-Identifier: Apache-2.0 def handler(event, context): records = event.get("Records") curRecordSequenceNumber = "" for record in records: try: # Process your record curRecordSequenceNumber = record["dynamodb"]["SequenceNumber"] except Exception as e: # Return failed record's sequence number return {"batchItemFailures":[{"itemIdentifier": curRecordSequenceNumber}]} return {"batchItemFailures":[]}

El siguiente ejemplo de código muestra cómo implementar una respuesta por lotes parcial para las funciones de Lambda que reciben eventos de una SQS cola. La función informa los errores de los elementos del lote en la respuesta y le indica a Lambda que vuelva a intentar esos mensajes más adelante.

SDKpara Python (Boto3)
nota

Hay más información. GitHub Busque el ejemplo completo y aprenda a configurar y ejecutar en el repositorio de ejemplos sin servidor.

Notificación de errores de elementos de SQS lote con Lambda mediante Python.

# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. # SPDX-License-Identifier: Apache-2.0 def lambda_handler(event, context): if event: batch_item_failures = [] sqs_batch_response = {} for record in event["Records"]: try: # process message except Exception as e: batch_item_failures.append({"itemIdentifier": record['messageId']}) sqs_batch_response["batchItemFailures"] = batch_item_failures return sqs_batch_response