This is Chapter 3 of the Infrastructure as Code with AWS CDK series. Chapters 1 and 2 put the S3 bucket and Lambda directly in MyStack. Both get extracted into a reusable construct here.
L1, L2, L3
CDK organises constructs into three levels:
- L1 - raw CloudFormation resource wrappers, prefixed with
Cfn.CfnBucket,CfnFunction. No defaults added - every property must be set explicitly. - L2 - CDK’s built-in higher-level constructs.
s3.Bucket,_lambda.Function. These set sensible defaults, expose helper methods, and handle IAM viagrant_*methods. - L3 - composite constructs you write that bundle multiple resources into a single unit. The previous chapters used L2 constructs directly inside a stack. This chapter builds an L3.
The problem
The current MyStack creates the bucket and Lambda directly:
class MyStack(Stack):
def __init__(self, scope, id, config, **kwargs):
super().__init__(scope, id, **kwargs)
bucket = s3.Bucket(self, "FilesBucket",
bucket_name=config["bucket_name"]
)
fn = _lambda.Function(self, "ProcessorFunction",
runtime=_lambda.Runtime.PYTHON_3_12,
handler="handler.handler",
code=_lambda.Code.from_asset("lambda"),
timeout=Duration.seconds(config["lambda_timeout"]),
environment={"LOG_LEVEL": config["log_level"]}
)
bucket.grant_read(fn)
If you need a second processor - say one for raw uploads and one for processed files - all of this gets duplicated. That’s what a construct solves.
Writing the construct
Add file_processor.py inside the existing app module:
my-cdk-app/
├── app.py
├── my_cdk_app/
│ ├── __init__.py
│ ├── my_cdk_app_stack.py
│ └── file_processor.py ← new
# my_cdk_app/file_processor.py
from aws_cdk import (
Duration,
aws_s3 as s3,
aws_lambda as _lambda,
)
from constructs import Construct
class FileProcessorConstruct(Construct):
def __init__(
self,
scope: Construct,
id: str,
*,
bucket_name: str,
log_level: str,
lambda_timeout: int,
):
super().__init__(scope, id)
self.bucket = s3.Bucket(self, "Bucket",
bucket_name=bucket_name
)
self.fn = _lambda.Function(self, "Function",
runtime=_lambda.Runtime.PYTHON_3_12,
handler="handler.handler",
code=_lambda.Code.from_asset("lambda"),
timeout=Duration.seconds(lambda_timeout),
environment={"LOG_LEVEL": log_level}
)
self.bucket.grant_read(self.fn)
The * before the keyword arguments makes them keyword-only - callers must pass them by name, not by position. self.bucket and self.fn are public so callers can reach in and extend if needed.
Using it in a stack
# my_cdk_app/my_cdk_app_stack.py
from aws_cdk import Stack
from constructs import Construct
from my_cdk_app.file_processor import FileProcessorConstruct
class MyStack(Stack):
def __init__(self, scope: Construct, id: str, config: dict, **kwargs):
super().__init__(scope, id, **kwargs)
FileProcessorConstruct(self, "FileProcessor",
bucket_name=config["bucket_name"],
log_level=config["log_level"],
lambda_timeout=config["lambda_timeout"]
)
Reusing the construct
Two processors in the same stack, no duplication:
class MyStack(Stack):
def __init__(self, scope: Construct, id: str, config: dict, **kwargs):
super().__init__(scope, id, **kwargs)
FileProcessorConstruct(self, "RawProcessor",
bucket_name=f"{config['bucket_name']}-raw",
log_level=config["log_level"],
lambda_timeout=config["lambda_timeout"]
)
FileProcessorConstruct(self, "ProcessedProcessor",
bucket_name=f"{config['bucket_name']}-processed",
log_level=config["log_level"],
lambda_timeout=config["lambda_timeout"]
)
CDK uses the id argument (RawProcessor, ProcessedProcessor) to make all resource logical IDs unique within the stack. Both constructs deploy without naming conflicts.
Using exposed attributes
Use self.bucket and self.fn when something needs to be added at the call site:
processor = FileProcessorConstruct(self, "FileProcessor",
bucket_name=config["bucket_name"],
log_level=config["log_level"],
lambda_timeout=config["lambda_timeout"]
)
# Grant a second function read access to the same bucket
processor.bucket.grant_read(reporting_fn)
If it’s specific to one stack, keep it there rather than baking it into the construct.
Notes
- Do not name the constructs directory
constructs/- it conflicts with the CDKconstructspackage import. Putting files inside the app module (my_cdk_app/) sidesteps this entirely. - The
*in the__init__signature enforces keyword arguments. Without it, callers can pass values positionally, which gets confusing with three or more parameters. self.bucketandself.fnbeing public is a choice. If the construct is fully self-contained and callers should never reach in, keep them private with a leading underscore (self._bucket).- To use a construct across multiple CDK projects, move it into a standalone Python package and install it as a dependency. Cross-language support (TypeScript, Java) requires
jsii- a separate topic.