**A production-ready, configuration-driven Lambda template that works with any upstream service (API Gateway, Tyk Gateway, EventBridge, SQS) and integrates with any downstream AWS service (DynamoDB, S3, SQS, SNS, and more).**This repository contains a simple AWS Lambda microservice implemented in
Python. It demonstrates how to use some of the essential and advanced
π‘ New to AWS Lambda? Check out UNDER_THE_HOOD.md for a beginner-friendly explanation of how everything works!features of AWS Lambda while keeping the code and infrastructure
definition clear and maintainable. The project uses a custom
---template generator (written in Python) to produce a CloudFormation
template at build time.
Instead of writing Lambda functions from scratch every time, this template lets you:
- Environment variables and configuration: The Lambda handler reads
-
Configure what you need in a YAML file (
config.yaml) greeting parameters and secret configuration from environment -
Write your business logic in Python (
handler.py) variables. This decouples configuration from code and allows -
Generate infrastructure code automatically different values per environment without redeploying. Concurrency
-
Deploy to AWS in minutes Labs recommends using versions, aliases and environment variables to
manage configuration across stagesγ380550635739302β L52-L83γ.
No need to manually write IAM policies, CloudFormation resources, or integration code!- Lambda layer: A separate layer contains a shared reverse_string
helper. Layers promote code reuse and keep the main function
--- package smallγ989117450241150β L296-L297γ.
- Reserved concurrency: The function reserves a fixed number of
accountβs concurrency limit and to guarantee availability when other
-
β API Gateway - Build REST APIs- Versioning and aliases: The generated CloudFormation template
-
β Lambda Function URL - Connect to Tyk Gateway or any external gateway publishes a version of the function and creates a
prodalias -
β EventBridge - Event-driven architectures pointing to it. Aliases abstract function versions and are the
-
β SQS - Process queues recommended way to implement canary or blueβgreen deployments
-
β Direct Calls - Invoke from other services γ121160137832048β L316-L376γ.
-
Secrets Manager integration: The handler optionally retrieves a
Integrate with Any AWS Service (Downstream) secret from AWS Secrets Manager if SECRETS_ENABLED=true and
-
β DynamoDB - NoSQL database (auto-creates tables!)
SECRET_NAMEare set. To keep costs down the secret value is -
β S3 - File storage (auto-creates buckets!) cached between invocations as suggested by Concurrency Labsγ380550635739302β L52-L83γ.
-
β SQS - Message queues
-
β SNS - Notifications## Repository structure
-
β Secrets Manager - Secure secret storage
-
β RDS - Relational databases-
lambda_function/handler.pyβ The Lambda handler implementing themicroservice logic.
-
β Monitoring - CloudWatch Logs, Metrics, Alarms-
generate_template.pyβ Script that reads the handler and layer -
β Tracing - X-Ray distributed tracing sources and produces a CloudFormation template (
template.yaml). -
β Error Handling - Automatic error logging and alerting-
tests/β Pytest-based unit tests that exercise the handler logic -
β Cost Control - Concurrency limits, efficient caching locally.
-
β Security - Auto-generated IAM policies with least-privilege access
To run the code and tests locally you need PythonΒ 3.11 or newer. No
tests depend on pytest.
pip install pyyaml boto3
```Run the generator script from the project root:
### Step 2: Choose a Template```bash
Pick the example that matches your use case:python generate_template.py
# For a REST API with DynamoDBThis will produce `template.yaml` in the same directory. The
cp examples/rest-api.yaml config.yamltemplate is fully selfβcontained and can be deployed to AWS via the
AWS Console, AWS CLI (`aws cloudformation deploy`) or the CDK
# For Tyk Gateway integrationbootstrap process.
cp examples/tyk-integration.yaml config.yaml
```## Running locally
### Step 3: Customize Your ProjectAlthough this Lambda is designed for AWS, you can invoke the handler
Edit `config.yaml` and change the project name:locally for rapid testing. For example:
```yaml
project:```python
name: my-awesome-api # β Change this!from lambda_function.handler import handler
environment: dev
```event = {"name": "World", "numbers": [1, 2, 3]}
response = handler(event, None)
### Step 4: Generate Infrastructureprint(response)
```bash```
python3 generate_advanced_template.py
```If you wish to simulate Secrets Manager retrieval locally, set the
This creates `template.yaml` with all your AWS resources!environment variables `SECRETS_ENABLED=false` and `SECRET_VALUE` to a
desired value before invoking the handler.
### Step 5: Deploy
```bash## Testing
aws cloudformation deploy \
--template-file template.yaml \Install `pytest` (if not already installed) and run the tests:
--stack-name my-awesome-api-stack \
--capabilities CAPABILITY_IAM```bash
```pip install pytest
pytest -q
### Step 6: Test```
```bash
# Get your API URLThe tests cover default behaviour, summing arrays of numbers, reading
aws cloudformation describe-stacks \custom environment variables and falling back to a local secret when
--stack-name my-awesome-api-stack \Secrets Manager is disabled.
--query 'Stacks[0].Outputs'
# Test it!
curl https://your-api-url.amazonaws.com/prodDone! π You now have a production-ready Lambda with monitoring and integrations!
What you want: A REST API that stores data in DynamoDB
Configure (config.yaml):
project:
name: user-api
environment: dev
upstream:
type: apigateway # Creates API Gateway
api_gateway:
enabled: true
downstream:
dynamodb:
enabled: true
create_table: true # Auto-creates table!
partition_key: user_idCode (lambda_function/handler.py):
from base_handler import BaseLambdaHandler
import uuid
class MyAPIHandler(BaseLambdaHandler):
def process(self, event, context):
method = event.get('httpMethod')
if method == 'POST':
# Create user
body = self._parse_api_gateway_body(event)
self.put_dynamodb_item({
'user_id': str(uuid.uuid4()),
'name': body['name'],
'email': body['email']
})
return {
'statusCode': 201,
'body': {'message': 'User created'}
}
elif method == 'GET':
# Get user
user_id = event['pathParameters']['id']
user = self.get_dynamodb_item({'user_id': user_id})
if not user:
return {'statusCode': 404, 'body': {'error': 'Not found'}}
return {'statusCode': 200, 'body': user}Deploy:
python3 generate_advanced_template.py
aws cloudformation deploy --template-file template.yaml \
--stack-name user-api-stack --capabilities CAPABILITY_IAMWhat you want: Connect your Lambda to Tyk Gateway (or any external API gateway)
Configure (config.yaml):
project:
name: public-api
environment: prod
upstream:
type: http # Use Function URL instead of API Gateway
function_url:
enabled: true
auth_type: NONE # Tyk handles authentication
cors:
allow_origins: ["https://yourdomain.com"]
downstream:
dynamodb:
enabled: true
create_table: true
secrets_manager:
enabled: true
secret_name: "api-keys"Deploy and Get URL:
python3 generate_advanced_template.py
aws cloudformation deploy --template-file template.yaml \
--stack-name public-api-stack --capabilities CAPABILITY_IAM
# Get the Function URL
aws cloudformation describe-stacks \
--stack-name public-api-stack \
--query 'Stacks[0].Outputs[?OutputKey==`FunctionUrl`].OutputValue' \
--output textConfigure Tyk: Point Tyk to your Lambda Function URL and you're done!
What you want: Run a task every hour (like sending reports)
Configure (config.yaml):
project:
name: daily-report
upstream:
type: eventbridge
eventbridge:
enabled: true
schedule_expression: "rate(1 hour)" # Or use cron
downstream:
s3:
enabled: true
create_bucket: true
sns:
enabled: true
create_topic: trueCode:
class ReportHandler(BaseLambdaHandler):
def process(self, event, context):
# Generate report
report = self.generate_report()
# Save to S3
self.upload_to_s3(
report.encode('utf-8'),
f"reports/{self.correlation_id}.pdf"
)
# Send notification
self.publish_sns_message(
f"Report generated: {self.correlation_id}",
subject="Daily Report"
)
return {'statusCode': 200}The config.yaml file controls everything. Here are the main sections:
project:
name: my-service # Your service name
description: "My API" # Description
environment: dev # dev, staging, or prodlambda:
runtime: python3.11 # Python version
timeout: 30 # Maximum runtime in seconds
memory_size: 256 # Memory in MB
architectures: x86_64 # x86_64 or arm64Option 1: API Gateway (Traditional REST API)
upstream:
type: apigateway
api_gateway:
enabled: true
enable_cors: true
throttle_rate_limit: 1000Option 2: Function URL (For Tyk or External Gateways)
upstream:
type: http
function_url:
enabled: true
auth_type: NONE # Or AWS_IAM
cors:
allow_origins: ["*"]Option 3: EventBridge (Events or Scheduled)
upstream:
type: eventbridge
eventbridge:
enabled: true
schedule_expression: "rate(5 minutes)"Option 4: SQS Queue
upstream:
type: sqs
sqs:
enabled: true
batch_size: 10DynamoDB (Database)
downstream:
dynamodb:
enabled: true
create_table: true # Automatically creates table
partition_key: id # Primary keyS3 (File Storage)
downstream:
s3:
enabled: true
create_bucket: true # Automatically creates bucket
enable_encryption: trueSQS (Send Messages)
downstream:
sqs:
enabled: true
create_queue: trueSNS (Notifications)
downstream:
sns:
enabled: true
create_topic: trueSecrets Manager
downstream:
secrets_manager:
enabled: true
secret_name: "my-api-keys"observability:
xray:
enabled: true # Distributed tracing
cloudwatch:
log_retention_days: 7
alarms:
enabled: true # Auto-create alarms
error_threshold: 5When you extend BaseLambdaHandler, you get easy-to-use helper methods:
# Save data
self.put_dynamodb_item({'id': '123', 'name': 'John'})
# Get data
item = self.get_dynamodb_item({'id': '123'})# Upload file
self.upload_to_s3(data, 'folder/file.json')
# Download file
data = self.get_from_s3('folder/file.json')# Send message
self.send_sqs_message('{"action": "process"}')# Send notification
self.publish_sns_message('Alert!', subject='Warning')# Get secret (automatically cached)
api_key = self.get_secret('my-api-key')# Track custom metrics
self.publish_custom_metric('ItemsProcessed', 1.0)# Install everything
make install
# Run tests
make test
# Generate CloudFormation template
make generate
# Or: python3 generate_advanced_template.py
# Deploy to AWS
make deploy
# View logs (real-time)
aws logs tail /aws/lambda/my-service-dev --follow
# Test locally
python3 -c "
from lambda_function.handler import handler
response = handler({'name': 'Test'}, None)
print(response)
"Create separate configs for dev, staging, and prod:
# Create environment-specific configs
cp config.yaml config-dev.yaml
cp config.yaml config-staging.yaml
cp config.yaml config-prod.yaml
# Edit each to set: environment: dev/staging/prod
# Deploy to each
cp config-dev.yaml config.yaml && make deploy
cp config-staging.yaml config.yaml && make deploy
cp config-prod.yaml config.yaml && make deploypip install pyyaml boto3Make sure to include --capabilities CAPABILITY_IAM:
aws cloudformation deploy \
--template-file template.yaml \
--stack-name my-stack \
--capabilities CAPABILITY_IAM- Check that DynamoDB is enabled in
config.yaml - Verify the table was created: Check AWS Console > DynamoDB
- Check CloudWatch Logs for specific errors
# Real-time logs
aws logs tail /aws/lambda/my-function-name --follow
# Search for errors
aws logs filter-log-events \
--log-group-name /aws/lambda/my-function-name \
--filter-pattern "ERROR"- Increase
timeoutinconfig.yaml(default is 30 seconds) - Check CloudWatch X-Ray traces to see where time is spent
- Make sure your database queries are efficient
lambda-template/
βββ config.yaml # β Configure everything here
βββ generate_advanced_template.py # β Generates CloudFormation
βββ template.yaml # Generated AWS infrastructure
β
βββ lambda_function/
β βββ handler.py # β Write your code here
β βββ base_handler.py # Framework (don't edit)
β βββ __init__.py
β
βββ layer/
β βββ python/
β βββ utils.py # Shared utilities
β
βββ tests/
β βββ test_advanced_handler.py # Tests
β βββ events/ # Sample API Gateway/SQS events
β
βββ examples/
β βββ rest-api.yaml # Example: REST API
β βββ tyk-integration.yaml # Example: Tyk Gateway
β
βββ Makefile # Quick commands
βββ requirements.txt # Python dependencies
βββ README.md # β You are here
βββ UNDER_THE_HOOD.md # Technical deep-dive
- UNDER_THE_HOOD.md - Beginner-friendly explanation of how everything works
- examples/ - More configuration examples
- AWS Lambda Guide: https://docs.aws.amazon.com/lambda/
- Tyk Gateway Docs: https://tyk.io/docs/
- Start simple - Use an example config, then add features as needed
- Test locally first - Use
make testbefore deploying - Enable X-Ray - Helps debug performance issues
- Use CloudWatch Alarms - Get notified when errors happen
- Never hardcode secrets - Use Secrets Manager
- Use correlation IDs - Track requests across services (built-in!)
- Set appropriate timeouts - Don't pay for runaway functions
- Monitor costs - Check AWS CloudWatch billing alerts
Q: Do I need to know CloudFormation? A: No! The template generates it for you.
Q: Can I use this with Tyk Gateway?
A: Yes! Use upstream.type: http with function_url.enabled: true
Q: How much does this cost? A: AWS Lambda free tier includes 1M requests/month. After that, it's pay-per-use.
Q: Can I customize the generated template?
A: Yes, but it's better to add to config.yaml so it's reproducible.
Q: Is this production-ready? A: Yes! Includes monitoring, alarms, error handling, and security best practices.
Q: Can I use other programming languages? A: The framework is Python, but you can modify it for Node.js, Go, etc.
This template handles the infrastructure so you can focus on your business logic.
Next Steps:
- Try the Quick Start above
- Read UNDER_THE_HOOD.md to understand how it works
- Check out examples in the
examples/folder - Customize
config.yamlfor your needs - Deploy and build something awesome!
Happy coding! π
MIT License - Use this for any project!