def deploy_images_remotely()

in cli/aws_orbit/remote_files/deploy.py [0:0]

• 56 lines of code
• 8 McCabe index (conditional complexity)

def deploy_images_remotely(env: str, requested_image: Optional[str] = None) -> None:
    _logger.debug(f"deploy_images_remotely args: {env} {requested_image}")
    image_dir = os.path.realpath(os.path.join(ORBIT_CLI_ROOT, "../../images"))
    context: "Context" = ContextSerDe.load_context_from_ssm(env_name=env, type=Context)

    _logger.debug(f"context loaded: {env}")
    codebuild_role = str(context.toolkit.admin_role)
    _logger.debug(f"The CODEBUILD_ROLE is {codebuild_role}")

    if requested_image:
        if os.path.isdir(f"{image_dir}/{requested_image}"):
            _logger.debug(f"Request build of single image {requested_image}")
            (image_name, image_addr, version) = _deploy_images_batch(
                path=f"{image_dir}/{requested_image}",
                image_name=requested_image,
                env=env,
                build_execution_role=codebuild_role,
            )
            _logger.debug(f"Returned from _deploy_images_batch: {image_name} {image_addr} {version}")
            im = str(image_name).replace("-", "_")
            new_image_manifest = ImageManifest(repository=str(image_addr), version=str(version))
            _logger.debug(f"New image manifest from single built image: {new_image_manifest}")
            context_latest: "Context" = ContextSerDe.load_context_from_ssm(env_name=env, type=Context)
            context_latest.images.__dict__[im] = new_image_manifest
            ContextSerDe.dump_context_to_ssm(context=context_latest)
        else:
            _logger.error("An image was requested to be built, but it doesn't exist in the images/ dir")

    else:
        new_images_manifest = {}
        # first build k8s-utilities
        (image_name, image_addr, version) = _deploy_images_batch(
            path=f"{image_dir}/k8s-utilities",
            image_name="k8s-utilities",
            env=env,
            build_execution_role=codebuild_role,
        )
        _logger.debug(f"Returned from _deploy_images_batch: {image_name} {image_addr} {version}")
        im = str(image_name).replace("-", "_")
        new_images_manifest[im] = ImageManifest(repository=str(image_addr), version=str(version))

        # now build the images dependent on k8s-utilities
        list_subfolders_with_paths = [f.path for f in os.scandir(image_dir) if f.is_dir()]
        max_workers = 4
        with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:

            def deploy_images_batch_helper(args_tuple: List[Any]) -> List[Tuple[str, str, str]]:
                return _deploy_images_batch(
                    path=args_tuple[0], image_name=args_tuple[1], env=args_tuple[2], build_execution_role=args_tuple[3]
                )

            args_tuples = [
                (path, path.split("/")[-1], env, codebuild_role)
                for path in list_subfolders_with_paths
                if "k8s-utilities" not in path
            ]

            results = list(executor.map(deploy_images_batch_helper, args_tuples))
            for res in results:
                im = str(res[0]).replace("-", "_")
                new_images_manifest[im] = ImageManifest(repository=str(res[1]), version=str(res[2]))
        _logger.debug(f"New image manifest from all images: {new_images_manifest}")
        context_latest_all: "Context" = ContextSerDe.load_context_from_ssm(env_name=env, type=Context)
        context_latest_all.images = ImagesManifest(**new_images_manifest)  # type: ignore
        ContextSerDe.dump_context_to_ssm(context=context_latest_all)