def run()

in servicecatalog_puppet/workflow/manifest/generate_manifest_with_ids_task.py [0:0]
87 lines of code
10 McCabe index (conditional complexity)

    def run(self):
        self.debug("starting")
        new_manifest = copy.deepcopy(self.manifest)
        regions = config.get_regions(self.puppet_account_id)
        global_id_cache = dict()
        new_manifest["id_cache"] = global_id_cache

        for region in regions:
            regional_id_cache = dict()
            r = self.input().get(region)
            for launch_name, launch_details in self.manifest.get_launches_items():
                self.debug(
                    f"processing launch_name={launch_name} in {region} for id_cache generation"
                )
                target = r.get(launch_details.get("portfolio")).get("details")
                portfolio_id = json.loads(target.open("r").read()).get("portfolio_id")
                portfolio_name = launch_details.get("portfolio")
                if regional_id_cache.get(portfolio_name) is None:
                    regional_id_cache[portfolio_name] = dict(
                        id=portfolio_id, products=dict()
                    )
                    self.debug(f"added {portfolio_name}={portfolio_id} to id_cache")

                product = launch_details.get("product")
                target = (
                    r.get(launch_details.get("portfolio")).get("products").get(product)
                )
                all_details = json.loads(target.open("r").read())
                all_products_and_their_versions = all_details
                for p in all_products_and_their_versions:
                    product_name = p.get("Name")
                    self.debug(f"processing product_name={product_name}")
                    if (
                        regional_id_cache[portfolio_name]["products"].get(product_name)
                        is None
                    ):
                        regional_id_cache[portfolio_name]["products"][
                            product_name
                        ] = dict(id=p.get("ProductId"), versions=dict())
                        self.debug(f"added {product_name} to id_cache")

                    for a in p.get("provisioning_artifact_details"):
                        version_id = a.get("Id")
                        version_name = a.get("Name")
                        self.debug(
                            f"added version {version_name}={version_id} to id_cache"
                        )
                        regional_id_cache[portfolio_name]["products"][product_name][
                            "versions"
                        ][version_name] = version_id

            global_id_cache[region] = regional_id_cache

        bucket = f"sc-puppet-spoke-deploy-{self.puppet_account_id}"

        cached_output_signed_url = None
        if self.input().get("parameters") or self.input().get("parameter_by_paths"):

            with zipfile.ZipFile(
                "output/GetSSMParamTask.zip", "w", zipfile.ZIP_DEFLATED
            ) as zip:
                files = glob.glob("output/GetSSMParam*/**", recursive=True)
                for filename in files:
                    zip.write(filename, filename)

            with self.hub_client("s3") as s3:
                key = f"{os.getenv('CODEBUILD_BUILD_NUMBER', '0')}-cached-output.zip"
                s3.upload_file(
                    Filename="output/GetSSMParamTask.zip", Bucket=bucket, Key=key,
                )
                cached_output_signed_url = s3.generate_presigned_url(
                    "get_object",
                    Params={"Bucket": bucket, "Key": key},
                    ExpiresIn=60 * 60 * 24,
                )

        with self.hub_client("s3") as s3:
            manifest_content = yaml.safe_dump(json.loads(json.dumps(new_manifest)))
            key = f"{os.getenv('CODEBUILD_BUILD_NUMBER', '0')}.yaml"
            self.debug(f"Uploading generated manifest {key} to {bucket}")
            s3.put_object(
                Body=manifest_content, Bucket=bucket, Key=key,
            )
            self.debug(f"Generating presigned URL for {key}")
            signed_url = s3.generate_presigned_url(
                "get_object",
                Params={"Bucket": bucket, "Key": key},
                ExpiresIn=60 * 60 * 24,
            )

        self.write_output(
            dict(
                manifest_content=manifest_content,
                signed_url=signed_url,
                cached_output_signed_url=cached_output_signed_url,
            )
        )