Assets

Lambda function and Docker image asset constructs.

assets

Modules

code_asset

Classes

CodeAsset dataclass

CodeAsset(
    asset_name: str,
    asset_props: AssetProps,
    default_runtime: Runtime,
    supported_runtimes: Sequence[Runtime] | None = None,
    environment: Mapping[str, str] | None = None,
)

Functions

create_py_code_asset classmethod

create_py_code_asset(
    path: Path,
    context_path: Path | None,
    requirements_file: Path | None = None,
    includes: Sequence[str] | None = None,
    excludes: Sequence[str] | None = None,
    runtime: Runtime = PYTHON_3_11,
    platform: str | None = "linux/amd64",
    environment: Mapping[str, str] | None = None,
    use_uv: bool = False,
) -> CodeAsset

Create and bundle a Python Lambda code asset

This factory method prepares a Lambda code asset whose source may live in a subdirectory of a larger repository. It computes a stable hash (excluding common Python / build artifacts) to maximize CDK asset caching, then leverages CDK's Docker bundling to install dependencies into /asset-output so they are packaged alongside the function code.

Dependency installation strategies

• Standard pip (default): Installs either the provided requirements file or the local package (.) into /asset-output.
• uv (if use_uv=True): Uses uv for faster, deterministic resolution. If a requirements_file is supplied it is installed directly; otherwise uv export produces a frozen requirements file which is then installed.

Private Git dependencies

The host's ~/.ssh directory is mounted read‑only into the bundling container so that private repositories referenced in dependency specifications (e.g., git+ssh URLs) can be resolved. Diagnostic SSH output (ssh -vT git@github.com) is executed to aid debugging but does not fail the build.

File permissions

Directory permissions are normalized to 755 and file permissions to 644 to ensure the Lambda runtime can read all packaged artifacts.

Parameters:

Name	Type	Description	Default
path	Path	Path to the Python project or module root you want to package (the leaf containing setup.cfg, pyproject.toml, requirements, or the code itself).	required
context_path	Optional[Path]	Root directory used as the asset's source context (often the monorepo root). If None, defaults to path. Everything under this directory (minus excluded patterns) is made available during bundling so that relative/local path dependencies can resolve correctly.	required
requirements_file	Optional[Path]	Path to a requirements file (e.g., requirements.txt). If provided: * pip mode: installs with pip -r * uv mode: installs with uv pip install -r If omitted: * pip mode: installs the local project (.) * uv mode: uv export produces a frozen requirements-autogen.txt, then that is installed.	None
includes	Optional[Sequence[str]]	Additional glob patterns to explicitly include when generating the asset hash. Useful for forcing hash sensitivity to files that might otherwise be excluded.	None
excludes	Optional[Sequence[str]]	Additional glob patterns to exclude (beyond built-in Python / tooling ignores) for hashing and packaging. Helps reduce asset size and improve cache hits.	None
runtime	Runtime	Lambda runtime whose bundling image will be used (default: PYTHON_3_11).	PYTHON_3_11
platform	Optional[str]	Docker platform string (e.g., linux/amd64) to enforce architecture consistency when building on heterogeneous hosts (e.g., Apple Silicon).	'linux/amd64'
environment	Optional[Mapping[str, str]]	Default environment variables to associate with the resulting CodeAsset (may be applied when the asset is used in Lambda functions).	None
use_uv	bool	If True, use uv for dependency resolution and installation; otherwise fall back to pip.	False

Returns:

Type	Description
CodeAsset	A new instance encapsulating the bundled Python code and dependencies.

Source code in src/aibs_informatics_cdk_lib/constructs_/assets/code_asset.py

@classmethod
def create_py_code_asset(
    cls,
    path: Path,
    context_path: Path | None,
    requirements_file: Path | None = None,
    includes: Sequence[str] | None = None,
    excludes: Sequence[str] | None = None,
    runtime: lambda_.Runtime = lambda_.Runtime.PYTHON_3_11,
    platform: str | None = "linux/amd64",
    environment: Mapping[str, str] | None = None,
    use_uv: bool = False,
) -> "CodeAsset":
    """Create and bundle a Python Lambda code asset

    This factory method prepares a Lambda code asset whose source may live in a subdirectory of a
    larger repository. It computes a stable hash (excluding common Python / build artifacts) to
    maximize CDK asset caching, then leverages CDK's Docker bundling to install dependencies into
    /asset-output so they are packaged alongside the function code.

    Dependency installation strategies:
        * Standard pip (default): Installs either the provided requirements file or the local package (.)
            into /asset-output.
        * uv (if use_uv=True): Uses uv for faster, deterministic resolution. If a requirements_file is
            supplied it is installed directly; otherwise uv export produces a frozen requirements file
            which is then installed.

    Private Git dependencies:
        The host's ~/.ssh directory is mounted read‑only into the bundling container so that private
        repositories referenced in dependency specifications (e.g., git+ssh URLs) can be resolved.
        Diagnostic SSH output (ssh -vT git@github.com) is executed to aid debugging but does not fail
        the build.

    File permissions:
        Directory permissions are normalized to 755 and file permissions to 644 to ensure the Lambda
        runtime can read all packaged artifacts.

    Parameters:
        path (Path):
            Path to the Python project or module root you want to package (the leaf containing setup.cfg,
            pyproject.toml, requirements, or the code itself).
        context_path (Optional[Path]):
            Root directory used as the asset's source context (often the monorepo root). If None, defaults
            to path. Everything under this directory (minus excluded patterns) is made available during
            bundling so that relative/local path dependencies can resolve correctly.
        requirements_file (Optional[Path]):
            Path to a requirements file (e.g., requirements.txt). If provided:
                * pip mode: installs with pip -r <file>
                * uv mode: installs with uv pip install -r <file>
            If omitted:
                * pip mode: installs the local project (.)
                * uv mode: uv export produces a frozen requirements-autogen.txt, then that is installed.
        includes (Optional[Sequence[str]]):
            Additional glob patterns to explicitly include when generating the asset hash. Useful for
            forcing hash sensitivity to files that might otherwise be excluded.
        excludes (Optional[Sequence[str]]):
            Additional glob patterns to exclude (beyond built-in Python / tooling ignores) for hashing
            and packaging. Helps reduce asset size and improve cache hits.
        runtime (aws_cdk.aws_lambda.Runtime):
            Lambda runtime whose bundling image will be used (default: PYTHON_3_11).
        platform (Optional[str]):
            Docker platform string (e.g., linux/amd64) to enforce architecture consistency when building
            on heterogeneous hosts (e.g., Apple Silicon).
        environment (Optional[Mapping[str, str]]):
            Default environment variables to associate with the resulting CodeAsset (may be applied when
            the asset is used in Lambda functions).
        use_uv (bool):
            If True, use uv for dependency resolution and installation; otherwise fall back to pip.

    Returns:
        A new instance encapsulating the bundled Python code and dependencies.
    """  # noqa: E501

    if context_path is None:
        context_path = path

    # This is relative to the repo root copied in asset props below
    package_path = path.relative_to(context_path)
    full_path = context_path / package_path

    asset_hash = generate_path_hash(
        path=str(context_path.resolve()),
        includes=list(includes) if includes else None,
        excludes=[*(excludes or []), *PYTHON_REGEX_EXCLUDES],
    )
    host_ssh_dir = str(Path.home() / ".ssh")

    ssh_setup_commands = [
        # Copy in host ssh keys that are needed to clone private git repos
        # first we make the .ssh directory if it doesn't exist
        "mkdir -p /root/.ssh",
        # then we copy in all files from host ssh dir to container ssh dir
        f"cp -a {host_ssh_dir}/. /root/.ssh/",
        # Useful debug if anything goes wrong with github SSH related things
        "ssh -vT git@github.com || true",
    ]

    uv_setup_commands = [
        # Install uv for this ephemeral bundling container and expose it on PATH
        "curl -LsSf https://astral.sh/uv/install.sh | sh",
        "export PATH=/root/.local/bin:$PATH",
        "uv --version",
    ]
    package_install_commands = []

    if use_uv:
        if requirements_file is not None:
            logging.warning(
                f"Using uv with requirements file {requirements_file}, "
                f"make sure that all dependencies are compatible with uv!"
            )
            package_install_commands.append(
                f"uv pip install -r {requirements_file.as_posix()} --no-cache --target /asset-output",  # noqa: E501
            )
        else:
            package_install_commands += [
                "uv export --frozen --no-dev --no-editable -o requirements-autogen.txt",
                "uv pip install --no-sources -r requirements-autogen.txt --no-cache --target /asset-output",  # noqa: E501
            ]
    else:
        package_install_commands += [
            "python3 -m pip install --upgrade pip --no-cache",
            f"python3 -m pip install {'-r ' + requirements_file.as_posix() if requirements_file else '.'} --no-cache --target /asset-output",  # noqa: E501
        ]

    asset_props = aws_s3_assets.AssetProps(
        # CDK bundles lambda assets in a docker container. This causes issues for our local
        # path dependencies. In order to resolve the relative local path dependency,
        # we need to specify the path to the root of the repo.
        path=str(context_path.resolve()),
        asset_hash=asset_hash,
        # It is important to exclude files from the git repo, because
        #   1. it effectively makes our caching for assets moot
        #   2. we also don't want to include certain files for size reasons.
        exclude=[
            *PYTHON_GLOB_EXCLUDES,
            "**/cdk.out/",
        ],
        # ignore_mode=cdk.IgnoreMode.GIT,
        bundling=cdk.BundlingOptions(
            image=runtime.bundling_image,
            working_directory=f"/asset-input/{package_path}",
            entrypoint=["/bin/bash", "-c"],
            command=[
                # This makes the following commands run together as one
                # WARNING Make sure not to modify {host_ssh_dir} in any way,
                # in this set of commands!
                " && ".join(
                    [
                        "set -x",
                        *ssh_setup_commands,
                        *(uv_setup_commands if use_uv else []),
                        *package_install_commands,
                        # TODO: remove botocore and boto3 from asset output
                        # Must make asset output permissions accessible to lambda
                        "find /asset-output -type d -print0 | xargs -0 chmod 755",
                        "find /asset-output -type f -print0 | xargs -0 chmod 644",
                    ]
                ),
            ],
            user="root:root",
            volumes=[
                cdk.DockerVolume(
                    host_path=host_ssh_dir,
                    container_path=host_ssh_dir,
                ),
            ],
            platform=platform,
        ),
    )

    return CodeAsset(
        asset_name=os.path.basename(full_path.resolve()),
        asset_props=asset_props,
        default_runtime=runtime,
        environment=environment,
    )

code_asset_definitions

Classes

AssetsMixin

Functions

resolve_repo_path classmethod

resolve_repo_path(
    repo_url: str, repo_path_env_var: str | None
) -> Path

Resolves the repo path from the environment or clones the repo from the url

This method is useful to quickly swapping between locally modified changes and the remote repo.

This should typically be used in the context of defining a code asset for a static name (e.g. AIBS_INFORMATICS_AWS_LAMBDA). You can then use the env var option to point to a local repo path for development.

Parameters:

Name	Type	Description	Default
repo_url	str	The git repo url. This is required. If the repo path is not in the environment, the repo will be cloned from this url.	required
repo_path_env_var	Optional[str]	The environment variable that contains the repo path or alternative repo url. This is optional. This is useful for local development.	required

Returns:

Type	Description
Path	The path to the repo

Source code in src/aibs_informatics_cdk_lib/constructs_/assets/code_asset_definitions.py

@classmethod
def resolve_repo_path(cls, repo_url: str, repo_path_env_var: str | None) -> Path:
    """Resolves the repo path from the environment or clones the repo from the url

    This method is useful to quickly swapping between locally modified changes and the remote
    repo.

    This should typically be used in the context of defining a code asset for a static name
    (e.g. AIBS_INFORMATICS_AWS_LAMBDA). You can then use the env var option to point to a local
    repo path for development.

    Args:
        repo_url (str): The git repo url. This is required.
            If the repo path is not in the environment, the repo will be cloned from this url.
        repo_path_env_var (Optional[str]): The environment variable that contains the
            repo path or alternative repo url. This is optional.
            This is useful for local development.

    Returns:
        The path to the repo
    """
    if repo_path_env_var and (repo_path := os.getenv(repo_path_env_var)) is not None:
        logger.info(f"Using {repo_path_env_var} from environment")
        if is_local_repo(repo_path):
            return Path(repo_path)
        elif is_repo_url(str(repo_path)):
            return clone_repo(repo_path, skip_if_exists=True)
        else:
            raise ValueError(f"Env variable {repo_path_env_var} is not a valid git repo")
    else:
        return clone_repo(repo_url, skip_if_exists=True)

AIBSInformaticsCodeAssets

AIBSInformaticsCodeAssets(
    scope: Construct,
    construct_id: str,
    env_base: EnvBase,
    runtime: Runtime | None = None,
    aibs_informatics_aws_lambda_repo: str | None = None,
)

Bases: Construct, AssetsMixin

Source code in src/aibs_informatics_cdk_lib/constructs_/assets/code_asset_definitions.py

def __init__(
    self,
    scope: constructs.Construct,
    construct_id: str,
    env_base: EnvBase,
    runtime: lambda_.Runtime | None = None,
    aibs_informatics_aws_lambda_repo: str | None = None,
) -> None:
    super().__init__(scope, construct_id)
    self.env_base = env_base
    self.runtime = runtime or lambda_.Runtime.PYTHON_3_11
    self.AIBS_INFORMATICS_AWS_LAMBDA_REPO = (
        aibs_informatics_aws_lambda_repo or AIBS_INFORMATICS_AWS_LAMBDA_REPO
    )

Functions

AIBS_INFORMATICS_AWS_LAMBDA

AIBS_INFORMATICS_AWS_LAMBDA() -> CodeAsset

Returns a NEW code asset for aibs-informatics-aws-lambda

Returns:

Type	Description
CodeAsset	The code asset

Source code in src/aibs_informatics_cdk_lib/constructs_/assets/code_asset_definitions.py

@cached_property
def AIBS_INFORMATICS_AWS_LAMBDA(self) -> CodeAsset:
    """Returns a NEW code asset for aibs-informatics-aws-lambda

    Returns:
        The code asset
    """

    repo_path = self.resolve_repo_path(
        self.AIBS_INFORMATICS_AWS_LAMBDA_REPO, AIBS_INFORMATICS_AWS_LAMBDA_REPO_ENV_VAR
    )

    asset_hash = generate_path_hash(
        path=str(repo_path.resolve()),
        excludes=PYTHON_REGEX_EXCLUDES,
    )
    logger.info(f"aibs-informatics-aws-lambda asset hash={asset_hash}")
    bundling_image = self.runtime.bundling_image
    host_ssh_dir = str(Path.home() / ".ssh")
    asset_props = aws_s3_assets.AssetProps(
        # CDK bundles lambda assets in a docker container. This causes issues for our local
        # path dependencies. In order to resolve the relative local path dependency,
        # we need to specify the path to the root of the repo.
        path=str(repo_path),
        asset_hash=asset_hash,
        # It is important to exclude files from the git repo, because
        #   1. it effectively makes our caching for assets moot
        #   2. we also don't want to include certain files for size reasons.
        exclude=[
            *PYTHON_GLOB_EXCLUDES,
            "**/cdk.out/",
            "**/scripts/**",
        ],
        bundling=cdk.BundlingOptions(
            image=bundling_image,
            working_directory="/asset-input",
            entrypoint=["/bin/bash", "-c"],
            command=[
                # This makes the following commands run together as one
                # WARNING Make sure not to modify {host_ssh_dir} in any way, in this set of commands!  # noqa: E501
                " && ".join(
                    [
                        "set -x",
                        # Copy in host ssh keys that are needed to clone private git repos
                        f"cp -r {host_ssh_dir} /root/.ssh",
                        # Useful debug if anything goes wrong with github SSH related things
                        "ssh -vT git@github.com || true",
                        # Must make sure that the package is not installing using --editable mode  # noqa: E501
                        "python3 -m pip install --upgrade pip --no-cache",
                        "pip3 install . --no-cache -t /asset-output",
                        # TODO: remove botocore and boto3 from asset output
                        # Must make asset output permissions accessible to lambda
                        "find /asset-output -type d -print0 | xargs -0 chmod 755",
                        "find /asset-output -type f -print0 | xargs -0 chmod 644",
                    ]
                ),
            ],
            user="root:root",
            volumes=[
                cdk.DockerVolume(
                    host_path=host_ssh_dir,
                    container_path=host_ssh_dir,
                ),
            ],
        ),
    )
    return CodeAsset(
        asset_name=os.path.basename(repo_path.resolve()),
        asset_props=asset_props,
        default_runtime=self.runtime,
        environment={
            self.env_base.ENV_BASE_KEY: self.env_base,
        },
    )

AIBSInformaticsDockerAssets

AIBSInformaticsDockerAssets(
    scope: Construct,
    construct_id: str,
    env_base: EnvBase,
    aibs_informatics_aws_lambda_repo: str | None = None,
)

Bases: Construct, AssetsMixin

Source code in src/aibs_informatics_cdk_lib/constructs_/assets/code_asset_definitions.py

def __init__(
    self,
    scope: constructs.Construct,
    construct_id: str,
    env_base: EnvBase,
    aibs_informatics_aws_lambda_repo: str | None = None,
) -> None:
    super().__init__(scope, construct_id)
    self.env_base = env_base
    self.AIBS_INFORMATICS_AWS_LAMBDA_REPO = (
        aibs_informatics_aws_lambda_repo or AIBS_INFORMATICS_AWS_LAMBDA_REPO
    )

Functions

AIBS_INFORMATICS_AWS_LAMBDA

AIBS_INFORMATICS_AWS_LAMBDA() -> DockerImageAsset

Returns a NEW docker asset for aibs-informatics-aws-lambda

Returns:

Type	Description
DockerImageAsset	aws_ecr_assets.DockerImageAsset: The docker asset

Source code in src/aibs_informatics_cdk_lib/constructs_/assets/code_asset_definitions.py

@cached_property
def AIBS_INFORMATICS_AWS_LAMBDA(self) -> aws_ecr_assets.DockerImageAsset:
    """Returns a NEW docker asset for aibs-informatics-aws-lambda

    Returns:
        aws_ecr_assets.DockerImageAsset: The docker asset
    """
    repo_path = self.resolve_repo_path(
        self.AIBS_INFORMATICS_AWS_LAMBDA_REPO, AIBS_INFORMATICS_AWS_LAMBDA_REPO_ENV_VAR
    )

    return aws_ecr_assets.DockerImageAsset(
        self,
        "aibs-informatics-aws-lambda",
        directory=repo_path.as_posix(),
        build_ssh="default",
        platform=aws_ecr_assets.Platform.LINUX_AMD64,
        asset_name="aibs-informatics-aws-lambda",
        file="docker/Dockerfile",
        extra_hash=generate_path_hash(
            path=str(repo_path.resolve()),
            excludes=PYTHON_REGEX_EXCLUDES,
        ),
        exclude=[
            *PYTHON_GLOB_EXCLUDES,
            *GLOBAL_GLOB_EXCLUDES,
        ],
    )

Functions