Aprendiendo AWS: Lambda usando VSCode y Python

Continuando con la serie sobre AWS, en este caso he decidido jugar un poco con una de las plataformas del momento, Lambda. Para ello he construido una API con Python para leer y escribir datos de una base de datos, en local y con diferentes capas de test.

¿Qué es Lambda?

Lambda es un producto de AWS que permite ejecutar trozos de código (funciones) de manera aislada de otras aplicaciones. Esto aporta una capa extra de abstracción con todo lo bueno y malo que ello implica, y además únicamente se factura por el tiempo en el que la función se está ejecutando.

Se puede interactuar con Lambda utilizando el editor en la nube que ofrecen, subir el código a S3 usando un formulario, o utilizar sistemas como SAM local para automatizar todo el proceso, y este último es el que se ha usado en este artículo.

El objetivo

Para este pequeño proyecto, el plan consistía en crear al menos un par de funciones que expusieran una API REST, escribir en una base de datos, definir todo lo posible mediante código y configuración y que el despliegue se realizara utilizando la consola.

El ejemplo es un banco con cinco funciones, crear, depositar, retirar, ver saldo, y hacer una transferencia.

Escribiendo la primera función

El lenguaje utilizado en este caso ha sido Python, así que parte del reto ha sido aprender a escribir utilizando la sintaxis del lenguaje de la serpiente:

def setup(event, context):
    payload = json.loads(event['body’])
    account = payload['id']
    bank = Bank()
    bank.save_account({"id": account, "balance": 0})
    return {
        "statusCode": 201
    }

En Lambda, y para el caso particular de Python, se define una función como un método que acepta dos parámetros, event que contiene los datos que se le pasan a la función, y context que contiene información relativa a la ejecución de la misma.

En este ejemplo se puede ver los valores que se pasan a esta función dentro de event

{'httpMethod': 'PUT', 'body': '{"id": "test_alice"}', 'resource': '/account', 'requestContext': {'resourcePath': '/account', 'httpMethod': 'PUT', 'stage': 'prod', 'identity': {'sourceIp': '127.0.0.1:61479'}}, 'queryStringParameters': {}, 'headers': {'Accept': '*/*', 'Accept-Encoding': 'gzip, deflate', 'Connection': 'keep-alive', 'Content-Length': '20', 'Host': 'localhost', 'User-Agent': 'python-requests/2.18.4', 'X-Forwarded-Port': '', 'X-Forwarded-Proto': ''}, 'pathParameters': None, 'stageVariables': None, 'path': '/account', 'isBase64Encoded': False}

Para poder continuar, es necesario interpretar los parámetros. En este caso, el valor body es un JSON con la información de la cuenta, y para ello se crea un objeto a partir de la cadena de texto utilizando json.loads(event['body’])

Esto permite interactuar con los datos que se pasan a la función, que en este caso lo que hace es guardar la nueva cuenta en la clase Bank creada para abstraer toda la información relativa a la base de datos.

Finalmente, la función ha de devolver un objeto que contenga, al menos un campo con el estado HTTP en statusCode, que en este caso es 201 (Created).

Utilizando la misma lógica se puede hacer las operaciones de depósito, retirada, saldo y transferencia, aunque para ello puede resultar interesante abstraer la función que recibe el evento de la lógica de la función, usando una función que se llamará dispatch:

def dispatch(event, context):
    requestContext = event["requestContext"]
    resourcePath = requestContext["resourcePath"]
    method = requestContext["httpMethod"]

    if(method == "GET"):
        payload = event['queryStringParameters']
    else:
        payload = json.loads(event['body'])
    try:

        if(resourcePath == "/account"):
            if(method == "PUT"):
                setup(payload)
                return {
                    "statusCode": 201
                }

            elif(method == "GET"):
                return {
                    "statusCode": 200,
                    "body": get_balance(payload)
                }
        ...

    except Exception:
        return {
            "statusCode": 500
        }

Esta función permite leer los parámetros, interpretar la ruta y llamar a la función específica de la misma, así como hacer un pequeño tratamiento de errores.

Integrando con DynamoDB

Una vez que están todas las funciones definidas, el siguiente paso es escribir la clase Bank que será la que escriba información en la base de datos.
Mantener estas responsabilidades separadas permite, en primer lugar, una mejor capacidad de pruebas, y en segundo lugar, hace al código de la función independiente del proveedor de bases de datos.

import boto3
import os
import uuid

class Bank:

    def __init__(self):
        endpoint = os.environ.get("ENDPOINT")
        if(endpoint is None):
            self.dynamodb = boto3.resource('dynamodb')
        else:
            self.dynamodb = boto3.resource('dynamodb', endpoint_url=endpoint)

        self.accounts = self.dynamodb.Table("Accounts")
        self.transactions = self.dynamodb.Table("Transactions")

    def get_account(self, account_name):
        try:
            result = self.accounts.get_item(Key={'id': account_name})
            return result['Item']
        except:
            return None

    def log_transaction(self, transaction):
        transaction["id"] = str(uuid.uuid1())
        self.transactions.put_item(Item=transaction)

    def save_account(self, account):
        self.accounts.put_item(Item=account)

El único detalle destacable de esta clase, es que comprueba si existe un endpoint para DynamoDB definido, esto permitirá utilizar la misma clase para los tests.

Probando la función

Una vez escrito el código de las funciones y la conexión con la base de datos, el siguiente paso es probar que todo funciona como se espera.

Configurando la base de datos

Para probar la interacción con la base de datos se puede, o bien configurarla utilizando el sdk de aws y el comando aws configure o bien utilizando una base de datos local, que podemos arrancar utilizando el siguiente comando:

java -Djava.library.path=./DynamoDBLocal_lib -jar DynamoDBLocal.jar --inMemory

Esto proporciona una base de datos de DynamoDB en local lista para utilizar. La creación de tablas, en este ejemplo, se realiza como parte de los test unitarios:

table = dynamodb.create_table(
    TableName="Accounts",
    KeySchema=[
        {
            'AttributeName': 'id',
            'KeyType': 'HASH'
        }],
    AttributeDefinitions=[
        {
                'AttributeName': 'id',
                'AttributeType': 'S'
        }],
    ProvisionedThroughput={
        'ReadCapacityUnits': 5,
        'WriteCapacityUnits': 5
})
            table.meta.client.get_waiter('table_exists').wait(TableName='Accounts')

Probando las funciones

Con la base de datos funcionando y las tablas creadas, llega el momento de escribir las diferentes pruebas, como esta, que comprueba que un ingreso ha sido efectivo:

...

def test_deposit(self):
    function.setup({'id': "Bob"})
    function.deposit({'id': "Bob", "amount": 10.0})
    self.assertEqual(function.get_balance({'id': "Bob"}), 10)

Para ejecutar los tests es tan sencillo como escribir python function-tests.pydesde la consola de comandos.

Configurando Lambda

Para convertir el código de las funciones en una API que se pueda consultar vía HTTP se ha recurrido a SAM (Serverless Application Model) local, un conjunto de herramientas para definir, probar y desplegar funciones Lambda.

La plantilla

Una plantilla de SAM no es más que un subconjunto de una plantilla de CloudFormation, que en un artículo anterior comentaba que es una manera de automatizar la creación y actualización de recursos dentro de AWS.

Para este ejemplo la plantilla tiene el siguiente aspecto:

AWSTemplateFormatVersion: '2010-09-09'
Transform: 'AWS::Serverless-2016-10-31'
Description: 'A simple banking app built on lambdas'

Resources:
  Bank:
    Type: AWS::Serverless::Function
    Properties:
      Handler: function.dispatch
      CodeUri: .
      Runtime: python3.6
      Events:
        CreateAccount:
          Type: Api
          Properties:
            Path: '/account/new'
            Method: put
        GetBalance:
          Type: Api
          Properties:
            Path: '/account'
            Method: get
        Deposit:
          Type: Api
          Properties:
            Path: '/account/deposit'
            Method: post
        Transfer:
          Type: Api
          Properties:
            Path: '/account/transfer'
            Method: post
        Withdraw:
          Type: Api
          Properties:
            Path: '/account/withdraw'
            Method: post
      Policies:
        - Version: '2012-10-17'
          Statement:
            - Effect: Allow
              Action:
                - 'dynamodb:PutItem'
                - 'dynamodb:GetItem'
              Resource: 'arn:aws:dynamodb:*:*:*'

En esta plantilla se define:
– El handler, o función que responderá a la llamada
– Los eventos que dispararán la llamada (varios eventos pueden disparar el mismo handler)
– Las políticas que se aplican (en esta caso necesitamos permisos de lectura y escritura para la base de datos).

Validando y probando la plantilla con SAM

Una vez definidos todos los valores para la plantilla, se puede ejecutar SAM para comprobar que la plantilla es válida:

sam validate

El cliente SAM permite, además, probar la definición de la plantilla de manera local:
sam local start-api

Este comando inicializa un servidor en el puerto 3000 con el que se pueden probar las funciones desde un cliente HTTP (en este ejemplo he utilizado Postman) y ver como la API responde a las diferentes llamadas.

Pruebas de integración

Para comprobar que todo funciona bien antes de subir la plantilla a AWS, he creado un conjunto de tests un tanto diferentes, esta vez para comprobar que los datos se transforman correctamente, llamando al cliente de localhost.

def test_deposit(self):
    r = requests.get('http://localhost:3000/account?id=test_bob')
    initial_bob_balance = float(r.text)

    r = requests.post('http://localhost:3000/account/deposit', data =  json.dumps({'id':'test_bob', 'amount': 100}))
    self.assertEqual(200, r.status_code)

    r = requests.get('http://localhost:3000/account?id=test_bob')
    new_balance = float(r.text)

    self.assertEqual(new_balance, initial_bob_balance + 100)

Para ejecutar los tests es tan sencillo como escribir python integration-tests.pydesde la consola de comandos.

Desplegando la plantilla

Finalmente queda empaquetar la función y desplegarla en la consola de AWS. En primer lugar se empaqueta la función en un fichero zip, se guarda en una carpeta (bucket) de S3 y se actualizan las referencias en el fichero que se ha denominado package.yaml:

sam package --template-file template.yaml --s3-bucket my-bucket --output-template-file packaged.yaml

En segundo lugar se invoca a CloudFormation para crear todos los recursos definidos en el paquete, incluyendo el código fuente y el conjunto de permisos:

sam deploy --template-file packaged.yaml --stack-name my-stack-name --capabilities CAPABILITY_IAM

Una vez se haya desplegado el paquete, se puede ir a la consola de AWS y ver cómo se ha desplegado la función:

Al hacer click, se puede ver además los diferentes endpoints que API Gateway ha creado. así como las diferentes opciones:

Con esto la función creada está lista para ser ejecutada.

En resumen

Eh este artículo se ha definido una función Lambda de manera local, utilizando tantas capas de abstracción como hayan sido necesarias. Se han creado diferentes tipos de test para probar los diferentes puntos de integración, y finalmente se ha empaquetado y desplegado la función desde la línea de comandos.

El código está disponible en GitHub.

Sin embargo, se han quedado algunas cosas fuera del alcance del artículo, y quedan como sugerencia al lector para continuar experimentando:

• Inyección de dependencias y Mocking para hacer tests unitarios
• Integración con un sistema como Jenkins para ejecutar tests
• Crear la base de datos utilizando la plantilla de SAM
• Uso de frameworks como Serverless que agregan una capa de abstracción incluso mayor, aislando de las implementaciones de AWS, Azure, Google Cloud o IBM.

¿Y tú, has hecho alguna prueba con Lambdas u otra plataforma Serverless?

Herramientas utilizadas

• SAM local para probar la función en local, así como para desplegarla en AWS.
• Visual Studio Code para editar el código y para interactuar con la API de SAM local usando su consola integrada.
• DynamoDB Local para emular una base de datos de DynamoDB.
• Boto 3 el SDK oficial de AWS para Python.
• Postman para probar las diferentes APIs tanto locales como remotas.

Más documentación y ejemplos

• GitHub – rlbisbe/serverless-bank
• Programming Model for Authoring Lambda Functions in Java – AWS Lambda
• aws-sam-local/samples at develop · awslabs/aws-sam-local · GitHub