def create_specs_file()

in aws/cloudformation.py [0:0]

• 32 lines of code
• 7 McCabe index (conditional complexity)

    def create_specs_file(self, specs_file, s3_bucket_name, efs_id):
        username = getpass.getuser()
        rand = "".join(random.choices(string.ascii_uppercase + string.digits, k=5))
        hash = f"{username}-{rand}"
        stack_name = f"torchelastic-{hash}"
        this_dir = os.path.dirname(__file__)
        cfn_template = os.path.join(this_dir, "cfn/setup.yml")
        sample_specs = os.path.join(this_dir, "config/sample_specs.json")

        params = {
            "WorkerRoleName": f"torchelastic_worker_role-{hash}",
            "RendezvousRoleName": f"torchelastic_rendezvous_role-{hash}",
        }

        if s3_bucket_name:
            params["S3BucketName"] = s3_bucket_name
        if efs_id:
            params["EFSFileSystemId"] = efs_id

        self.create_stack(stack_name, cfn_template, **params)

        for _ in wait_for(
            f"cfn stack: {stack_name} to create", timeout=600, interval=2
        ):
            status, outputs = self.describe_stack(stack_name)
            if status == "CREATE_COMPLETE":
                break
            elif status == "CREATE_FAILED" or status.startswith("ROLLBACK_"):
                # when stack creation fails cfn starts rolling the stack back
                raise RuntimeError(
                    f"Error creating stack {stack_name}, status = {status}"
                )

        outputs["User"] = username

        log.info(f"Writing specs file to: {specs_file}")
        with open(sample_specs) as f:
            specs_template = Template(f.read())
            specs_template.stream(**outputs).dump(specs_file)