def _create_or_update_deployment_environment()

in asfyaml/feature/github/deployment_environments.py [0:0]

• 40 lines of code
• 9 McCabe index (conditional complexity)

def _create_or_update_deployment_environment(self: ASFGitHubFeature, env_name, env_config):
    wait_timer = env_config.get("wait_timer", 0)  # by default disabled

    # Get the user id for the required reviewers, this endpoint only accepts user ids not usernames
    def get_reviewer(reviewer: Mapping[str, Any]) -> ReviewerParams:
        reviewer_id = reviewer["id"]
        reviewer_type = reviewer.get("type", "User")
        github_id = _get_user_id(self, reviewer_id) if reviewer_type == "User" else _get_team_id(self, reviewer_id)
        return ReviewerParams(type_=reviewer_type, id_=github_id)

    required_reviewers = env_config.get("required_reviewers", [])
    required_reviewers_with_id = [get_reviewer(reviewer) for reviewer in required_reviewers]

    # prevent_self_review is not supported by pygithub yet, https://github.com/PyGithub/PyGithub/pull/3246 is open
    # prevent_self_review = env_config.get("prevent_self_review", True)

    if "deployment_branch_policy" in env_config:
        deployment_branch_policy = env_config.get("deployment_branch_policy")
        protected_branches = deployment_branch_policy.get("protected_branches", False)
        policies = deployment_branch_policy.get("policies", [])

        deployment_branch_policy = EnvironmentDeploymentBranchPolicyParams(
            protected_branches=protected_branches,
            custom_branch_policies=len(policies) > 0,
        )
    else:
        deployment_branch_policy = None
        policies = []

    print(f"Updates to deployment environment {env_name}")
    print(f"  - Set required_reviewers to {[r.id for r in required_reviewers_with_id]}")
    print(f"  - Set wait_timer to {wait_timer}")
    if deployment_branch_policy is None:
        print("  - Set deployment branch policy = None")
    else:
        print(
            f"  - Set deployment branch policy = "
            f"(protected_branches={deployment_branch_policy.protected_branches}, "
            f"custom_branch_policies={deployment_branch_policy.custom_branch_policies})"
        )

    if not self.noop("environment"):
        self.ghrepo.create_environment(
            environment_name=env_name,
            wait_timer=wait_timer,
            reviewers=required_reviewers_with_id,
            deployment_branch_policy=deployment_branch_policy,
        )

        if deployment_branch_policy is not None and deployment_branch_policy.custom_branch_policies is True:
            _create_or_update_deployment_branch_policy(self, env_name, policies)