core-aws-cdk 3.1.1

This project contains the commons elements to create infrastructure in AWS using AWS CDK...

core-aws-cdk

This project contains common elements and constructs to create infrastructure in AWS using AWS CDK with Python.

---

Features

• Base Stacks: Pre-configured CDK stacks with tagging support

• Lambda Functions: Simplified Lambda creation with automatic packaging

• S3 Buckets: S3 bucket creation with security best practices

• SQS Queues: Queue creation with dead-letter queue support

• SNS Topics: Topic creation with subscription management

• Network Stack: VPC and networking resource management

• ZIP Asset Packaging: Automatic Lambda ZIP creation with dependencies

Quick Start

Setting Up Environment

1. Install required libraries:

pip install --upgrade pip
pip install virtualenv

2. Create Python virtual environment:

virtualenv --python=python3.12 .venv

3. Activate the virtual environment:

source .venv/bin/activate

Install packages

pip install .
pip install -e ".[dev]"

Check tests and coverage

python manager.py run-tests
python manager.py run-tests --test-type integration
python manager.py run-coverage

# Having proper AWS credentials...
python manager.py run-tests --test-type functional --pattern "*.py"

# Or using `pytest`...
pytest -n auto

Run specific test file:

pytest tests/functional/test_lambda_creation.py

Run specific test:

pytest tests/functional/test_lambda_creation.py::TestLambdaCreation::test_create_and_invoke_lambda_with_inline_code

Run tests in parallel using all CPUs:

pytest -n auto

Run with specific number of workers:

pytest -n 4  # Use 4 parallel workers

Run functional tests with limited parallelism (recommended):

pytest tests/functional/ -n 2  # Avoid AWS rate limits

IMPORTANT: Functional tests deploy real resources to AWS and may incur costs and require AWS credentials configured.

Prerequisites

1. AWS credentials configured.

2. CDK CLI installed npm install -g aws-cdk.

3. Required AWS permissions: * Lambda (create, invoke, delete) * S3 (create bucket, put/get objects, delete) * SNS (create topic, publish) * SQS (create queue, send/receive messages) * CloudFormation (create/update/delete stacks) * IAM (create roles and policies)

Important Notes:

• Tests automatically clean up resources after completion

• Each test uses temporary directories and unique resource names

• Tests include 10-minute timeouts for deployment and cleanup

• All logs captured with DEBUG level for troubleshooting

Architecture Examples

Complete SNS → SQS → Lambda → S3 Integration

from aws_cdk import App, Environment, Duration, CfnOutput
from aws_cdk.aws_lambda import Code, Runtime
from aws_cdk.aws_lambda_event_sources import SqsEventSource
from aws_cdk.aws_sns_subscriptions import SqsSubscription
from core_aws_cdk.stacks.lambdas import BaseLambdaStack
from core_aws_cdk.stacks.s3 import BaseS3Stack
from core_aws_cdk.stacks.sns import BaseSnsStack
from core_aws_cdk.stacks.sqs import BaseSqsStack

class IntegratedStack(
    BaseSnsStack,
    BaseSqsStack,
    BaseLambdaStack,
    BaseS3Stack
):
    """ My Custom Stack """

app = App()

stack = IntegratedStack(
    app,
    "IntegratedStack",
    env=Environment(
        account="123456789",
        region="us-east-1"
    ))

# Create S3 bucket
bucket = stack.create_bucket(
    bucket_id="DataBucket",
    bucket_name=None  # Auto-generate
)

# Create SQS queue with DLQ
queue = stack.create_sqs_queue(
    queue_id="ProcessQueue",
    queue_name="process-queue",
    with_dlq=True,
    dlq_id="ProcessQueueDLQ",
    max_receive_count=3
)

# Create SNS topic
topic = stack.create_sns_topic(
    topic_id="EventTopic",
    topic_name="event-topic"
)

# Subscribe queue to topic
topic.add_subscription(SqsSubscription(queue))

# Create Lambda processor
lambda_function = stack.create_lambda(
    function_id="Processor",
    handler="handler.lambda_handler",
    code=Code.from_asset("./lambda"),
    runtime=Runtime.PYTHON_3_12,
    timeout=Duration.minutes(5),
    environment={"BUCKET_NAME": bucket.bucket_name}
)

# Configure SQS as Lambda trigger
lambda_function.add_event_source(SqsEventSource(queue))

# Grant permissions
bucket.grant_write(lambda_function)

# Export outputs
CfnOutput(stack, "TopicArn", value=topic.topic_arn)
CfnOutput(stack, "BucketName", value=bucket.bucket_name)

app.synth()

Issue: CDK version mismatch

Solution: Ensure CDK CLI version matches library version

npm install -g aws-cdk@latest
cdk --version

Issue: Node.js version warning

Solution: Ensure Node.js v20 or v22 is installed and accessible

node --version
which node

Contributing

Contributions are welcome! Please:

1. Fork the repository

2. Create a feature branch

3. Write tests for new functionality

4. Ensure all tests pass: pytest -n auto

5. Run linting: pylint core_aws_cdk

6. Run security checks: bandit -r core_aws_cdk

7. Submit a pull request

License

This project is licensed under the MIT License. See the LICENSE file for details.

Links

• Documentation: https://core-aws-cdk.readthedocs.io/en/latest/

• Repository: https://gitlab.com/bytecode-solutions/core/core-aws-cdk

• Issues: https://gitlab.com/bytecode-solutions/core/core-aws-cdk/-/issues

• Changelog: https://gitlab.com/bytecode-solutions/core/core-aws-cdk/-/blob/master/CHANGELOG.md

• PyPI: https://pypi.org/project/core-aws-cdk/

Support

For questions or support, please open an issue on GitLab or contact the maintainers.

Authors

• Alejandro Cora González - Initial work - [e-mail]