Python Terraform CDK: Cloud Infrastructure in Python

CDKTF (Cloud Development Kit for Terraform) lets you define Terraform infrastructure using Python, TypeScript, or other languages instead of HCL. The Python code synthesizes to Terraform JSON, then standard terraform apply deploys it. This means you get full Python expressiveness — loops, functions, classes, type hints, unit tests — while Terraform handles the actual provisioning, state management, and provider ecosystem. This guide builds a production AWS stack (VPC, ECS Fargate, RDS, ALB) using CDKTF Python.

Installation and Project Setup

# Prerequisites: Node.js 18+, Terraform CLI, Python 3.10+
pip install cdktf cdktf-cdktf-provider-aws

# Initialize project
cdktf init --template=python --local

# Install AWS provider
pip install cdktf-cdktf-provider-aws

# Project structure:
# infra/
#   main.py           ← entry point
#   stacks/
#     network.py
#     compute.py
#     database.py
#   constructs/
#     fargate_service.py
#   cdktf.json

// cdktf.json
{
  "language": "python",
  "app": "python3 main.py",
  "terraformProviders": ["aws@~> 5.0"],
  "terraformModules": [],
  "sendCrashReports": false
}

Stacks and Constructs

A Stack is a deployable unit corresponding to one Terraform state file. A Construct is a reusable building block — analogous to a Python class that encapsulates related infrastructure. Constructs compose into stacks; stacks compose into apps.

# main.py
from cdktf import App, TerraformStack, TerraformOutput
from cdktf_cdktf_provider_aws.provider import AwsProvider
from constructs import Construct
from stacks.network import NetworkStack
from stacks.compute import ComputeStack
from stacks.database import DatabaseStack


class TechoralApp(App):
    def __init__(self):
        super().__init__()

        env = "production"
        region = "ap-south-1"

        # Network stack (VPC, subnets, security groups)
        network = NetworkStack(self, f"techoral-network-{env}",
                               env=env, region=region)

        # Database stack depends on network
        database = DatabaseStack(self, f"techoral-db-{env}",
                                 env=env, region=region,
                                 vpc_id=network.vpc_id,
                                 subnet_ids=network.private_subnet_ids,
                                 security_group_id=network.db_sg_id)

        # Compute stack depends on network and database
        ComputeStack(self, f"techoral-compute-{env}",
                     env=env, region=region,
                     vpc_id=network.vpc_id,
                     public_subnet_ids=network.public_subnet_ids,
                     private_subnet_ids=network.private_subnet_ids,
                     alb_sg_id=network.alb_sg_id,
                     ecs_sg_id=network.ecs_sg_id,
                     database_url=database.database_url)


if __name__ == "__main__":
    TechoralApp().synth()

VPC and Networking

# stacks/network.py
from cdktf import TerraformStack, TerraformOutput
from cdktf_cdktf_provider_aws.provider import AwsProvider
from cdktf_cdktf_provider_aws.vpc import Vpc
from cdktf_cdktf_provider_aws.subnet import Subnet
from cdktf_cdktf_provider_aws.internet_gateway import InternetGateway
from cdktf_cdktf_provider_aws.nat_gateway import NatGateway
from cdktf_cdktf_provider_aws.eip import Eip
from cdktf_cdktf_provider_aws.route_table import RouteTable
from cdktf_cdktf_provider_aws.route_table_association import RouteTableAssociation
from cdktf_cdktf_provider_aws.route import Route
from cdktf_cdktf_provider_aws.security_group import SecurityGroup
from constructs import Construct


class NetworkStack(TerraformStack):
    def __init__(self, scope: Construct, id: str, env: str, region: str):
        super().__init__(scope, id)

        AwsProvider(self, "aws", region=region)

        # VPC
        self._vpc = Vpc(self, "vpc",
                        cidr_block="10.0.0.0/16",
                        enable_dns_hostnames=True,
                        enable_dns_support=True,
                        tags={"Name": f"techoral-{env}", "Environment": env})

        # Public subnets (ALB, NAT gateways)
        azs = ["ap-south-1a", "ap-south-1b"]
        self._public_subnets = [
            Subnet(self, f"public-{i}",
                   vpc_id=self._vpc.id,
                   cidr_block=f"10.0.{i}.0/24",
                   availability_zone=azs[i],
                   map_public_ip_on_launch=True,
                   tags={"Name": f"techoral-public-{i}-{env}"})
            for i in range(len(azs))
        ]

        # Private subnets (ECS, RDS)
        self._private_subnets = [
            Subnet(self, f"private-{i}",
                   vpc_id=self._vpc.id,
                   cidr_block=f"10.0.{i+10}.0/24",
                   availability_zone=azs[i],
                   tags={"Name": f"techoral-private-{i}-{env}"})
            for i in range(len(azs))
        ]

        # Internet Gateway
        igw = InternetGateway(self, "igw", vpc_id=self._vpc.id)

        # NAT Gateway (one per AZ for HA)
        eip = Eip(self, "nat-eip", domain="vpc")
        nat = NatGateway(self, "nat",
                         subnet_id=self._public_subnets[0].id,
                         allocation_id=eip.id)

        # Route tables
        public_rt = RouteTable(self, "public-rt", vpc_id=self._vpc.id)
        Route(self, "public-route", route_table_id=public_rt.id,
              destination_cidr_block="0.0.0.0/0", gateway_id=igw.id)
        for i, subnet in enumerate(self._public_subnets):
            RouteTableAssociation(self, f"public-rta-{i}",
                                  route_table_id=public_rt.id, subnet_id=subnet.id)

        private_rt = RouteTable(self, "private-rt", vpc_id=self._vpc.id)
        Route(self, "private-route", route_table_id=private_rt.id,
              destination_cidr_block="0.0.0.0/0", nat_gateway_id=nat.id)
        for i, subnet in enumerate(self._private_subnets):
            RouteTableAssociation(self, f"private-rta-{i}",
                                  route_table_id=private_rt.id, subnet_id=subnet.id)

        # Security groups
        self._alb_sg = SecurityGroup(self, "alb-sg",
                                      vpc_id=self._vpc.id,
                                      name=f"alb-sg-{env}",
                                      ingress=[
                                          {"from_port": 80, "to_port": 80, "protocol": "tcp", "cidr_blocks": ["0.0.0.0/0"]},
                                          {"from_port": 443, "to_port": 443, "protocol": "tcp", "cidr_blocks": ["0.0.0.0/0"]},
                                      ],
                                      egress=[{"from_port": 0, "to_port": 0, "protocol": "-1", "cidr_blocks": ["0.0.0.0/0"]}])

        self._ecs_sg = SecurityGroup(self, "ecs-sg",
                                      vpc_id=self._vpc.id,
                                      name=f"ecs-sg-{env}",
                                      ingress=[{"from_port": 8000, "to_port": 8000, "protocol": "tcp", "security_groups": [self._alb_sg.id]}],
                                      egress=[{"from_port": 0, "to_port": 0, "protocol": "-1", "cidr_blocks": ["0.0.0.0/0"]}])

        self._db_sg = SecurityGroup(self, "db-sg",
                                     vpc_id=self._vpc.id,
                                     name=f"db-sg-{env}",
                                     ingress=[{"from_port": 5432, "to_port": 5432, "protocol": "tcp", "security_groups": [self._ecs_sg.id]}],
                                     egress=[{"from_port": 0, "to_port": 0, "protocol": "-1", "cidr_blocks": ["0.0.0.0/0"]}])

        # Expose outputs for other stacks
        TerraformOutput(self, "vpc-id", value=self._vpc.id)

    @property
    def vpc_id(self) -> str:
        return self._vpc.id

    @property
    def public_subnet_ids(self) -> list[str]:
        return [s.id for s in self._public_subnets]

    @property
    def private_subnet_ids(self) -> list[str]:
        return [s.id for s in self._private_subnets]

    @property
    def alb_sg_id(self) -> str:
        return self._alb_sg.id

    @property
    def ecs_sg_id(self) -> str:
        return self._ecs_sg.id

    @property
    def db_sg_id(self) -> str:
        return self._db_sg.id

ECS Fargate Service

# stacks/compute.py (abbreviated)
from cdktf import TerraformStack
from cdktf_cdktf_provider_aws.ecs_cluster import EcsCluster
from cdktf_cdktf_provider_aws.ecs_task_definition import EcsTaskDefinition
from cdktf_cdktf_provider_aws.ecs_service import EcsService
from cdktf_cdktf_provider_aws.lb import Lb
from cdktf_cdktf_provider_aws.lb_target_group import LbTargetGroup
from cdktf_cdktf_provider_aws.lb_listener import LbListener
import json


class ComputeStack(TerraformStack):
    def __init__(self, scope, id, *, env, region, vpc_id, public_subnet_ids,
                 private_subnet_ids, alb_sg_id, ecs_sg_id, database_url):
        super().__init__(scope, id)

        from cdktf_cdktf_provider_aws.provider import AwsProvider
        AwsProvider(self, "aws", region=region)

        cluster = EcsCluster(self, "cluster", name=f"techoral-{env}")

        # ALB
        alb = Lb(self, "alb",
                 name=f"techoral-alb-{env}",
                 internal=False,
                 load_balancer_type="application",
                 security_groups=[alb_sg_id],
                 subnets=public_subnet_ids)

        tg = LbTargetGroup(self, "tg",
                           name=f"techoral-tg-{env}",
                           port=8000,
                           protocol="HTTP",
                           vpc_id=vpc_id,
                           target_type="ip",
                           health_check={"path": "/health", "healthy_threshold": 2})

        LbListener(self, "listener",
                   load_balancer_arn=alb.arn,
                   port=80, protocol="HTTP",
                   default_action=[{"type": "forward", "target_group_arn": tg.arn}])

        # Task definition
        task_def = EcsTaskDefinition(self, "task",
            family=f"techoral-api-{env}",
            requires_compatibilities=["FARGATE"],
            network_mode="awsvpc",
            cpu="256", memory="512",
            container_definitions=json.dumps([{
                "name": "api",
                "image": f"registry.techoral.com/api:latest",
                "portMappings": [{"containerPort": 8000}],
                "environment": [
                    {"name": "DATABASE_URL", "value": database_url},
                    {"name": "ENVIRONMENT", "value": env},
                ],
                "logConfiguration": {
                    "logDriver": "awslogs",
                    "options": {
                        "awslogs-group": f"/ecs/techoral-api-{env}",
                        "awslogs-region": region,
                        "awslogs-stream-prefix": "ecs",
                    }
                }
            }])
        )

        EcsService(self, "service",
                   name=f"techoral-api-{env}",
                   cluster=cluster.id,
                   task_definition=task_def.arn,
                   desired_count=2,
                   launch_type="FARGATE",
                   network_configuration={
                       "subnets": private_subnet_ids,
                       "security_groups": [ecs_sg_id],
                       "assign_public_ip": False,
                   },
                   load_balancer=[{
                       "target_group_arn": tg.arn,
                       "container_name": "api",
                       "container_port": 8000,
                   }])

RDS PostgreSQL

# stacks/database.py
from cdktf import TerraformStack, TerraformOutput
from cdktf_cdktf_provider_aws.db_instance import DbInstance
from cdktf_cdktf_provider_aws.db_subnet_group import DbSubnetGroup
from cdktf_cdktf_provider_aws.ssm_parameter import SsmParameter
import secrets


class DatabaseStack(TerraformStack):
    def __init__(self, scope, id, *, env, region, vpc_id, subnet_ids, security_group_id):
        super().__init__(scope, id)

        from cdktf_cdktf_provider_aws.provider import AwsProvider
        AwsProvider(self, "aws", region=region)

        db_password = secrets.token_urlsafe(32)

        subnet_group = DbSubnetGroup(self, "subnet-group",
                                     name=f"techoral-db-{env}",
                                     subnet_ids=subnet_ids)

        rds = DbInstance(self, "rds",
                         identifier=f"techoral-{env}",
                         engine="postgres",
                         engine_version="16.2",
                         instance_class="db.t4g.small",
                         allocated_storage=50,
                         max_allocated_storage=200,
                         storage_type="gp3",
                         storage_encrypted=True,
                         db_name="techoral",
                         username="techoral_admin",
                         password=db_password,
                         db_subnet_group_name=subnet_group.name,
                         vpc_security_group_ids=[security_group_id],
                         multi_az=env == "production",
                         backup_retention_period=7,
                         skip_final_snapshot=env != "production",
                         deletion_protection=env == "production",
                         tags={"Environment": env})

        # Store password in SSM (ECS task retrieves it at startup)
        SsmParameter(self, "db-password",
                     name=f"/techoral/{env}/db-password",
                     type="SecureString",
                     value=db_password)

        self._database_url = f"postgresql://techoral_admin:{db_password}@{rds.endpoint}/techoral"
        TerraformOutput(self, "db-endpoint", value=rds.endpoint)

    @property
    def database_url(self) -> str:
        return self._database_url

Reusable Constructs

# constructs/fargate_service.py — reusable ECS Fargate construct
from constructs import Construct
from cdktf_cdktf_provider_aws.ecs_task_definition import EcsTaskDefinition
from cdktf_cdktf_provider_aws.ecs_service import EcsService
import json


class FargateService(Construct):
    """Reusable construct for an ECS Fargate service with ALB target group."""

    def __init__(self, scope: Construct, id: str, *,
                 cluster_id: str, service_name: str, image: str,
                 cpu: str = "256", memory: str = "512",
                 desired_count: int = 2, port: int = 8000,
                 env_vars: dict | None = None,
                 subnet_ids: list[str],
                 security_group_ids: list[str],
                 target_group_arn: str | None = None):
        super().__init__(scope, id)

        container_defs = [{
            "name": service_name,
            "image": image,
            "portMappings": [{"containerPort": port}],
            "environment": [{"name": k, "value": v} for k, v in (env_vars or {}).items()],
        }]

        task = EcsTaskDefinition(self, "task",
                                  family=service_name,
                                  requires_compatibilities=["FARGATE"],
                                  network_mode="awsvpc",
                                  cpu=cpu, memory=memory,
                                  container_definitions=json.dumps(container_defs))

        lb_config = []
        if target_group_arn:
            lb_config = [{"target_group_arn": target_group_arn,
                          "container_name": service_name, "container_port": port}]

        self.service = EcsService(self, "service",
                                   name=service_name,
                                   cluster=cluster_id,
                                   task_definition=task.arn,
                                   desired_count=desired_count,
                                   launch_type="FARGATE",
                                   network_configuration={
                                       "subnets": subnet_ids,
                                       "security_groups": security_group_ids,
                                       "assign_public_ip": False,
                                   },
                                   load_balancer=lb_config)


# Usage — reuse the construct for any service
# api = FargateService(self, "api-service", cluster_id=cluster.id, service_name="api", image="...", ...)
# worker = FargateService(self, "worker", cluster_id=cluster.id, service_name="worker", image="...", ...)

Testing Infrastructure Code

# tests/test_network_stack.py
import pytest
from cdktf import App, Testing
from stacks.network import NetworkStack


class TestNetworkStack:
    def setup_method(self):
        self.app = App()
        self.stack = NetworkStack(self.app, "test-network", env="test", region="ap-south-1")

    def test_vpc_created(self):
        synth = Testing.synth(self.stack)
        assert "aws_vpc" in synth

    def test_two_public_subnets(self):
        synth = Testing.synth(self.stack)
        subnets = synth.get("aws_subnet", {})
        public_subnets = [k for k in subnets if "public" in k]
        assert len(public_subnets) == 2

    def test_nat_gateway_exists(self):
        synth = Testing.synth(self.stack)
        assert "aws_nat_gateway" in synth

    def test_security_groups_count(self):
        synth = Testing.synth(self.stack)
        sgs = synth.get("aws_security_group", {})
        # alb-sg, ecs-sg, db-sg
        assert len(sgs) >= 3

# Synthesize to Terraform JSON
cdktf synth

# Plan (shows what Terraform will create)
cdktf diff

# Deploy (runs terraform apply)
cdktf deploy techoral-network-production
cdktf deploy techoral-db-production
cdktf deploy techoral-compute-production

# Destroy
cdktf destroy techoral-compute-production

# Run tests
pytest tests/

Frequently Asked Questions

CDKTF vs AWS CDK vs Pulumi — which to use?

Use CDKTF if you're already invested in Terraform (state files, providers, modules) and want to write Python instead of HCL. Use AWS CDK for AWS-only infrastructure — better AWS integrations, CloudFormation under the hood. Use Pulumi for a true Python runtime (dynamic values, secrets management) without the HCL intermediary. CDKTF is the best choice when migrating an existing Terraform codebase to code.

Does CDKTF support Terraform modules?

Yes — add modules to cdktf.json under terraformModules and CDKTF generates Python classes for them. You can use any public Terraform module (VPC, EKS, RDS) as a Python class with type hints and autocompletion.

How do I handle secrets in CDKTF?

Never hard-code secrets in Python code — they end up in Terraform state files. Use Terraform variables with TerraformVariable(sensitive=True) passed via environment variables, or reference existing secrets with data sources (AwsSecretsmanagerSecretVersion). Use SSM Parameter Store or AWS Secrets Manager to store and retrieve secrets at runtime.