Chapter 3: Basic Constructs Usage

Haiyue
11min
Learning Objectives
  • Master the differences and use cases of L1, L2, and L3 Constructs
  • Learn to create and configure basic AWS resources (S3, Lambda, IAM)
  • Understand dependencies between resources
  • Master parameter passing and configuration management in CDK

Deep Understanding of Construct Levels

Three-Tier Architecture Comparison

🔄 正在渲染 Mermaid 图表...

Choosing the Right Construct Level

ScenarioRecommended LevelReason
Rapid prototypingL3Reduce code, quick validation
Production standard configurationL2Balance usability and control
Special configuration needsL1Complete control over all properties
Legacy system migrationL1Precisely map existing configuration

Complete S3 Bucket Examples

L1 Level: Full Control

from aws_cdk import (
    Stack,
    aws_s3 as s3,
    CfnOutput
)
from constructs import Construct

class S3L1Stack(Stack):
    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        # L1 construct: Direct CloudFormation resource
        bucket_l1 = s3.CfnBucket(
            self,
            "MyBucketL1",
            bucket_name="my-app-data-bucket-l1",
            # Versioning configuration
            versioning_configuration=s3.CfnBucket.VersioningConfigurationProperty(
                status="Enabled"
            ),
            # Encryption configuration
            bucket_encryption=s3.CfnBucket.BucketEncryptionProperty(
                server_side_encryption_configuration=[
                    s3.CfnBucket.ServerSideEncryptionRuleProperty(
                        server_side_encryption_by_default=s3.CfnBucket.ServerSideEncryptionByDefaultProperty(
                            sse_algorithm="AES256"
                        )
                    )
                ]
            ),
            # Public access block
            public_access_block_configuration=s3.CfnBucket.PublicAccessBlockConfigurationProperty(
                block_public_acls=True,
                block_public_policy=True,
                ignore_public_acls=True,
                restrict_public_buckets=True
            ),
            # Lifecycle rules
            lifecycle_configuration=s3.CfnBucket.LifecycleConfigurationProperty(
                rules=[
                    s3.CfnBucket.RuleProperty(
                        id="DeleteOldVersions",
                        status="Enabled",
                        noncurrent_version_expiration=s3.CfnBucket.NoncurrentVersionExpirationProperty(
                            noncurrent_days=30
                        )
                    )
                ]
            ),
            # Notification configuration
            notification_configuration=s3.CfnBucket.NotificationConfigurationProperty(
                # Can configure Lambda, SQS, SNS notifications
            )
        )

        CfnOutput(self, "BucketNameL1", value=bucket_l1.ref)

L2 Level: Best Practices

from aws_cdk import (
    Stack,
    aws_s3 as s3,
    aws_s3_notifications as s3n,
    aws_lambda as lambda_,
    RemovalPolicy,
    Duration,
    CfnOutput
)
from constructs import Construct

class S3L2Stack(Stack):
    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        # L2 construct: Simplified API with best practices
        bucket_l2 = s3.Bucket(
            self,
            "MyBucketL2",
            bucket_name="my-app-data-bucket-l2",
            # Versioning (simplified)
            versioned=True,
            # Encryption (simplified)
            encryption=s3.BucketEncryption.S3_MANAGED,
            # Public access (simplified)
            block_public_access=s3.BlockPublicAccess.BLOCK_ALL,
            # Auto-delete objects (for development)
            auto_delete_objects=True,
            removal_policy=RemovalPolicy.DESTROY,
            # Lifecycle rules (simplified)
            lifecycle_rules=[
                s3.LifecycleRule(
                    id="delete-old-versions",
                    enabled=True,
                    noncurrent_version_expiration=Duration.days(30),
                    abort_incomplete_multipart_upload_after=Duration.days(1)
                )
            ],
            # CORS configuration
            cors=[
                s3.CorsRule(
                    allowed_methods=[s3.HttpMethods.GET, s3.HttpMethods.POST],
                    allowed_origins=["https://example.com"],
                    allowed_headers=["*"],
                    max_age=3000
                )
            ]
        )

        # Create processor function
        processor_function = lambda_.Function(
            self,
            "S3Processor",
            runtime=lambda_.Runtime.PYTHON_3_9,
            handler="index.handler",
            code=lambda_.Code.from_inline("""
def handler(event, context):
    for record in event['Records']:
        bucket = record['s3']['bucket']['name']
        key = record['s3']['object']['key']
        print(f"File {key} uploaded to bucket {bucket}")
    return {'statusCode': 200}
            """)
        )

        # Add S3 event notification (L2 advantage: simple API)
        bucket_l2.add_event_notification(
            s3.EventType.OBJECT_CREATED,
            s3n.LambdaDestination(processor_function),
            s3.NotificationKeyFilter(prefix="uploads/", suffix=".jpg")
        )

        # Grant Lambda read permissions
        bucket_l2.grant_read(processor_function)

        CfnOutput(self, "BucketNameL2", value=bucket_l2.bucket_name)
        CfnOutput(self, "BucketArnL2", value=bucket_l2.bucket_arn)

L3 Level: Pattern Composition

from constructs import Construct
from aws_cdk import (
    Stack,
    aws_s3 as s3,
    aws_s3_deployment as s3deploy,
    aws_cloudfront as cloudfront,
    aws_cloudfront_origins as origins,
    RemovalPolicy,
    CfnOutput
)

class StaticWebsiteConstruct(Construct):
    """L3 construct: Static website hosting pattern"""

    def __init__(self, scope: Construct, construct_id: str,
                 domain_name: str = None, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        # Website content bucket
        self.website_bucket = s3.Bucket(
            self,
            "WebsiteBucket",
            # Enable static website hosting
            website_index_document="index.html",
            website_error_document="error.html",
            # Public read access
            public_read_access=True,
            block_public_access=s3.BlockPublicAccess(
                block_public_acls=False,
                block_public_policy=False,
                ignore_public_acls=False,
                restrict_public_buckets=False
            ),
            removal_policy=RemovalPolicy.DESTROY,
            auto_delete_objects=True
        )

        # Logs bucket
        self.logs_bucket = s3.Bucket(
            self,
            "LogsBucket",
            removal_policy=RemovalPolicy.DESTROY,
            auto_delete_objects=True
        )

        # CloudFront distribution
        self.distribution = cloudfront.Distribution(
            self,
            "WebsiteDistribution",
            default_behavior=cloudfront.BehaviorOptions(
                origin=origins.S3Origin(self.website_bucket),
                viewer_protocol_policy=cloudfront.ViewerProtocolPolicy.REDIRECT_TO_HTTPS,
                cache_policy=cloudfront.CachePolicy.CACHING_OPTIMIZED
            ),
            # Default root object
            default_root_object="index.html",
            # Error pages
            error_responses=[
                cloudfront.ErrorResponse(
                    http_status=404,
                    response_http_status=404,
                    response_page_path="/error.html"
                )
            ],
            # Access logs
            enable_logging=True,
            log_bucket=self.logs_bucket,
            log_file_prefix="access-logs/"
        )

    @property
    def bucket_name(self) -> str:
        return self.website_bucket.bucket_name

    @property
    def distribution_domain_name(self) -> str:
        return self.distribution.distribution_domain_name

    @property
    def website_url(self) -> str:
        return f"https://{self.distribution.distribution_domain_name}"

class S3L3Stack(Stack):
    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        # Use L3 construct
        website = StaticWebsiteConstruct(
            self,
            "MyWebsite",
            domain_name="example.com"
        )

        # Deploy website content (if website-dist directory exists)
        # s3deploy.BucketDeployment(
        #     self,
        #     "DeployWebsite",
        #     sources=[s3deploy.Source.asset("./website-dist")],
        #     destination_bucket=website.website_bucket,
        #     distribution=website.distribution,
        #     distribution_paths=["/*"]
        # )

        CfnOutput(self, "WebsiteURL", value=website.website_url)
        CfnOutput(self, "BucketName", value=website.bucket_name)

This chapter comprehensively introduces the use of basic Constructs in CDK, including selecting different levels, resource configuration, permission management, and dependency relationships. Mastering these fundamentals will lay a solid foundation for building complex cloud infrastructure.