Server-less Python Web Services for AWS Lambda and API Gateway
AWS integration helpers, framework detection, file operations, and various utility functions for S3 operations, environment management, and deployment assistance. These utilities support Zappa's core functionality and provide convenience functions for common operations.
Automatically detect and configure popular Python web frameworks.
def detect_django_settings():
"""
Auto-detect Django settings module path.
Searches for Django settings in common locations and patterns
to automatically configure Django applications for deployment.
Returns:
str or None: Django settings module path if found
"""def detect_flask_apps():
"""
Auto-detect Flask application instances in code.
Scans Python files for Flask app instances to automatically
configure Flask applications for serverless deployment.
Returns:
list: List of detected Flask app paths
"""Utilities for working with AWS services and S3 operations.
def parse_s3_url(url):
"""
Parse S3 URLs into bucket and key components.
Extracts bucket name and object key from various S3 URL formats
including s3://, https://, and virtual-hosted-style URLs.
Parameters:
- url: str, S3 URL to parse
Returns:
tuple: (bucket_name, key) or (None, None) if invalid
"""def is_valid_bucket_name(bucket):
"""
Validate S3 bucket name format compliance.
Checks bucket name against S3 naming rules including length,
character restrictions, and DNS compliance.
Parameters:
- bucket: str, bucket name to validate
Returns:
bool: True if bucket name is valid
"""def merge_headers(event):
"""
Merge HTTP headers from Lambda event data.
Combines headers from API Gateway event, handling case-insensitive
merging and multi-valued headers properly.
Parameters:
- event: dict, API Gateway Lambda proxy event
Returns:
dict: Merged headers dictionary
"""Utilities for managing Python environments and runtime detection.
def get_venv_from_python_version(python_version=None):
"""
Get virtual environment path for Python version.
Determines appropriate virtual environment path based on
Python version and system configuration.
Parameters:
- python_version: str, Python version string (optional)
Returns:
str: Path to virtual environment
"""def get_runtime_from_python_version(python_version=None):
"""
Map Python version to AWS Lambda runtime identifier.
Converts Python version string to corresponding Lambda
runtime identifier for function configuration.
Parameters:
- python_version: str, Python version string (optional)
Returns:
str: Lambda runtime identifier (e.g., 'python3.9')
"""Enhanced file operations for deployment package management.
def copytree(src, dst, symlinks=False, ignore=None, copy_function=None):
"""
Enhanced directory copying with metadata preservation.
Copies directory tree with support for symbolic links,
ignore patterns, and custom copy functions.
Parameters:
- src: str, source directory path
- dst: str, destination directory path
- symlinks: bool, copy symbolic links as links
- ignore: callable, function to ignore files/directories
- copy_function: callable, custom file copy function
Returns:
str: Destination directory path
"""def contains_python_files_or_subdirs(folder):
"""
Check if directory contains Python files or subdirectories.
Recursively scans directory to determine if it contains
Python source files or subdirectories with Python code.
Parameters:
- folder: str, directory path to check
Returns:
bool: True if Python files or subdirs found
"""def conflicts_with_a_neighbouring_module(module_path):
"""
Detect Python module naming conflicts.
Checks if module name conflicts with neighboring modules
that could cause import resolution issues.
Parameters:
- module_path: str, path to module to check
Returns:
bool: True if naming conflict detected
"""Manage AWS event sources for Lambda functions.
def add_event_source(event_source, function_name, **kwargs):
"""
Add event source trigger to Lambda function.
Configures event source (S3, DynamoDB, Kinesis, etc.) to
trigger Lambda function execution.
Parameters:
- event_source: dict, event source configuration
- function_name: str, Lambda function name
- **kwargs: Additional configuration options
Returns:
dict: Event source mapping configuration
"""def remove_event_source(event_source, function_name):
"""
Remove event source trigger from Lambda function.
Removes configured event source mapping and stops
triggering function from the event source.
Parameters:
- event_source: dict, event source configuration
- function_name: str, Lambda function name
Returns:
bool: True if removal successful
"""def get_event_source_status(event_source, function_name):
"""
Get current event source configuration status.
Retrieves status and configuration of event source mapping
for the specified Lambda function.
Parameters:
- event_source: dict, event source configuration
- function_name: str, Lambda function name
Returns:
dict: Event source status and configuration
"""def get_topic_name(lambda_name):
"""
Generate SNS topic name for Lambda integration.
Creates standardized SNS topic name for Lambda function
based on function name and naming conventions.
Parameters:
- lambda_name: str, Lambda function name
Returns:
str: Generated SNS topic name
"""General-purpose data processing and formatting functions.
def human_size(bytes_size):
"""
Convert byte counts to human-readable format.
Formats byte size with appropriate units (B, KB, MB, GB)
for user-friendly display.
Parameters:
- bytes_size: int, size in bytes
Returns:
str: Human-readable size string (e.g., "1.5 MB")
"""def string_to_timestamp(time_string):
"""
Convert time strings to Unix timestamps.
Parses various time string formats and converts to
Unix timestamp for consistent time handling.
Parameters:
- time_string: str, time string to convert
Returns:
float: Unix timestamp
"""def titlecase_keys(dictionary):
"""
Convert dictionary keys to title case format.
Transforms dictionary keys to title case for consistent
formatting in HTTP headers and API responses.
Parameters:
- dictionary: dict, dictionary to transform
Returns:
dict: Dictionary with title-cased keys
"""Validation utilities for names, data, and configurations.
def validate_name(name):
"""
Validate AWS resource name compliance.
Checks resource name against AWS naming conventions
and restrictions for Lambda functions and other resources.
Parameters:
- name: str, resource name to validate
Returns:
bool: True if name is valid
"""def validate_json_serializable(data):
"""
Validate data structures for JSON serialization.
Checks if data can be properly serialized to JSON,
useful for Lambda payloads and API responses.
Parameters:
- data: any, data to validate
Returns:
bool: True if JSON serializable
"""Check for updates and manage version information.
def check_new_version_available():
"""
Check PyPI for newer Zappa versions.
Queries PyPI API to determine if a newer version of
Zappa is available for upgrade.
Returns:
dict or None: Version information if update available
"""Specialized logging formatters and utilities.
def ApacheNCSAFormatter():
"""
Create Apache NCSA compatible log formatter.
Returns logging formatter that produces Apache NCSA
Common Log Format entries for web request logging.
Returns:
logging.Formatter: Configured log formatter
"""# Default text MIME types for response handling
DEFAULT_TEXT_MIMETYPES = (
'application/json',
'application/javascript',
'application/xml',
'application/vnd.api+json',
'text/css',
'text/html',
'text/javascript',
'text/plain',
'text/xml'
)from zappa.utilities import detect_django_settings, detect_flask_apps
# Auto-detect Django settings
django_settings = detect_django_settings()
if django_settings:
print(f"Found Django settings: {django_settings}")
# Auto-detect Flask apps
flask_apps = detect_flask_apps()
for app_path in flask_apps:
print(f"Found Flask app: {app_path}")from zappa.utilities import parse_s3_url, is_valid_bucket_name
# Parse S3 URLs
bucket, key = parse_s3_url('s3://my-bucket/path/to/file.zip')
print(f"Bucket: {bucket}, Key: {key}")
# Validate bucket names
if is_valid_bucket_name('my-deployment-bucket'):
print("Bucket name is valid")from zappa.utilities import get_runtime_from_python_version, get_venv_from_python_version
# Get Lambda runtime for Python version
runtime = get_runtime_from_python_version('3.9.7')
print(f"Lambda runtime: {runtime}") # python3.9
# Get virtual environment path
venv_path = get_venv_from_python_version('3.9')
print(f"Virtual env: {venv_path}")from zappa.utilities import copytree, contains_python_files_or_subdirs
# Copy directory with exclusions
def ignore_patterns(dir, files):
return [f for f in files if f.endswith('.pyc')]
copytree('src/', 'dist/', ignore=ignore_patterns)
# Check for Python content
if contains_python_files_or_subdirs('my_package/'):
print("Directory contains Python code")from zappa.utilities import human_size, string_to_timestamp
# Format file sizes
size = human_size(1536000) # "1.5 MB"
print(f"Package size: {size}")
# Convert timestamps
timestamp = string_to_timestamp('2023-01-01T12:00:00Z')
print(f"Timestamp: {timestamp}")from zappa.utilities import add_event_source, get_topic_name
# Configure S3 event source
event_source = {
'Events': ['s3:ObjectCreated:*'],
'Bucket': 'my-uploads-bucket',
'Filter': {'Key': {'FilterRules': [{'Name': 'prefix', 'Value': 'uploads/'}]}}
}
add_event_source(event_source, 'my-processor-function')
# Generate SNS topic name
topic_name = get_topic_name('my-function')
print(f"SNS topic: {topic_name}")Install with Tessl CLI
npx tessl i tessl/pypi-zappa