DevOps

Tutorial: Deploying an AWS Lambda Function with Dependencies, Environment Variables, and Refresh Token Using the AWS Console

Posted on by Karmina Clemente
22
May

1. Create the function in AWS Lambda

Step 1: Access the Lambda console

  1. Log in to your AWS Console.
  2. In the search bar, type “Lambda” and select AWS Lambda.
  3. Click on Create function.

Step 2: Configure the function

  1. Select “Author from scratch.”
  2. Fill out the fields:
    • Function name: assign_badges_lambda
    • Runtime: Python 3.11
    • Architecture: x86_64
    • Execution role:
      • Select “Create a new role with basic Lambda permissions”.
  3. Click on Create function.

2. Create the code and its dependencies

We need a deployment package (ZIP file) with our code in lambda_function.py and libraries (e.g., requests).

Step 1: Prepare the package locally

  1. Create a folder, for example: su-lambda.
  2. Inside su-lambda, create lambda_function.py with example code that uses a refresh token if the access token is expired. Let’s assume your API has an endpoint /auth/login/refresh/ that receives {"refresh": "<REFRESH_TOKEN>"} and returns a new access token.
import requests
import os
import logging

logger = logging.getLogger()
logger.setLevel(logging.INFO)

def refresh_access_token(session, refresh_url, refresh_token):
    """
    Requests a new access token using the refresh token.
    Returns the new token or None if it fails.
    """
    logger.info("Attempting to refresh access token...")
    try:
        # Assume your refresh API is a POST to refresh_url with JSON: {"refresh": "<REFRESH_TOKEN>"}
        response = session.post(refresh_url, json={"refresh": refresh_token})
        response.raise_for_status()
        data = response.json()
        new_access = data.get("access")
        if new_access:
            logger.info("Access token refreshed successfully.")
            return new_access
        else:
            logger.error("Refresh response did not include an 'access' token.")
            return None
    except requests.exceptions.RequestException as e:
        logger.error(f"Error refreshing token: {str(e)}")
        return None

def lambda_handler(event, context):
    logger.info("Lambda function started execution.")

    # Read environment variables
    API_URL = os.environ.get("DJANGO_API_URL", "https://api.ejemplo.com/assign-badges/")
    REFRESH_URL = os.environ.get("DJANGO_REFRESH_URL", "https://api.ejemplo.com/auth/login/refresh/")
    ACCESS_TOKEN = os.environ.get("DJANGO_API_ACCESS_TOKEN")  # Initial Access Token
    REFRESH_TOKEN = os.environ.get("DJANGO_API_REFRESH_TOKEN")  # Refresh Token

    if not ACCESS_TOKEN or not REFRESH_TOKEN:
        logger.error("Access or Refresh token is missing. Check environment variables.")
        return {
            'statusCode': 500,
            'body': 'Missing tokens in environment variables'
        }

    # Use requests.Session() to handle cookies if needed.
    session = requests.Session()

    headers = {
        'Authorization': f'Bearer {ACCESS_TOKEN}',
        'Content-Type': 'application/json'
    }

    try:
        # Attempt main call
        logger.info(f"Sending request to {API_URL}")
        response = session.post(API_URL, headers=headers)
        
        # If access token is expired, API might respond 401
        if response.status_code == 401:
            logger.info("Access token expired. Attempting to refresh...")
            new_access = refresh_access_token(session, REFRESH_URL, REFRESH_TOKEN)
            if new_access:
                # Update header with new token
                headers['Authorization'] = f'Bearer {new_access}'
                # Retry the request
                logger.info("Retrying the API call with new access token...")
                response = session.post(API_URL, headers=headers)
            else:
                logger.error("Could not refresh access token. Aborting.")
                return {
                    'statusCode': 401,
                    'body': 'Unable to refresh token'
                }

        # Final response handling
        response.raise_for_status()
        logger.info(f"Response Status: {response.status_code}")
        logger.info(f"Response Content: {response.text}")

        return {
            'statusCode': 200,
            'body': 'Badges assigned successfully!'
        }
    except requests.exceptions.RequestException as e:
        logger.error(f"Error calling the API: {str(e)}")
        return {
            'statusCode': 500,
            'body': f'Error: {str(e)}'
        }
  1. Compress everything into a ZIP:bashCopyzip -r deployment_package.zip .

Notes:

  • refresh_access_token performs the flow to request a new access token.
  • It is assumed the API returns JSON with a key "access" if successful.
  • If your API handles cookies/CSRF, you should extend the code to get a csrftoken and include it in every call.

3.- From the folder su-lambda, install the dependencies (e.g., requests):

pip3 install requests -t .

4.- Compress everything into a ZIP:

zip -r deployment_package.zip .

Step 2: Upload the ZIP to AWS Lambda

  1. Return to the AWS Lambda console, select your function assign_badges_lambda.
  2. Go to the Code tab and click “Upload from” → “.zip file”.
  3. Select deployment_package.zip and confirm.
  4. Click Deploy.

3. Configure environment variables

Go to the Configuration tab → Environment variables to specify your API URL, the access token, the refresh token, etc.

For example:

  • Key: DJANGO_API_URL
    Value: https://api.ejemplo.com/assign-badges
    • Key: DJANGO_API_URL
      Value: https://api.ejemplo.com/assign-badges
Karmina Clemente