def fetch_toolkit_data()

in cli/aws_orbit/models/context.py [0:0]

• 70 lines of code
• 21 McCabe index (conditional complexity)

    def fetch_toolkit_data(context: V) -> None:
        _logger.debug("Fetching Toolkit data...")
        if not (isinstance(context, Context) or isinstance(context, FoundationContext)):
            raise ValueError("Unknown 'context' Type")

        top_level = "orbit" if isinstance(context, Context) else "orbit-f"

        resp_type = Dict[str, List[Dict[str, List[Dict[str, str]]]]]

        codeseeder_toolkit_stack_name = cfn.get_stack_name("orbit")
        codeseeder_toolkit_stack_exists, stack_outputs = cfn.does_stack_exist(stack_name=codeseeder_toolkit_stack_name)

        if codeseeder_toolkit_stack_exists:
            context.toolkit.kms_arn = stack_outputs["KmsKeyArn"]
            context.toolkit.s3_bucket = stack_outputs["Bucket"]
            context.toolkit.codeseeder_policy = stack_outputs["SeedkitResourcesPolicyArn"]
            context.toolkit.codebuild_project = stack_outputs["CodeBuildProject"]
            context.toolkit.codeartifact_domain = stack_outputs["CodeArtifactDomain"]
            context.toolkit.codeartifact_repo = stack_outputs["CodeArtifactRepository"]

        try:
            response: resp_type = boto3_client("cloudformation").describe_stacks(StackName=context.toolkit.stack_name)
            _logger.debug("%s stack found.", context.toolkit.stack_name)
        except botocore.exceptions.ClientError as ex:
            error: Dict[str, Any] = ex.response["Error"]
            if error["Code"] == "ValidationError" and f"{context.toolkit.stack_name} not found" in error["Message"]:
                _logger.debug("Toolkit stack not found.")
                return
            if (
                error["Code"] == "ValidationError"
                and f"{context.toolkit.stack_name} does not exist" in error["Message"]
            ):
                _logger.debug("Toolkit stack does not exist.")
                return
            raise
        if len(response["Stacks"]) < 1:
            _logger.debug("Toolkit stack not found.")
            return
        if "Outputs" not in response["Stacks"][0]:
            _logger.debug("Toolkit stack with empty outputs")
            return

        for output in response["Stacks"][0]["Outputs"]:
            if output["ExportName"] == f"{top_level}-{context.name}-deploy-id":
                _logger.debug("Export value: %s", output["OutputValue"])
                context.toolkit.deploy_id = output["OutputValue"]
            if output["ExportName"] == f"{top_level}-{context.name}-admin-role":
                _logger.debug("Export value: %s", output["OutputValue"])
                context.toolkit.admin_role = output["OutputValue"]
            if output["ExportName"] == f"{top_level}-{context.name}-admin-role-arn":
                _logger.debug("Export value: %s", output["OutputValue"])
                context.toolkit.admin_role_arn = output["OutputValue"]

        if context.toolkit.deploy_id is None:
            raise RuntimeError(
                f"Stack {context.toolkit.stack_name} does not have the expected {top_level}-{context.name}-deploy-id output."
            )
        if context.toolkit.kms_arn is None:
            raise RuntimeError(
                f"Stack {context.toolkit.stack_name} does not have the expected {top_level}-{context.name}-kms-arn output."
            )
        if context.toolkit.admin_role is None:
            raise RuntimeError(
                f"Stack {context.toolkit.stack_name} does not have the expected {top_level}-{context.name}-admin-role output."
            )
        if context.toolkit.admin_role_arn is None:
            raise RuntimeError(
                f"Stack {context.toolkit.stack_name} does not have the expected {top_level}-{context.name}-admin-role-arn output."
            )
        # context.toolkit.kms_alias = f"{top_level}-{context.name}-{context.toolkit.deploy_id}"

        context.cdk_toolkit.s3_bucket = (
            f"{top_level}-{context.name}-cdk-toolkit-{context.account_id}-{context.toolkit.deploy_id}"
        )

        if isinstance(context, Context):
            # Set the Helm repositories
            _logger.debug(f"context.helm_repository: s3://{context.toolkit.s3_bucket}/helm/repositories/env")
            context.helm_repository = f"s3://{context.toolkit.s3_bucket}/helm/repositories/env"

            _logger.debug("context.toolkit: %s", context.toolkit)

        _logger.debug("Toolkit data fetched successfully.")