def deploy_modules()

in aws_codeseeder/commands/_module_commands.py [0:0]

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

def deploy_modules(seedkit_name: str, python_modules: List[str]) -> None:
    """Deploy local Python modules to the CodeArtifact Domain/Repository associated with a Seedkit

    This is a utility function that attempts to package and deploy local Python projects to CodeArtifact for use in
    CodeBuild executions.

    Parameters
    ----------
    seedkit_name : str
        Name of a previously deployed Seedkit
    python_modules : List[str]
        List of local Python modules/projects to deploy. Each module is of the form
        "[package-name]:[directory]" where [package-name] is the name of the Python package and [directory] is the
        local location of the module/project

    Raises
    ------
    ValueError
        If module names are of the wrong form
    """
    stack_name: str = cfn.get_stack_name(seedkit_name=seedkit_name)
    LOGGER.info("Deploying Modules for Seedkit %s with Stack Name %s", seedkit_name, stack_name)
    LOGGER.debug("Python Modules: %s", python_modules)

    stack_exists, stack_outputs = cfn.does_stack_exist(stack_name=stack_name)
    if not stack_exists:
        LOGGER.warn("Seedkit/Stack does not exist")
        return
    domain = stack_outputs.get("CodeArtifactDomain")
    repository = stack_outputs.get("CodeArtifactRepository")

    if any([":" not in pm for pm in python_modules]):
        raise ValueError(
            "Invalid `python_module`. Modules are identified with '[package-name]:[directory]': %s", python_modules
        )

    out_dir = _prep_modules_directory()
    modules = {ms[0]: ms[1] for ms in [m.split(":") for m in python_modules]}

    for module, dir in modules.items():
        LOGGER.info("Creating working directory for Module %s", module)
        _bundle.generate_dir(out_dir=out_dir, dir=dir, name=module)

    for module, dir in modules.items():
        LOGGER.info("Deploy Module %s to Seedkit Domain/Repository %s/%s", module, domain, repository)
        subprocess.check_call(
            [os.path.join(out_dir, FILENAME), cast(str, domain), cast(str, repository), module, module]
        )