AWS Lambda Deployment and Alias Manager

AWS Lambda Deployment and Alias Manager

Problem Statement

Deploying a Lambda function is easy; safely rolling out a new version to production without breaking callers is hard. Teams typically need to upload a new code revision, publish a numbered version, and then shift a small percentage of live traffic to that version through a Lambda alias — the classic canary/blue-green pattern. Doing this manually via the AWS Console is error-prone, hard to audit, and impossible to script into a CI/CD pipeline.

This script solves that by wrapping the full deployment lifecycle into a single CLI command: it uploads a zip, updates function code, publishes a version, points an alias at it, and optionally configures weighted traffic shifting. It also supports rollback by re-pointing an alias to a previous version, which is exactly what you need at 2 a.m. when a bad deploy hits production.

How It Works

The script uses boto3 to talk to the Lambda control plane. The flow is: (1) read the zip file from disk, (2) call update_function_code to push new code to $LATEST, (3) wait for the update to finish using a waiter, (4) call publish_version to freeze an immutable snapshot, (5) call update_alias with an optional RoutingConfig for weighted traffic shifting. If --traffic-shift is omitted, the alias points 100% at the new version (instant cutover). If provided, AWS applies the weights and shifts them over time.

The design follows a command pattern: each CLI subcommand (deploy, rollback, status) maps to a function, and a shared LambdaDeployer class encapsulates the boto3 client and error handling. Configuration is resolved from CLI args first, then environment variables, then sensible defaults. All AWS errors are caught and re-raised as a custom DeploymentError with actionable messages.

Full Script

#!/usr/bin/env python3
"""
AWS Lambda Deployment and Alias Manager.

Uploads a zip to an existing Lambda function, publishes a new version,
and points a named alias at it with optional weighted traffic shifting.

Usage:
    python3 lambda_deploy.py --function my-api --zip dist.zip --alias prod --traffic-shift 10
    python3 lambda_deploy.py --function my-api --zip dist.zip --alias prod
    python3 lambda_deploy.py --function my-api --alias prod --rollback-to 12
    python3 lambda_deploy.py --function my-api --alias prod --status
"""

from __future__ import annotations

import argparse
import json
import logging
import os
import sys
import time
from pathlib import Path
from typing import Any

import boto3
from botocore.exceptions import BotoCoreError, ClientError

# ---------------------------------------------------------------------------
# Configuration (overridable via environment variables)
# ---------------------------------------------------------------------------
DEFAULT_REGION = os.environ.get("AWS_REGION", os.environ.get("AWS_DEFAULT_REGION", "us-east-1"))
DEFAULT_ALIAS = os.environ.get("LAMBDA_DEFAULT_ALIAS", "prod")
DEFAULT_TIMEOUT = int(os.environ.get("LAMBDA_UPDATE_TIMEOUT", "300"))  # seconds
LOG_LEVEL = os.environ.get("LOG_LEVEL", "INFO").upper()

logging.basicConfig(
    level=LOG_LEVEL,
    format="%(asctime)s %(levelname)-8s %(message)s",
    datefmt="%Y-%m-%dT%H:%M:%S",
)
log = logging.getLogger("lambda-deploy")


class DeploymentError(RuntimeError):
    """Raised when a deployment step fails in a way the user can act on."""


class LambdaDeployer:
    """Encapsulates Lambda code updates, version publishing, and alias management."""

    def __init__(self, function_name: str, region: str = DEFAULT_REGION) -> None:
        self.function_name = function_name
        self.region = region
        try:
            self.client = boto3.client("lambda", region_name=region)
        except BotoCoreError as exc:  # credentials/endpoint problems
            raise DeploymentError(f"Failed to create Lambda client: {exc}") from exc

    # -- helpers ----------------------------------------------------------
    def _call(self, operation: str, **kwargs: Any) -> dict[str, Any]:
        """Wrap a boto3 call with consistent error translation."""
        try:
            return getattr(self.client, operation)(**kwargs)
        except ClientError as exc:
            code = exc.response.get("Error", {}).get("Code", "Unknown")
            msg = exc.response.get("Error", {}).get("Message", str(exc))
            raise DeploymentError(f"{operation} failed [{code}]: {msg}") from exc
        except BotoCoreError as exc:
            raise DeploymentError(f"{operation} failed (transport): {exc}") from exc

    def _wait_for_update(self, timeout: int) -> None:
        """Block until the function's LastUpdateStatus is Successful."""
        deadline = time.time() + timeout
        while time.time() < deadline:
            cfg = self._call("get_function_configuration", FunctionName=self.function_name)
            status = cfg.get("LastUpdateStatus", "Successful")
            if status == "Successful":
                return
            if status == "Failed":
                reason = cfg.get("LastUpdateStatusReason", "unknown reason")
                raise DeploymentError(f"Lambda update failed: {reason}")
            log.info("Waiting for update to complete (status=%s)...", status)
            time.sleep(3)
        raise DeploymentError(f"Timed out after {timeout}s waiting for Lambda update")

    # -- core operations --------------------------------------------------
    def deploy(
        self,
        zip_path: Path,
        alias: str,
        traffic_shift: int | None = None,
        description: str | None = None,
    ) -> dict[str, Any]:
        """Upload code, publish a version, and point the alias at it."""
        if not zip_path.is_file():
            raise DeploymentError(f"Zip file not found: {zip_path}")
        if traffic_shift is not None and not (0 < traffic_shift < 100):
            raise DeploymentError("--traffic-shift must be between 1 and 99")

        payload = zip_path.read_bytes()
        log.info("Uploading %s (%d bytes) to %s", zip_path, len(payload), self.function_name)

        # 1. Push new code to $LATEST.
        self._call(
            "update_function_code",
            FunctionName=self.function_name,
            ZipFile=payload,
            Publish=False,
        )
        self._wait_for_update(DEFAULT_TIMEOUT)

        # 2. Publish an immutable version from $LATEST.
        version_resp = self._call("publish_version", FunctionName=self.function_name)
        new_version = version_resp["Version"]
        log.info("Published version %s", new_version)

        # 3. Point (or create) the alias at the new version.
        routing: dict[str, Any] | None = None
        if traffic_shift is not None:
            # Weighted alias: new version gets traffic_shift%, old gets the rest.
            try:
                current = self._call("get_alias", FunctionName=self.function_name, Name=alias)
                old_version = current["FunctionVersion"]
            except DeploymentError:
                old_version = new_version  # first deploy, no previous alias
            routing = {
                "AdditionalVersionWeights": {
                    old_version: (100 - traffic_shift) / 100.0
                }
            }
            log.info("Canary: %d%% -> v%s, %d%% -> v%s",
                     traffic_shift, new_version, 100 - traffic_shift, old_version)

        update_kwargs: dict[str, Any] = {
            "FunctionName": self.function_name,
            "Name": alias,
            "FunctionVersion": new_version,
        }
        if description:
            update_kwargs["Description"] = description
        if routing:
            update_kwargs["RoutingConfig"] = routing

        try:
            alias_resp = self._call("update_alias", **update_kwargs)
        except DeploymentError:
            # Alias doesn't exist yet -> create it.
            log.info("Alias %s not found, creating it", alias)
            alias_resp = self._call("create_alias", **update_kwargs)

        log.info("Alias %s now points at version %s", alias, alias_resp["FunctionVersion"])
        return alias_resp

    def rollback(self, alias: str, target_version: str) -> dict[str, Any]:
        """Re-point an alias to a specific previously published version."""
        log.warning("Rolling back alias %s to version %s", alias, target_version)
        resp = self._call(
            "update_alias",
            FunctionName=self.function_name,
            Name=alias,
            FunctionVersion=target_version,
        )
        log.info("Alias %s rolled back to version %s", alias, target_version)
        return resp

    def status(self, alias: str) -> dict[str, Any]:
        """Return the alias configuration and any weighted routing."""
        alias_info = self._call("get_alias", FunctionName=self.function_name, Name=alias)
        return {
            "function": self.function_name,
            "alias": alias,
            "version": alias_info["FunctionVersion"],
            "routing": alias_info.get("RoutingConfig", {}).get("AdditionalVersionWeights", {}),
        }


def build_parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(
        description="Deploy an AWS Lambda function and manage its alias.",
        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
    )
    p.add_argument("--function", required=True, help="Lambda function name or ARN")
    p.add_argument("--region", default=DEFAULT_REGION, help="AWS region")
    p.add_argument("--alias", default=DEFAULT_ALIAS, help="Alias name to update")

    # Mutually exclusive deployment actions.
    action = p.add_mutually_exclusive_group()
    action.add_argument("--zip", type=Path, help="Path to deployment zip")
    action.add_argument("--rollback-to", metavar="VERSION",
                        help="Roll the alias back to this published version")
    action.add_argument("--status", action="store_true", help="Print alias status and exit")

    p.add_argument("--traffic-shift", type=int, default=None,
                   help="Percent of traffic to send to the new version (canary)")
    p.add_argument("--description", default=None, help="Alias description")
    return p


def main(argv: list[str] | None = None) -> int:
    args = build_parser().parse_args(argv)

    if not args.zip and not args.rollback_to and not args.status:
        log.error("One of --zip, --rollback-to, or --status is required")
        return 2

    try:
        deployer = LambdaDeployer(args.function, args.region)

        if args.status:
            print(json.dumps(deployer.status(args.alias), indent=2))
            return 0

        if args.rollback_to:
            deployer.rollback(args.alias, args.rollback_to)
            return 0

        deployer.deploy(
            zip_path=args.zip,
            alias=args.alias,
            traffic_shift=args.traffic_shift,
            description=args.description,
        )
        return 0

    except DeploymentError as exc:
        log.error("Deployment aborted: %s", exc)
        return 1
    except KeyboardInterrupt:
        log.warning("Interrupted by user")
        return 130


if __name__ == "__main__":
    sys.exit(main())

Configuration

AWS credentials are picked up from the standard boto3 chain (~/.aws/credentials, environment variables, IAM roles, SSO). The IAM principal needs lambda:UpdateFunctionCode, lambda:PublishVersion, lambda:GetFunctionConfiguration, lambda:GetAlias, lambda:UpdateAlias, and lambda:CreateAlias.

Usage Examples

1. Full canary deploy — 10% to the new version:

$ python3 lambda_deploy.py --function my-api --zip dist.zip --alias prod --traffic-shift 10
2025-01-15T10:00:01 INFO     Uploading dist.zip (48213 bytes) to my-api
2025-01-15T10:00:04 INFO     Published version 13
2025-01-15T10:00:04 INFO     Canary: 10% -> v13, 90% -> v12
2025-01-15T10:00:05 INFO     Alias prod now points at version 13

2. Instant cutover (no canary):

$ python3 lambda_deploy.py --function my-api --zip dist.zip --alias prod
2025-01-15T10:05:01 INFO     Uploading dist.zip (48213 bytes) to my-api
2025-01-15T10:05:03 INFO     Published version 14
2025-01-15T10:05:03 INFO     Alias prod now points at version 14

3. Roll back to a known-good version:

$ python3 lambda_deploy.py --function my-api --alias prod --rollback-to 12
2025-01-15T10:07:00 WARNING  Rolling back alias prod to version 12
2025-01-15T10:07:01 INFO     Alias prod rolled back to version 12

4. Inspect current routing:

$ python3 lambda_deploy.py --function my-api --alias prod --status
{
  "function": "my-api",
  "alias": "prod",
  "version": "13",
  "routing": {
    "12": 0.9
  }
}

Error Handling

• Missing zip file → DeploymentError before any AWS call is made, so you never upload garbage.
• Invalid --traffic-shift (0, 100, negative) → rejected up front; AWS only accepts weights strictly between 0 and 1.
• AWS ClientError (throttling, ResourceNotFoundException, AccessDeniedException) → caught in _call, re-raised with the error code and message so the log line is actionable.
• Update failure → _wait_for_update polls LastUpdateStatus; a Failed status aborts with the AWS-provided reason instead of hanging.
• Update timeout → configurable via LAMBDA_UPDATE_TIMEOUT, defaults to 300s.
• Alias missing on first deploy → update_alias raises ResourceNotFoundException, which the deploy path catches and falls back to create_alias.
• Transport errors (BotoCoreError: DNS, TLS, credentials) → wrapped in DeploymentError.
• Ctrl+C → exits with code 130 for shell-friendly scripting.

The script always exits 0 on success and non-zero on failure, so it composes cleanly with && chains and CI pipelines.

Extending Ideas

• Auto-rollback on CloudWatch alarms: After a canary deploy, poll cloudwatch.get_metric_data for Errors on the new version and automatically call rollback() if the error rate exceeds a threshold.
• Step Functions orchestration: Replace the synchronous _wait_for_update loop with a Step Functions state machine so long deployments survive a CI runner restart.
• Multi-function deploys: Accept a YAML manifest listing several functions and their zips, then deploy them in parallel with concurrent.futures.ThreadPoolExecutor.
• Slack/Teams notifications: Emit a webhook on deploy start, canary promotion, and rollback so the team has an audit trail outside CloudTrail.
• S3-based uploads: For zips over the 50 MB direct-upload limit, upload to S3 first and use S3Bucket/S3Key in update_function_code.