giuseppe-trisciuoglio/developer-kit

Comprehensive developer toolkit providing reusable skills for Java/Spring Boot, TypeScript/NestJS/React/Next.js, Python, PHP, AWS CloudFormation, AI/RAG, DevOps, and more.

Quality

82%

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Securityby

Risky

Do not use without reviewing

Validation failed for skills in this tile

One or more skills have errors that need to be fixed before they can move to Implementation and Discovery review.

Raw Python Lambda

Name: giuseppe-trisciuoglio/developer-kit
Rating: 82.6470588235294 (1 reviews)
Author: giuseppe-trisciuoglio

Guide for creating minimal AWS Lambda functions in Python without frameworks.

When to Use Raw Python

Simple handlers with minimal dependencies
Maximum control over the execution environment
Smallest deployment package size
Learning Lambda fundamentals
Custom runtime requirements

Basic Handler

Minimal Handler

# lambda_function.py
import json

def lambda_handler(event, context):
    """
    Main entry point for Lambda function.

    Args:
        event: Event data passed to the function
        context: Lambda runtime context

    Returns:
        dict: Response object
    """
    return {
        'statusCode': 200,
        'body': json.dumps({'message': 'Hello from Lambda!'})
    }

Handler with Different Triggers

# lambda_function.py
import json

def lambda_handler(event, context):
    """Handle different event sources."""

    # Detect event source
    if 'httpMethod' in event:
        return handle_api_gateway(event, context)
    elif 'Records' in event:
        return handle_s3_event(event, context)
    elif 'source' in event and event['source'] == 'aws.events':
        return handle_cloudwatch_event(event, context)
    else:
        return handle_direct_invoke(event, context)

def handle_api_gateway(event, context):
    """Handle API Gateway proxy integration."""
    return {
        'statusCode': 200,
        'headers': {'Content-Type': 'application/json'},
        'body': json.dumps({
            'message': 'API Gateway request',
            'path': event.get('path'),
            'method': event.get('httpMethod')
        })
    }

def handle_s3_event(event, context):
    """Handle S3 trigger events."""
    for record in event['Records']:
        bucket = record['s3']['bucket']['name']
        key = record['s3']['object']['key']
        print(f"S3 event: {bucket}/{key}")

    return {'statusCode': 200, 'processed': len(event['Records'])}

def handle_cloudwatch_event(event, context):
    """Handle CloudWatch scheduled events."""
    print(f"Scheduled event: {event}")
    return {'statusCode': 200}

def handle_direct_invoke(event, context):
    """Handle direct Lambda invocation."""
    return {
        'statusCode': 200,
        'result': event
    }

Cold Start Optimization

Module-Level Caching

# lambda_function.py
import boto3
import os

# Initialize at module level - persists across warm invocations
_dynamodb = None
_table = None
_s3 = None

def get_dynamodb_table():
    """Lazy initialization with caching."""
    global _dynamodb, _table
    if _table is None:
        _dynamodb = boto3.resource('dynamodb')
        _table = _dynamodb.Table(os.environ['TABLE_NAME'])
    return _table

def get_s3_client():
    """Lazy initialization with caching."""
    global _s3
    if _s3 is None:
        _s3 = boto3.client('s3')
    return _s3

def lambda_handler(event, context):
    # Uses cached clients
    table = get_dynamodb_table()
    s3 = get_s3_client()

    # Handler logic here
    return {'statusCode': 200}

Lazy Loading Heavy Dependencies

# lambda_function.py
_heavy_service = None

def get_heavy_service():
    """Defer loading heavy modules until needed."""
    global _heavy_service
    if _heavy_service is None:
        # Import here to avoid loading during cold start
        from heavy_library import HeavyService
        _heavy_service = HeavyService()
    return _heavy_service

def lambda_handler(event, context):
    # Only load heavy service when needed
    if event.get('needs_heavy_processing'):
        service = get_heavy_service()
        return service.process(event['data'])

    return {'statusCode': 200, 'light': True}

API Gateway Integration

REST API Handler

# lambda_function.py
import json
import re

# Route definitions
ROUTES = {
    r'^GET /users$': 'list_users',
    r'^GET /users/(?P<user_id>[^/]+)$': 'get_user',
    r'^POST /users$': 'create_user',
    r'^PUT /users/(?P<user_id>[^/]+)$': 'update_user',
    r'^DELETE /users/(?P<user_id>[^/]+)$': 'delete_user',
}

def lambda_handler(event, context):
    """Main router for API Gateway requests."""
    http_method = event.get('httpMethod', 'GET')
    path = event.get('path', '/')

    # Match route
    route_key = f"{http_method} {path}"

    for pattern, handler_name in ROUTES.items():
        match = re.match(pattern, route_key)
        if match:
            handler = globals()[handler_name]
            return handler(event, context, **match.groupdict())

    return response(404, {'error': 'Not found'})

def list_users(event, context):
    """GET /users"""
    # Implementation here
    return response(200, {'users': []})

def get_user(event, context, user_id):
    """GET /users/{user_id}"""
    return response(200, {'user_id': user_id})

def create_user(event, context):
    """POST /users"""
    body = json.loads(event.get('body', '{}'))
    return response(201, {'created': body})

def update_user(event, context, user_id):
    """PUT /users/{user_id}"""
    body = json.loads(event.get('body', '{}'))
    return response(200, {'updated': user_id, 'data': body})

def delete_user(event, context, user_id):
    """DELETE /users/{user_id}"""
    return response(200, {'deleted': user_id})

def response(status_code, body, headers=None):
    """Helper for API Gateway responses."""
    default_headers = {
        'Content-Type': 'application/json',
        'Access-Control-Allow-Origin': '*'
    }
    if headers:
        default_headers.update(headers)

    return {
        'statusCode': status_code,
        'headers': default_headers,
        'body': json.dumps(body)
    }

Request/Response Helpers

# lambda_function.py
import json
from typing import Dict, Any, Optional

class APIGatewayRequest:
    """Wrapper for API Gateway events."""

    def __init__(self, event: Dict[str, Any]):
        self.event = event
        self.method = event.get('httpMethod', 'GET')
        self.path = event.get('path', '/')
        self.query_params = event.get('queryStringParameters') or {}
        self.path_params = event.get('pathParameters') or {}
        self.headers = {k.lower(): v for k, v in (event.get('headers') or {}).items()}

        body = event.get('body')
        self.body = json.loads(body) if body and event.get('isBase64Encoded') else body

    def get_header(self, name: str, default: str = None) -> Optional[str]:
        return self.headers.get(name.lower(), default)

    def get_query(self, name: str, default: str = None) -> Optional[str]:
        return self.query_params.get(name, default)

class APIGatewayResponse:
    """Builder for API Gateway responses."""

    def __init__(self):
        self.status_code = 200
        self.headers = {'Content-Type': 'application/json'}
        self.body = {}

    def status(self, code: int) -> 'APIGatewayResponse':
        self.status_code = code
        return self

    def header(self, name: str, value: str) -> 'APIGatewayResponse':
        self.headers[name] = value
        return self

    def json(self, data: Dict) -> Dict[str, Any]:
        return {
            'statusCode': self.status_code,
            'headers': self.headers,
            'body': json.dumps(data)
        }

    def text(self, text: str) -> Dict[str, Any]:
        self.headers['Content-Type'] = 'text/plain'
        return {
            'statusCode': self.status_code,
            'headers': self.headers,
            'body': text
        }

# Usage
def lambda_handler(event, context):
    request = APIGatewayRequest(event)

    if request.method == 'GET':
        return APIGatewayResponse().json({'path': request.path})

    return APIGatewayResponse().status(405).json({'error': 'Method not allowed'})

Error Handling

Structured Error Responses

# lambda_function.py
import json
import traceback
from typing import Dict, Any

class LambdaError(Exception):
    """Custom error with HTTP status code."""

    def __init__(self, message: str, status_code: int = 500, details: Dict = None):
        self.message = message
        self.status_code = status_code
        self.details = details or {}
        super().__init__(message)

class ValidationError(LambdaError):
    def __init__(self, message: str, details: Dict = None):
        super().__init__(message, 400, details)

class NotFoundError(LambdaError):
    def __init__(self, message: str):
        super().__init__(message, 404)

def error_response(error: LambdaError, request_id: str = None) -> Dict[str, Any]:
    """Format error for API Gateway response."""
    return {
        'statusCode': error.status_code,
        'headers': {'Content-Type': 'application/json'},
        'body': json.dumps({
            'error': error.__class__.__name__,
            'message': error.message,
            'details': error.details,
            'requestId': request_id
        })
    }

def lambda_handler(event, context):
    """Handler with centralized error handling."""
    try:
        return process_request(event, context)
    except LambdaError as e:
        return error_response(e, context.aws_request_id)
    except Exception as e:
        # Log full traceback for debugging
        print(f"Unhandled error: {traceback.format_exc()}")
        return error_response(
            LambdaError("Internal server error"),
            context.aws_request_id
        )

def process_request(event, context):
    """Main request processing logic."""
    user_id = event.get('pathParameters', {}).get('user_id')

    if not user_id:
        raise ValidationError("user_id is required")

    user = find_user(user_id)
    if not user:
        raise NotFoundError(f"User {user_id} not found")

    return {
        'statusCode': 200,
        'headers': {'Content-Type': 'application/json'},
        'body': json.dumps({'user': user})
    }

def find_user(user_id: str) -> Dict:
    # Implementation
    return None

Context Usage

Lambda Context Object

# lambda_function.py

def lambda_handler(event, context):
    """Demonstrate context object usage."""
    context_info = {
        # Identity
        'function_name': context.function_name,
        'function_version': context.function_version,
        'memory_limit_mb': context.memory_limit_in_mb,
        'aws_request_id': context.aws_request_id,
        'invoked_function_arn': context.invoked_function_arn,
        'log_group_name': context.log_group_name,
        'log_stream_name': context.log_stream_name,

        # Timing
        'remaining_time_ms': context.get_remaining_time_in_millis(),
    }

    # Check if we have enough time
    if context.get_remaining_time_in_millis() < 1000:
        print("Warning: Running low on time!")

    # Identity (for Cognito authorizer)
    if hasattr(context, 'identity'):
        context_info['identity'] = {
            'cognito_identity_id': context.identity.cognito_identity_id,
            'cognito_identity_pool_id': context.identity.cognito_identity_pool_id,
        }

    # Client context (for mobile SDK)
    if hasattr(context, 'client_context'):
        context_info['client_context'] = context.client_context

    return {
        'statusCode': 200,
        'body': json.dumps(context_info)
    }

Environment Configuration

Configuration Class

# config.py
import os
from typing import List

class Config:
    """Configuration management with validation."""

    # Required environment variables
    REQUIRED = ['TABLE_NAME']

    # Optional with defaults
    DEBUG: bool = os.environ.get('DEBUG', 'false').lower() == 'true'
    REGION: str = os.environ.get('AWS_REGION', 'us-east-1')
    LOG_LEVEL: str = os.environ.get('LOG_LEVEL', 'INFO')
    TABLE_NAME: str = os.environ.get('TABLE_NAME', '')
    BUCKET_NAME: str = os.environ.get('BUCKET_NAME', '')

    # Numeric values
    TIMEOUT_SECONDS: int = int(os.environ.get('TIMEOUT_SECONDS', '30'))
    MAX_RETRIES: int = int(os.environ.get('MAX_RETRIES', '3'))

    # Lists
    ALLOWED_ORIGINS: List[str] = os.environ.get('ALLOWED_ORIGINS', '*').split(',')

    @classmethod
    def validate(cls) -> None:
        """Validate required configuration."""
        missing = []
        for var in cls.REQUIRED:
            if not getattr(cls, var):
                missing.append(var)

        if missing:
            raise ValueError(f"Missing required environment variables: {missing}")

    @classmethod
    def to_dict(cls) -> dict:
        """Export configuration as dictionary."""
        return {
            'DEBUG': cls.DEBUG,
            'REGION': cls.REGION,
            'LOG_LEVEL': cls.LOG_LEVEL,
            'TABLE_NAME': cls.TABLE_NAME,
            'BUCKET_NAME': cls.BUCKET_NAME,
            'TIMEOUT_SECONDS': cls.TIMEOUT_SECONDS,
            'MAX_RETRIES': cls.MAX_RETRIES,
        }

# lambda_function.py
from config import Config

# Validate on module load
Config.validate()

def lambda_handler(event, context):
    if Config.DEBUG:
        print(f"Config: {Config.to_dict()}")

    return {'statusCode': 200, 'config': Config.to_dict()}

Logging

Structured Logging

# lambda_function.py
import json
import logging
from datetime import datetime, timezone

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

class StructuredLog:
    """Structured logging for CloudWatch."""

    @staticmethod
    def info(message: str, **kwargs):
        log_entry = {
            'timestamp': datetime.now(timezone.utc).isoformat(),
            'level': 'INFO',
            'message': message,
            **kwargs
        }
        logger.info(json.dumps(log_entry))

    @staticmethod
    def error(message: str, error: Exception = None, **kwargs):
        log_entry = {
            'timestamp': datetime.now(timezone.utc).isoformat(),
            'level': 'ERROR',
            'message': message,
            **kwargs
        }
        if error:
            log_entry['error_type'] = error.__class__.__name__
            log_entry['error_message'] = str(error)

        logger.error(json.dumps(log_entry))

    @staticmethod
    def metric(name: str, value: float, unit: str = 'Count'):
        """Emit CloudWatch metric via log."""
        log_entry = {
            '_aws': {
                'Timestamp': int(datetime.now(timezone.utc).timestamp() * 1000),
                'CloudWatchMetrics': [{
                    'Namespace': 'Lambda/Custom',
                    'Dimensions': [['FunctionName']],
                    'Metrics': [{'Name': name, 'Unit': unit}]
                }]
            },
            'FunctionName': os.environ.get('AWS_LAMBDA_FUNCTION_NAME', 'unknown'),
            name: value
        }
        logger.info(json.dumps(log_entry))

# Usage
def lambda_handler(event, context):
    StructuredLog.info(
        'Processing request',
        request_id=context.aws_request_id,
        path=event.get('path')
    )

    try:
        result = process_event(event)
        StructuredLog.metric('SuccessCount', 1)
        return {'statusCode': 200, 'body': result}
    except Exception as e:
        StructuredLog.error('Processing failed', error=e)
        StructuredLog.metric('ErrorCount', 1)
        raise

Deployment Packaging

requirements.txt

# Minimal requirements for raw Python Lambda
boto3>=1.35.0

# Add only what you need
requests>=2.32.0  # For HTTP calls
pydantic>=2.5.0   # For data validation

Build Script

#!/bin/bash
# build.sh - Build deployment package

set -e

PACKAGE_DIR="package"
OUTPUT="deployment.zip"

# Clean previous builds
rm -rf $PACKAGE_DIR $OUTPUT

# Install dependencies
pip install -r requirements.txt -t $PACKAGE_DIR

# Copy source code
cp lambda_function.py config.py $PACKAGE_DIR/

# Create zip
cd $PACKAGE_DIR
zip -r ../$OUTPUT .
cd ..

echo "Created $OUTPUT"

Best Practices

Keep handlers small - Delegate to separate functions/classes
Initialize outside handler - Module-level for warm start reuse
Use lazy loading - Defer heavy imports until needed
Handle timeouts - Check context.get_remaining_time_in_millis()
Validate input - Always check event structure
Use structured logging - JSON for CloudWatch Insights
Environment configuration - No hardcoded values
Error handling - Graceful degradation with proper HTTP codes

plugins

developer-kit-aws

developer-kit-core

developer-kit-java

developer-kit-php

developer-kit-python

skills

aws-lambda-python-integration

references

chalice-lambda.md

raw-python-lambda.md

serverless-deployment.md

testing-lambda.md

SKILL.md

developer-kit-tools

developer-kit-typescript

README.md

tile.json

giuseppe-trisciuoglio/developer-kit