AWS Lambda Layers Best Practices

# AWS Lambda Layers Best Practices

Master Lambda layers for code reuse with Google Antigravity IDE. This comprehensive guide covers layer creation, versioning, and dependency management for serverless functions.

## Why Lambda Layers?

Lambda layers enable code sharing across functions and reduce deployment sizes. Google Antigravity IDE's Gemini 3 engine suggests optimal layer organization.

## Layer Structure

```
layers/
├── common/
│   ├── python/
│   │   └── common/
│   │       ├── __init__.py
│   │       ├── utils.py
│   │       ├── logging.py
│   │       └── exceptions.py
│   └── requirements.txt
├── database/
│   ├── python/
│   │   └── database/
│   │       ├── __init__.py
│   │       ├── connection.py
│   │       └── models.py
│   └── requirements.txt
└── build.sh
```

## Common Utilities Layer

```python
# layers/common/python/common/utils.py
import json
import os
from typing import Any, TypeVar, Callable
from functools import wraps
from datetime import datetime, timezone

T = TypeVar("T")

def get_env(key: str, default: str | None = None, required: bool = False) -> str:
    """Get environment variable with validation."""
    value = os.environ.get(key, default)
    if required and value is None:
        raise ValueError(f"Required environment variable {key} not set")
    return value

def json_response(
    status_code: int,
    body: dict[str, Any],
    headers: dict[str, str] | None = None
) -> dict[str, Any]:
    """Create API Gateway compatible response."""
    default_headers = {
        "Content-Type": "application/json",
        "Access-Control-Allow-Origin": "*",
    }
    
    return {
        "statusCode": status_code,
        "headers": {**default_headers, **(headers or {})},
        "body": json.dumps(body, default=str),
    }

def parse_body(event: dict) -> dict[str, Any]:
    """Parse request body from API Gateway event."""
    body = event.get("body", "{}")
    if isinstance(body, str):
        return json.loads(body)
    return body

def now_utc() -> datetime:
    """Get current UTC timestamp."""
    return datetime.now(timezone.utc)

# layers/common/python/common/logging.py
import logging
import json
import sys
from typing import Any

class JSONFormatter(logging.Formatter):
    """JSON log formatter for CloudWatch."""
    
    def format(self, record: logging.LogRecord) -> str:
        log_data = {
            "timestamp": self.formatTime(record),
            "level": record.levelname,
            "message": record.getMessage(),
            "logger": record.name,
            "function": record.funcName,
        }
        
        if record.exc_info:
            log_data["exception"] = self.formatException(record.exc_info)
        
        if hasattr(record, "extra"):
            log_data.update(record.extra)
        
        return json.dumps(log_data)

def get_logger(name: str, level: str = "INFO") -> logging.Logger:
    """Get configured logger instance."""
    logger = logging.getLogger(name)
    logger.setLevel(getattr(logging, level))
    
    if not logger.handlers:
        handler = logging.StreamHandler(sys.stdout)
        handler.setFormatter(JSONFormatter())
        logger.addHandler(handler)
    
    return logger

# layers/common/python/common/decorators.py
from functools import wraps
from typing import Callable, TypeVar, ParamSpec
from .logging import get_logger
from .utils import json_response

P = ParamSpec("P")
R = TypeVar("R")

logger = get_logger(__name__)

def error_handler(func: Callable[P, R]) -> Callable[P, R]:
    """Decorator for Lambda error handling."""
    
    @wraps(func)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        try:
            return func(*args, **kwargs)
        except ValidationError as e:
            logger.warning(f"Validation error: {e}")
            return json_response(400, {"error": str(e)})
        except NotFoundError as e:
            logger.warning(f"Not found: {e}")
            return json_response(404, {"error": str(e)})
        except Exception as e:
            logger.exception(f"Unhandled error: {e}")
            return json_response(500, {"error": "Internal server error"})
    
    return wrapper

def log_execution(func: Callable[P, R]) -> Callable[P, R]:
    """Decorator for logging function execution."""
    
    @wraps(func)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        logger.info(f"Starting {func.__name__}")
        start = time.time()
        
        try:
            result = func(*args, **kwargs)
            elapsed = time.time() - start
            logger.info(f"Completed {func.__name__} in {elapsed:.2f}s")
            return result
        except Exception as e:
            elapsed = time.time() - start
            logger.error(f"Failed {func.__name__} after {elapsed:.2f}s: {e}")
            raise
    
    return wrapper
```

## Database Layer

```python
# layers/database/python/database/connection.py
import os
from contextlib import asynccontextmanager
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

DATABASE_URL = os.environ.get("DATABASE_URL")

engine = create_async_engine(
    DATABASE_URL,
    pool_size=5,
    max_overflow=10,
    pool_pre_ping=True,
)

AsyncSessionLocal = sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

@asynccontextmanager
async def get_session():
    """Get database session context manager."""
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
```

## Build Script

```bash
#!/bin/bash
# layers/build.sh

set -e

LAYERS=("common" "database")

for layer in "${LAYERS[@]}"; do
    echo "Building layer: $layer"
    
    cd "layers/$layer"
    
    # Install dependencies
    pip install -r requirements.txt -t python/ --upgrade
    
    # Create zip
    zip -r "../$layer-layer.zip" python/
    
    cd ../..
    
    echo "Built: layers/$layer-layer.zip"
done
```

## CDK Infrastructure

```typescript
// infrastructure/layers-stack.ts
import * as cdk from "aws-cdk-lib";
import * as lambda from "aws-cdk-lib/aws-lambda";

export class LayersStack extends cdk.Stack {
  public readonly commonLayer: lambda.LayerVersion;
  public readonly databaseLayer: lambda.LayerVersion;
  
  constructor(scope: cdk.App, id: string) {
    super(scope, id);
    
    this.commonLayer = new lambda.LayerVersion(this, "CommonLayer", {
      code: lambda.Code.fromAsset("layers/common"),
      compatibleRuntimes: [lambda.Runtime.PYTHON_3_12],
      description: "Common utilities layer",
    });
    
    this.databaseLayer = new lambda.LayerVersion(this, "DatabaseLayer", {
      code: lambda.Code.fromAsset("layers/database"),
      compatibleRuntimes: [lambda.Runtime.PYTHON_3_12],
      description: "Database connection layer",
    });
  }
}
```

## Best Practices

- Keep layers focused and single-purpose
- Use semantic versioning for layers
- Minimize layer size for cold start
- Share common dependencies across functions
- Test layers independently
- Document layer contents and usage

Google Antigravity IDE provides Lambda layer templates and automatically suggests optimal dependency organization for your serverless functions.