tessl/pypi-zappa
Server-less Python Web Services for AWS Lambda and API Gateway
core-deployment.mddocs/
Core Deployment Management
Primary AWS resource management functions that form the foundation of Zappa's serverless deployment capabilities. These methods handle Lambda function lifecycle, API Gateway deployment, and AWS infrastructure management.
Capabilities
Lambda Function Management
Create, update, configure, and delete AWS Lambda functions with comprehensive deployment options.
def create_lambda_function(
self,
bucket=None,
function_name=None,
handler=None,
s3_key=None,
description="Zappa Deployment",
timeout=30,
memory_size=512,
ephemeral_storage={"Size": 512},
publish=True,
layers=None,
dead_letter_config=None,
environment_variables=None,
vpc_config=None,
**kwargs
):
"""
Create a new Lambda function.
Parameters:
- bucket: str, S3 bucket containing deployment package
- function_name: str, name of the Lambda function
- handler: str, entry point for the function (module.function)
- s3_key: str, S3 key of the deployment package
- description: str, function description (default "Zappa Deployment")
- timeout: int, function timeout in seconds (1-900)
- memory_size: int, memory allocation in MB (128-10240)
- ephemeral_storage: dict, ephemeral storage configuration
- publish: bool, publish a version of the function
- layers: list, ARNs of Lambda layers
- dead_letter_config: dict, dead letter queue configuration
- environment_variables: dict, environment variables
- vpc_config: dict, VPC configuration
Returns:
dict: Lambda function configuration
"""def update_lambda_function(
self,
bucket,
function_name,
s3_key=None,
publish=True,
**kwargs
):
"""
Update Lambda function code.
Parameters:
- bucket: str, S3 bucket containing deployment package
- function_name: str, name of the function to update
- s3_key: str, S3 key of the new deployment package
- publish: bool, publish a new version
Returns:
dict: Updated function configuration
"""def update_lambda_configuration(
self,
function_name,
timeout=None,
memory_size=None,
environment_variables=None,
vpc_config=None,
layers=None,
**kwargs
):
"""
Update Lambda function configuration.
Parameters:
- function_name: str, name of the function
- timeout: int, function timeout in seconds
- memory_size: int, memory allocation in MB
- environment_variables: dict, environment variables
- vpc_config: dict, VPC configuration
- layers: list, ARNs of Lambda layers
Returns:
dict: Updated function configuration
"""def delete_lambda_function(self, function_name):
"""
Delete a Lambda function.
Parameters:
- function_name: str, name of the function to delete
Returns:
bool: True if deletion successful
"""Lambda Function Information
Retrieve Lambda function metadata, versions, and invocation capabilities.
def get_lambda_function(self, function_name):
"""
Get Lambda function configuration.
Parameters:
- function_name: str, name of the function
Returns:
dict: Function configuration and metadata
"""def get_lambda_function_versions(self, function_name):
"""
Get all versions of a Lambda function.
Parameters:
- function_name: str, name of the function
Returns:
list: List of function versions
"""def invoke_lambda_function(
self,
function_name,
payload=None,
invocation_type='RequestResponse'
):
"""
Invoke a Lambda function.
Parameters:
- function_name: str, name of the function
- payload: dict, input data for the function
- invocation_type: str, 'RequestResponse' (sync) or 'Event' (async)
Returns:
dict: Function response and metadata
"""Version Management
Handle Lambda function versioning and rollback operations.
def rollback_lambda_function_version(
self,
function_name,
versions_back=1,
publish=True
):
"""
Roll back Lambda function to a previous version.
Parameters:
- function_name: str, name of the function
- versions_back: int, number of versions to roll back
- publish: bool, publish the rollback as a new version
Returns:
dict: Updated function configuration
"""API Gateway Management
Deploy and manage API Gateway configurations for Lambda functions.
def deploy_api_gateway(
self,
api_id,
stage_name='dev',
stage_description=None,
cache_cluster_enabled=False,
cache_cluster_size=None,
variables=None,
description=None,
**kwargs
):
"""
Deploy API Gateway stage.
Parameters:
- api_id: str, API Gateway REST API ID
- stage_name: str, name of the deployment stage
- stage_description: str, description of the stage
- cache_cluster_enabled: bool, enable caching
- cache_cluster_size: str, cache cluster size
- variables: dict, stage variables
- description: str, deployment description
Returns:
dict: Deployment configuration
"""def undeploy_api_gateway(self, api_id, stage_name=None):
"""
Remove API Gateway deployment.
Parameters:
- api_id: str, API Gateway REST API ID
- stage_name: str, stage to undeploy (optional)
Returns:
bool: True if removal successful
"""def create_api_gateway_routes(
self,
lambda_arn,
api_id=None,
resource_id=None,
**kwargs
):
"""
Create API Gateway routes for Lambda function.
Parameters:
- lambda_arn: str, ARN of the Lambda function
- api_id: str, API Gateway REST API ID
- resource_id: str, API Gateway resource ID
Returns:
dict: Route configuration
"""Package Management
Create and manage deployment packages for Lambda functions.
def create_lambda_zip(
self,
prefix='lambda_package',
handler_file=None,
use_precompiled_packages=True,
exclude=None,
**kwargs
):
"""
Create deployment ZIP package for Lambda.
Parameters:
- prefix: str, prefix for the ZIP filename
- handler_file: str, path to the handler file
- use_precompiled_packages: bool, use precompiled packages
- exclude: list, files/patterns to exclude
Returns:
str: Path to created ZIP file
"""S3 Upload Operations
Upload deployment packages and other files to S3 buckets.
def upload_to_s3(
self,
file_path,
bucket_name,
key_name=None,
disable_progress=False
):
"""
Upload file to S3 bucket.
Parameters:
- file_path: str, local path to file
- bucket_name: str, S3 bucket name
- key_name: str, S3 object key (optional)
- disable_progress: bool, disable progress display
Returns:
str: S3 URL of uploaded file
"""IAM Role Management
Create and manage IAM roles for Lambda function execution.
def create_iam_roles(self):
"""
Create IAM roles and policies for Lambda execution.
Creates execution role with basic Lambda permissions
and attaches necessary policies for AWS service access.
Returns:
str: IAM role ARN
"""def get_credentials_arn(self):
"""
Get IAM role ARN for Lambda execution.
Retrieves the ARN of the IAM role used for Lambda
function execution and AWS service access.
Returns:
str: IAM role ARN
"""Event Scheduling
Manage CloudWatch Events for scheduled Lambda execution.
def schedule_events(self, lambda_arn, lambda_name, events):
"""
Schedule CloudWatch Events for Lambda function.
Creates CloudWatch Event rules with cron or rate expressions
to trigger Lambda function execution on schedule.
Parameters:
- lambda_arn: str, Lambda function ARN
- lambda_name: str, Lambda function name
- events: list, event configuration with schedules
Returns:
dict: Event rule configuration
"""def unschedule_events(self, events, lambda_name):
"""
Remove scheduled events for Lambda function.
Deletes CloudWatch Event rules and removes
Lambda function triggers for scheduled execution.
Parameters:
- events: list, event configurations to remove
- lambda_name: str, Lambda function name
Returns:
bool: True if removal successful
"""Application Load Balancer Integration
Deploy Lambda functions with Application Load Balancer integration.
def deploy_lambda_alb(self, **kwargs):
"""
Deploy Lambda function with ALB integration.
Creates Application Load Balancer infrastructure and
configures Lambda function as target for ALB requests.
Parameters:
- **kwargs: ALB configuration options
Returns:
dict: ALB deployment configuration
"""def undeploy_lambda_alb(self, **kwargs):
"""
Remove ALB integration for Lambda function.
Removes Application Load Balancer infrastructure
and cleans up ALB-related resources.
Parameters:
- **kwargs: ALB configuration options
Returns:
bool: True if removal successful
"""Domain Management
Manage custom domains and SSL certificates for API Gateway.
def create_domain_name(self, domain, certificate_arn, **kwargs):
"""
Create custom domain for API Gateway.
Sets up custom domain with SSL certificate
for API Gateway REST API endpoints.
Parameters:
- domain: str, custom domain name
- certificate_arn: str, SSL certificate ARN
- **kwargs: Domain configuration options
Returns:
dict: Domain configuration
"""def update_domain_name(self, domain, certificate_arn):
"""
Update SSL certificate for custom domain.
Updates existing custom domain with new
SSL certificate for certificate renewal.
Parameters:
- domain: str, custom domain name
- certificate_arn: str, new SSL certificate ARN
Returns:
dict: Updated domain configuration
"""def get_domain_name(self, domain):
"""
Get custom domain configuration and status.
Retrieves configuration and status information
for existing custom domain setup.
Parameters:
- domain: str, custom domain name
Returns:
dict: Domain configuration and status
"""CloudFormation Integration
Generate and manage CloudFormation stacks for infrastructure.
def create_stack_template(self, **kwargs):
"""
Generate CloudFormation template for deployment.
Creates CloudFormation template with Lambda function,
API Gateway, IAM roles, and other necessary resources.
Parameters:
- **kwargs: Stack template configuration
Returns:
dict: CloudFormation template
"""def update_stack(self, stack_name, template, **kwargs):
"""
Update existing CloudFormation stack.
Updates CloudFormation stack with new template
and configuration changes.
Parameters:
- stack_name: str, CloudFormation stack name
- template: dict, CloudFormation template
- **kwargs: Update configuration
Returns:
dict: Stack update response
"""def delete_stack(self, stack_name):
"""
Delete CloudFormation stack and resources.
Removes CloudFormation stack and all associated
AWS resources.
Parameters:
- stack_name: str, CloudFormation stack name
Returns:
bool: True if deletion initiated successfully
"""API Gateway Advanced Configuration
Advanced API Gateway configuration and management.
def add_binary_support(self, api_id, cors_config=None):
"""
Enable binary content support for API Gateway.
Configures API Gateway to handle binary content types
like images, PDFs, and other non-text responses.
Parameters:
- api_id: str, API Gateway REST API ID
- cors_config: dict, CORS configuration (optional)
Returns:
dict: Binary support configuration
"""def remove_binary_support(self, api_id):
"""
Disable binary content support for API Gateway.
Removes binary content type handling from
API Gateway configuration.
Parameters:
- api_id: str, API Gateway REST API ID
Returns:
bool: True if removal successful
"""def add_api_compression(self, api_id):
"""
Enable response compression for API Gateway.
Configures API Gateway to compress responses
for improved performance and reduced bandwidth.
Parameters:
- api_id: str, API Gateway REST API ID
Returns:
dict: Compression configuration
"""Async Task Infrastructure
Create infrastructure for asynchronous task processing.
def create_async_sns_topic(self, lambda_name):
"""
Create SNS topic for async task processing.
Sets up SNS topic and subscriptions for
asynchronous Lambda task execution.
Parameters:
- lambda_name: str, Lambda function name
Returns:
str: SNS topic ARN
"""def create_async_dynamodb_table(self, lambda_name):
"""
Create DynamoDB table for async task responses.
Sets up DynamoDB table to store responses from
asynchronous Lambda task executions.
Parameters:
- lambda_name: str, Lambda function name
Returns:
str: DynamoDB table name
"""Constants
# AWS regions
API_GATEWAY_REGIONS = [
'us-east-1', 'us-east-2', 'us-west-1', 'us-west-2',
'eu-west-1', 'eu-west-2', 'eu-central-1',
'ap-southeast-1', 'ap-southeast-2', 'ap-northeast-1'
]
LAMBDA_REGIONS = [
'us-east-1', 'us-east-2', 'us-west-1', 'us-west-2',
'eu-west-1', 'eu-west-2', 'eu-central-1',
'ap-southeast-1', 'ap-southeast-2', 'ap-northeast-1'
]
# Architecture constants
X86_ARCHITECTURE = 'x86_64'
ARM_ARCHITECTURE = 'arm64'
VALID_ARCHITECTURES = ('x86_64', 'arm64')
# Default exclusions for ZIP packages
ZIP_EXCLUDES = [
"*.exe",
"*.DS_Store",
"*.Python",
"*.git",
".git/*",
"*.zip",
"*.tar.gz",
"*.hg",
"pip",
"docutils*",
"setuputils*",
"__pycache__/*"
]
# IAM policy templates
ASSUME_POLICY = """{
"Version": "2012-10-17",
"Statement": [
{
"Sid": "",
"Effect": "Allow",
"Principal": {
"Service": [
"apigateway.amazonaws.com",
"lambda.amazonaws.com",
"events.amazonaws.com"
]
},
"Action": "sts:AssumeRole"
}
]
}"""
ATTACH_POLICY = """{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"logs:*"
],
"Resource": "arn:aws:logs:*:*:*"
},
{
"Effect": "Allow",
"Action": [
"lambda:InvokeFunction"
],
"Resource": [
"*"
]
}
]
}"""Install with Tessl CLI
npx tessl i tessl/pypi-zappa