How to use the packit.utils.nested_get function in packit

To help you get started, we’ve selected a few packit examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github packit-service / packit / packit / jobs.py View on Github external
def run(self):
        # rev is a commit
        # we use branch on purpose so we get the latest thing
        # TODO: check if rev is HEAD on {branch}, warn then?
        branch = nested_get(self.event, "msg", "commit", "branch")

        # self.project is dist-git, we need to get upstream

        dg = DistGit(self.config, self.package_config)
        self.package_config.upstream_project_url = (
            dg.get_project_url_from_distgit_spec()
        )

        if not self.package_config.upstream_project_url:
            raise PackitException(
                "URL in specfile is not set. We don't know where the upstream project lives."
            )

        n, r = get_namespace_and_repo_name(self.package_config.upstream_project_url)
        up = self.upstream_service.get_project(repo=r, namespace=n)
        lp = LocalProject(git_project=up)
github packit-service / packit / packit / jobs.py View on Github external
def get_job_input_from_dist_git_commit(
        self, event: dict
    ) -> Optional[Tuple[JobTriggerType, PackageConfig, GitProject]]:
        """ this corresponds to dist-git event when someone pushes new commits """
        topic = nested_get(event, "topic")
        logger.debug(f"topic = {topic}")
        if topic == NewDistGitCommit.topic:
            repo_namespace = nested_get(event, "msg", "commit", "namespace")
            repo_name = nested_get(event, "msg", "commit", "repo")
            ref = nested_get(event, "msg", "commit", "branch")
            if not (repo_namespace and repo_name):
                logger.warning(
                    "We could not figure out the full name of the repository."
                )
                return None
            if not ref:
                logger.warning("Target branch for the new commits is not set.")
                return None
            logger.info(
                f"New commits added to dist-git repo {repo_namespace}/{repo_name}, branch {ref}."
            )
            msg_id = nested_get(event, "msg_id")
            logger.info(f"msg_id = {msg_id}")
            dg_proj = self.pagure_service.get_project(
github packit-service / packit / packit / jobs.py View on Github external
def get_job_input_from_github_pr(
        self, event: dict
    ) -> Optional[Tuple[JobTriggerType, PackageConfig, GitProject]]:
        """ look into the provided event and see if it's one for a new github pr """
        action = nested_get(event, "action")
        logger.debug(f"action = {action}")
        pr_id = nested_get(event, "number")
        is_pr = nested_get(event, "pull_request")
        if not is_pr:
            logger.info("Not a pull request event.")
            return None
        if action in ["opened", "reopened", "synchronize"] and pr_id:
            # we can't use head repo here b/c the app is set up against the upstream repo
            # and not the fork, on the other hand, we don't process packit.yaml from
            # the PR but what's in the upstream
            base_repo_namespace = nested_get(
                event, "pull_request", "base", "repo", "owner", "login"
            )
            base_repo_name = nested_get(event, "pull_request", "base", "repo", "name")

            if not (base_repo_name and base_repo_namespace):
                logger.warning(
                    "We could not figure out the full name of the repository."
                )
                return None
            base_ref = nested_get(event, "pull_request", "head", "sha")
            if not base_ref:
                logger.warning("Ref where the PR is coming from is not set.")
                return None
            target_repo = nested_get(event, "repository", "full_name")
            logger.info(f"GitHub pull request {pr_id} event for repo {target_repo}.")
            gh_proj = get_github_project(
github packit-service / packit / packit / jobs.py View on Github external
return None
        if action in ["opened", "reopened", "synchronize"] and pr_id:
            # we can't use head repo here b/c the app is set up against the upstream repo
            # and not the fork, on the other hand, we don't process packit.yaml from
            # the PR but what's in the upstream
            base_repo_namespace = nested_get(
                event, "pull_request", "base", "repo", "owner", "login"
            )
            base_repo_name = nested_get(event, "pull_request", "base", "repo", "name")

            if not (base_repo_name and base_repo_namespace):
                logger.warning(
                    "We could not figure out the full name of the repository."
                )
                return None
            base_ref = nested_get(event, "pull_request", "head", "sha")
            if not base_ref:
                logger.warning("Ref where the PR is coming from is not set.")
                return None
            target_repo = nested_get(event, "repository", "full_name")
            logger.info(f"GitHub pull request {pr_id} event for repo {target_repo}.")
            gh_proj = get_github_project(
                self.config, repo=base_repo_name, namespace=base_repo_namespace
            )
            package_config = get_package_config_from_repo(gh_proj, base_ref)
            https_url = event["repository"]["html_url"]
            package_config.upstream_project_url = https_url
            return JobTriggerType.pull_request, package_config, gh_proj
        return None
github packit-service / packit / packit / jobs.py View on Github external
"""
        look into the provided event and see if it's one for a published github release;
        if it is, process it and return input for the job handler
        """
        action = nested_get(event, "action")
        logger.debug(f"action = {action}")
        release = nested_get(event, "release")
        if action == "published" and release:
            repo_namespace = nested_get(event, "repository", "owner", "login")
            repo_name = nested_get(event, "repository", "name")
            if not (repo_namespace and repo_name):
                logger.warning(
                    "We could not figure out the full name of the repository."
                )
                return None
            release_ref = nested_get(event, "release", "tag_name")
            if not release_ref:
                logger.warning("Release tag name is not set.")
                return None
            logger.info(
                f"New release event {release_ref} for repo {repo_namespace}/{repo_name}."
            )
            gh_proj = get_github_project(
                self.config, repo=repo_name, namespace=repo_namespace
            )
            package_config = get_package_config_from_repo(gh_proj, release_ref)
            https_url = event["repository"]["html_url"]
            package_config.upstream_project_url = https_url
            return JobTriggerType.release, package_config, gh_proj
        return None
github packit-service / packit / packit / jobs.py View on Github external
def get_job_input_from_github_pr(
        self, event: dict
    ) -> Optional[Tuple[JobTriggerType, PackageConfig, GitProject]]:
        """ look into the provided event and see if it's one for a new github pr """
        action = nested_get(event, "action")
        logger.debug(f"action = {action}")
        pr_id = nested_get(event, "number")
        is_pr = nested_get(event, "pull_request")
        if not is_pr:
            logger.info("Not a pull request event.")
            return None
        if action in ["opened", "reopened", "synchronize"] and pr_id:
            # we can't use head repo here b/c the app is set up against the upstream repo
            # and not the fork, on the other hand, we don't process packit.yaml from
            # the PR but what's in the upstream
            base_repo_namespace = nested_get(
                event, "pull_request", "base", "repo", "owner", "login"
            )
            base_repo_name = nested_get(event, "pull_request", "base", "repo", "name")

            if not (base_repo_name and base_repo_namespace):
                logger.warning(
                    "We could not figure out the full name of the repository."
                )
github packit-service / packit / packit / jobs.py View on Github external
if topic == NewDistGitCommit.topic:
            repo_namespace = nested_get(event, "msg", "commit", "namespace")
            repo_name = nested_get(event, "msg", "commit", "repo")
            ref = nested_get(event, "msg", "commit", "branch")
            if not (repo_namespace and repo_name):
                logger.warning(
                    "We could not figure out the full name of the repository."
                )
                return None
            if not ref:
                logger.warning("Target branch for the new commits is not set.")
                return None
            logger.info(
                f"New commits added to dist-git repo {repo_namespace}/{repo_name}, branch {ref}."
            )
            msg_id = nested_get(event, "msg_id")
            logger.info(f"msg_id = {msg_id}")
            dg_proj = self.pagure_service.get_project(
                repo=repo_name, namespace=repo_namespace
            )
            package_config = get_package_config_from_repo(dg_proj, ref)
            return JobTriggerType.commit, package_config, dg_proj
        return None
github packit-service / packit / packit / jobs.py View on Github external
def get_job_input_from_github_release(
        self, event: dict
    ) -> Optional[Tuple[JobTriggerType, PackageConfig, GitProject]]:
        """
        look into the provided event and see if it's one for a published github release;
        if it is, process it and return input for the job handler
        """
        action = nested_get(event, "action")
        logger.debug(f"action = {action}")
        release = nested_get(event, "release")
        if action == "published" and release:
            repo_namespace = nested_get(event, "repository", "owner", "login")
            repo_name = nested_get(event, "repository", "name")
            if not (repo_namespace and repo_name):
                logger.warning(
                    "We could not figure out the full name of the repository."
                )
                return None
            release_ref = nested_get(event, "release", "tag_name")
            if not release_ref:
                logger.warning("Release tag name is not set.")
                return None
            logger.info(
                f"New release event {release_ref} for repo {repo_namespace}/{repo_name}."
            )
            gh_proj = get_github_project(
                self.config, repo=repo_name, namespace=repo_namespace
            )
github packit-service / packit / packit / jobs.py View on Github external
def get_job_input_from_github_release(
        self, event: dict
    ) -> Optional[Tuple[JobTriggerType, PackageConfig, GitProject]]:
        """
        look into the provided event and see if it's one for a published github release;
        if it is, process it and return input for the job handler
        """
        action = nested_get(event, "action")
        logger.debug(f"action = {action}")
        release = nested_get(event, "release")
        if action == "published" and release:
            repo_namespace = nested_get(event, "repository", "owner", "login")
            repo_name = nested_get(event, "repository", "name")
            if not (repo_namespace and repo_name):
                logger.warning(
                    "We could not figure out the full name of the repository."
                )
                return None
            release_ref = nested_get(event, "release", "tag_name")
            if not release_ref:
                logger.warning("Release tag name is not set.")
                return None
            logger.info(
                f"New release event {release_ref} for repo {repo_namespace}/{repo_name}."
            )
            gh_proj = get_github_project(
                self.config, repo=repo_name, namespace=repo_namespace
            )
            package_config = get_package_config_from_repo(gh_proj, release_ref)
github packit-service / packit / packit / jobs.py View on Github external
def get_job_input_from_github_pr(
        self, event: dict
    ) -> Optional[Tuple[JobTriggerType, PackageConfig, GitProject]]:
        """ look into the provided event and see if it's one for a new github pr """
        action = nested_get(event, "action")
        logger.debug(f"action = {action}")
        pr_id = nested_get(event, "number")
        is_pr = nested_get(event, "pull_request")
        if not is_pr:
            logger.info("Not a pull request event.")
            return None
        if action in ["opened", "reopened", "synchronize"] and pr_id:
            # we can't use head repo here b/c the app is set up against the upstream repo
            # and not the fork, on the other hand, we don't process packit.yaml from
            # the PR but what's in the upstream
            base_repo_namespace = nested_get(
                event, "pull_request", "base", "repo", "owner", "login"
            )
            base_repo_name = nested_get(event, "pull_request", "base", "repo", "name")

            if not (base_repo_name and base_repo_namespace):