CDK Construct Team Productivity

CDK Construct Team Productivity คืออะไร

AWS CDK (Cloud Development Kit) เป็น framework สำหรับสร้าง cloud infrastructure ด้วย programming languages เช่น TypeScript, Python, Java และ Go CDK Constructs คือ building blocks ที่ encapsulate AWS resources และ configurations ไว้เป็น reusable components Team Productivity หมายถึงการใช้ CDK Constructs เพิ่มประสิทธิภาพการทำงานของทีม ลด boilerplate code สร้าง golden paths สำหรับ infrastructure และทำให้ทีมทุกู้คืน deploy ได้อย่างรวดเร็วและปลอดภัยตาม best practices ขององค์กร

CDK Constructs พื้นฐาน

# cdk_basics.py — CDK Construct fundamentals
import json

class CDKBasics:
    CONSTRUCT_LEVELS = {
        "L1": {
            "name": "L1 — CloudFormation Resources",
            "description": "1:1 mapping กับ CloudFormation resources (Cfn prefix)",
            "example": "CfnBucket, CfnFunction, CfnTable",
            "use": "เมื่อต้องการ full control, ไม่มี L2 construct",
        },
        "L2": {
            "name": "L2 — Curated Constructs",
            "description": "AWS-maintained constructs ที่มี sensible defaults + helper methods",
            "example": "s3.Bucket, lambda_.Function, dynamodb.Table",
            "use": "ใช้บ่อยที่สุด — balance ระหว่าง convenience และ control",
        },
        "L3": {
            "name": "L3 — Patterns (Custom Constructs)",
            "description": "Opinionated patterns ที่รวมหลาย resources เป็น solution",
            "example": "ApiGatewayToLambda, LambdaToDynamoDB",
            "use": "เพิ่ม productivity — reusable across teams",
        },
    }

    PYTHON_EXAMPLE = """
# cdk_stack.py — CDK Stack example (Python)
from aws_cdk import (
    Stack, Duration, RemovalPolicy,
    aws_s3 as s3,
    aws_lambda as lambda_,
    aws_dynamodb as dynamodb,
    aws_apigateway as apigw,
)
from constructs import Construct

class ApiStack(Stack):
    def __init__(self, scope: Construct, id: str, **kwargs):
        super().__init__(scope, id, **kwargs)
        
        # DynamoDB table
        table = dynamodb.Table(self, "Items",
            partition_key=dynamodb.Attribute(
                name="id", type=dynamodb.AttributeType.STRING
            ),
            billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST,
            removal_policy=RemovalPolicy.DESTROY,
        )
        
        # Lambda function
        handler = lambda_.Function(self, "Handler",
            runtime=lambda_.Runtime.PYTHON_3_12,
            code=lambda_.Code.from_asset("lambda"),
            handler="index.handler",
            timeout=Duration.seconds(30),
            environment={"TABLE_NAME": table.table_name},
        )
        table.grant_read_write_data(handler)
        
        # API Gateway
        api = apigw.RestApi(self, "Api",
            rest_api_name="Items API",
            deploy_options=apigw.StageOptions(stage_name="prod"),
        )
        items = api.root.add_resource("items")
        items.add_method("GET", apigw.LambdaIntegration(handler))
        items.add_method("POST", apigw.LambdaIntegration(handler))
"""

    def show_levels(self):
        print("=== CDK Construct Levels ===\n")
        for key, level in self.CONSTRUCT_LEVELS.items():
            print(f"[{level['name']}]")
            print(f"  {level['description']}")
            print(f"  Example: {level['example']}")
            print()

    def show_example(self):
        print("=== Python CDK Example ===")
        print(self.PYTHON_EXAMPLE[:600])

cdk = CDKBasics()
cdk.show_levels()
cdk.show_example()

Custom L3 Constructs สำหรับทีม

# custom_constructs.py — Team productivity constructs
import json

class TeamConstructs:
    MICROSERVICE = """
# constructs/microservice.py — Reusable microservice construct
from aws_cdk import (
    Stack, Duration, RemovalPolicy,
    aws_lambda as lambda_,
    aws_dynamodb as dynamodb,
    aws_apigateway as apigw,
    aws_logs as logs,
    aws_iam as iam,
)
from constructs import Construct

class Microservice(Construct):
    def __init__(self, scope: Construct, id: str, *,
                 service_name: str,
                 handler_path: str,
                 runtime: lambda_.Runtime = lambda_.Runtime.PYTHON_3_12,
                 timeout: Duration = Duration.seconds(30),
                 memory_size: int = 256,
                 with_dynamodb: bool = True,
                 with_api: bool = True,
                 environment: dict = None):
        super().__init__(scope, id)
        
        env = environment or {}
        
        # Lambda
        self.function = lambda_.Function(self, "Function",
            function_name=f"{service_name}-handler",
            runtime=runtime,
            code=lambda_.Code.from_asset(handler_path),
            handler="index.handler",
            timeout=timeout,
            memory_size=memory_size,
            environment=env,
            log_retention=logs.RetentionDays.ONE_MONTH,
            tracing=lambda_.Tracing.ACTIVE,
        )
        
        # DynamoDB (optional)
        if with_dynamodb:
            self.table = dynamodb.Table(self, "Table",
                table_name=f"{service_name}-data",
                partition_key=dynamodb.Attribute(
                    name="pk", type=dynamodb.AttributeType.STRING),
                sort_key=dynamodb.Attribute(
                    name="sk", type=dynamodb.AttributeType.STRING),
                billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST,
            )
            self.table.grant_read_write_data(self.function)
            self.function.add_environment("TABLE_NAME", self.table.table_name)
        
        # API Gateway (optional)
        if with_api:
            self.api = apigw.RestApi(self, "Api",
                rest_api_name=f"{service_name}-api",
            )
            resource = self.api.root.add_resource(service_name)
            resource.add_method("ANY", apigw.LambdaIntegration(self.function))

# Usage — สร้าง microservice ใน 5 บรรทัด!
# user_svc = Microservice(self, "UserService",
#     service_name="user",
#     handler_path="services/user",
#     with_dynamodb=True,
#     with_api=True,
# )
"""

    CONSTRUCTS_LIB = {
        "microservice": {"name": "Microservice", "saves": "~100 lines", "includes": "Lambda + DynamoDB + API GW + Logging"},
        "static_site": {"name": "StaticWebsite", "saves": "~80 lines", "includes": "S3 + CloudFront + Route53 + ACM cert"},
        "data_pipeline": {"name": "DataPipeline", "saves": "~120 lines", "includes": "S3 + Glue + Athena + EventBridge"},
        "monitoring": {"name": "ServiceMonitoring", "saves": "~60 lines", "includes": "CloudWatch Alarms + Dashboard + SNS"},
        "vpc_setup": {"name": "StandardVPC", "saves": "~90 lines", "includes": "VPC + subnets + NAT + security groups"},
    }

    def show_microservice(self):
        print("=== Microservice Construct ===")
        print(self.MICROSERVICE[:600])

    def show_library(self):
        print(f"\n=== Team Construct Library ===")
        for key, c in self.CONSTRUCTS_LIB.items():
            print(f"  [{c['name']}] Saves {c['saves']} | {c['includes']}")

tc = TeamConstructs()
tc.show_microservice()
tc.show_library()

Construct Library Management

# library.py — Managing construct library
import json

class ConstructLibrary:
    STRUCTURE = """
    company-cdk-constructs/
    ├── constructs/
    │   ├── __init__.py
    │   ├── microservice.py
    │   ├── static_website.py
    │   ├── data_pipeline.py
    │   ├── monitoring.py
    │   └── vpc.py
    ├── tests/
    │   ├── test_microservice.py
    │   ├── test_static_website.py
    │   └── snapshots/
    ├── examples/
    │   ├── simple_api/
    │   └── full_stack/
    ├── setup.py
    ├── pyproject.toml
    └── README.md
    """

    PUBLISHING = """
# pyproject.toml — Package configuration
[build-system]
requires = ["setuptools>=68.0"]

[project]
name = "company-cdk-constructs"
version = "1.5.0"
dependencies = [
    "aws-cdk-lib>=2.100.0",
    "constructs>=10.0.0",
]

# Publish to internal PyPI
# pip install twine
# python -m build
# twine upload --repository internal dist/*

# Usage in other projects
# pip install company-cdk-constructs
# from company_constructs import Microservice, StaticWebsite
"""

    VERSIONING = {
        "semver": "Major.Minor.Patch (เช่น 1.5.0)",
        "breaking": "Major bump: เปลี่ยน API ที่ไม่ backward compatible",
        "feature": "Minor bump: เพิ่ม feature ใหม่ (backward compatible)",
        "fix": "Patch bump: bug fix, documentation",
        "changelog": "CHANGELOG.md: document ทุก change",
    }

    def show_structure(self):
        print("=== Library Structure ===")
        print(self.STRUCTURE)

    def show_publishing(self):
        print("=== Publishing ===")
        print(self.PUBLISHING[:400])

    def show_versioning(self):
        print(f"\n=== Versioning ===")
        for key, desc in self.VERSIONING.items():
            print(f"  [{key}] {desc}")

lib = ConstructLibrary()
lib.show_structure()
lib.show_publishing()
lib.show_versioning()

Testing & Quality

# testing.py — CDK construct testing
import json
import random

class CDKTesting:
    TESTING_TYPES = {
        "snapshot": {
            "name": "Snapshot Tests",
            "description": "เปรียบเทียบ CloudFormation template กับ snapshot ที่บันทึกไว้",
            "command": "pytest tests/ -k snapshot",
        },
        "fine_grained": {
            "name": "Fine-Grained Assertions",
            "description": "ตรวจสอบว่า template มี resources และ properties ที่ต้องการ",
            "command": "pytest tests/ -k assertion",
        },
        "validation": {
            "name": "Validation Tests",
            "description": "ตรวจ input validation, error handling",
            "command": "pytest tests/ -k validation",
        },
    }

    TEST_EXAMPLE = """
# tests/test_microservice.py
import aws_cdk as cdk
from aws_cdk.assertions import Template, Match
from constructs.microservice import Microservice

def test_microservice_creates_lambda():
    app = cdk.App()
    stack = cdk.Stack(app, "TestStack")
    
    Microservice(stack, "TestService",
        service_name="test",
        handler_path="lambda",
        with_dynamodb=True,
        with_api=True,
    )
    
    template = Template.from_stack(stack)
    
    # Assert Lambda function exists
    template.has_resource_properties("AWS::Lambda::Function", {
        "FunctionName": "test-handler",
        "Runtime": "python3.12",
        "Timeout": 30,
        "TracingConfig": {"Mode": "Active"},
    })
    
    # Assert DynamoDB table exists
    template.has_resource_properties("AWS::DynamoDB::Table", {
        "TableName": "test-data",
        "BillingMode": "PAY_PER_REQUEST",
    })
    
    # Assert API Gateway exists
    template.resource_count_is("AWS::ApiGateway::RestApi", 1)
    
    # Assert IAM permissions
    template.has_resource_properties("AWS::IAM::Policy",
        Match.object_like({
            "PolicyDocument": Match.object_like({
                "Statement": Match.array_with([
                    Match.object_like({
                        "Action": Match.array_with(["dynamodb:GetItem"]),
                    })
                ])
            })
        })
    )
"""

    def show_types(self):
        print("=== Testing Types ===\n")
        for key, t in self.TESTING_TYPES.items():
            print(f"[{t['name']}] {t['description']}")

    def show_example(self):
        print(f"\n=== Test Example ===")
        print(self.TEST_EXAMPLE[:600])

    def metrics(self):
        print(f"\n=== Quality Metrics ===")
        metrics = {
            "Construct count": random.randint(8, 20),
            "Test coverage": f"{random.randint(85, 98)}%",
            "Teams using": random.randint(3, 15),
            "Deployments/week": random.randint(20, 100),
            "Lines saved (est)": f"{random.randint(5, 30)}K",
        }
        for m, v in metrics.items():
            print(f"  {m}: {v}")

test = CDKTesting()
test.show_types()
test.show_example()
test.metrics()

Productivity Impact

# productivity.py — Measuring productivity impact
import json
import random

class ProductivityImpact:
    BEFORE_AFTER = {
        "new_service": {"before": "2-3 days", "after": "2-4 hours", "improvement": "80-90%"},
        "static_site": {"before": "1-2 days", "after": "30 minutes", "improvement": "90%+"},
        "monitoring": {"before": "4-8 hours", "after": "15 minutes", "improvement": "95%+"},
        "vpc_setup": {"before": "1 day", "after": "1 hour", "improvement": "85%"},
        "data_pipeline": {"before": "3-5 days", "after": "4-8 hours", "improvement": "75%"},
    }

    def show_impact(self):
        print("=== Productivity Impact ===\n")
        print(f"{'Task':<20} {'Before':<15} {'After':<15} {'Improvement'}")
        print(f"{'-'*65}")
        for task, data in self.BEFORE_AFTER.items():
            print(f"  {task:<18} {data['before']:<15} {data['after']:<15} {data['improvement']}")

    def roi(self):
        print(f"\n=== ROI Calculator ===")
        devs = 20
        hours_saved_per_dev = random.randint(5, 15)
        hourly_rate = 1500  # THB
        monthly = devs * hours_saved_per_dev * hourly_rate
        yearly = monthly * 12
        print(f"  Developers: {devs}")
        print(f"  Hours saved/dev/month: {hours_saved_per_dev}")
        print(f"  Monthly savings: {monthly:,.0f} THB")
        print(f"  Yearly savings: {yearly:,.0f} THB")

prod = ProductivityImpact()
prod.show_impact()
prod.roi()

FAQ - คำถามที่พบบ่อย

Q: CDK กับ Terraform อันไหนดีกว่า?

A: CDK: ใช้ programming language จริง (loops, conditions, classes), AWS-native, L3 constructs Terraform: multi-cloud, HCL declarative, mature ecosystem, state management ใช้ CDK: AWS only, ทีม developers, complex logic ใช้ Terraform: multi-cloud, ops team, declarative preference

Q: ควรสร้าง custom constructs เมื่อไหร่?

A: เมื่อ: pattern ใช้ซ้ำ 3+ ครั้ง, ต้อง enforce org standards, ลด boilerplate ไม่ควร: one-off infrastructure, ยัง explore อยู่, team เล็กมาก (1-2 คน) Rule of Three: ถ้าทำซ้ำ 3 ครั้ง → extract เป็น construct

Q: CDK Construct Library distribute อย่างไร?

A: Internal PyPI/npm: publish package สำหรับทีม Git submodule: simple แต่ manage ยาก Monorepo: รวมทุก constructs ใน repo เดียว AWS CodeArtifact: managed package repository แนะนำ: internal package registry + semantic versioning

Q: CDK construct testing จำเป็นไหม?

A: จำเป็นมาก โดยเฉพาะ shared constructs Snapshot tests: จับ unintended changes Assertion tests: verify resources สร้างถูกต้อง ไม่ test = breaking change ไม่รู้ตัว → กระทบทุกทีมที่ใช้ construct