Package ogr

Module providing one api for multiple git services (github/gitlab/pagure)

Expand source code
# Copyright Contributors to the Packit project.
# SPDX-License-Identifier: MIT

"""
Module providing one api for multiple git services (github/gitlab/pagure)
"""

from pkg_resources import get_distribution, DistributionNotFound

from ogr.factory import (
    get_project,
    get_service_class,
    get_service_class_or_none,
    get_instances_from_dict,
)
from ogr.services.github import GithubService
from ogr.services.gitlab import GitlabService
from ogr.services.pagure import PagureService

try:
    __version__ = get_distribution(__name__).version
except DistributionNotFound:
    # package is not installed
    pass

__all__ = [
    GithubService.__name__,
    PagureService.__name__,
    GitlabService.__name__,
    get_project.__name__,
    get_service_class.__name__,
    get_service_class_or_none.__name__,
    get_instances_from_dict.__name__,
]

Sub-modules

ogr.abstract
ogr.constant
ogr.deprecation
ogr.exceptions
ogr.factory
ogr.parsing
ogr.read_only
ogr.services
ogr.utils

Functions

def get_instances_from_dict(instances: Dict[~KT, ~VT]) ‑> Set[GitService]

Load the service instances from the dictionary in the following form:

  • key : hostname, url or name that can be mapped to the service-type
  • value : dictionary with arguments used when creating a new instance of the service (passed to the __init__ method)

e.g.:

get_instances_from_dict({
    "github.com": {"token": "abcd"},
    "pagure": {
        "token": "abcd",
        "instance_url": "https://src.fedoraproject.org",
    },
}) == {
    GithubService(token="abcd"),
    PagureService(token="abcd", instance_url="https://src.fedoraproject.org")
}

When the mapping key->service-type is not recognised, you can add a type key to the dictionary and specify the type of the instance. (It can be either name, hostname or url. The used mapping is same as for key->service-type.)

The provided key is used as an instance_url and passed to the __init__ method as well.

e.g.:

get_instances_from_dict({
    "https://my.gtlb": {"token": "abcd", "type": "gitlab"},
}) == {GitlabService(token="abcd", instance_url="https://my.gtlb")}

Args

instances
Mapping from service name/url/hostname to attributes for the service creation.

Returns

Set of the service instances.

Expand source code
def get_instances_from_dict(instances: Dict) -> Set[GitService]:
    """
    Load the service instances from the dictionary in the following form:

    - `key`   : hostname, url or name that can be mapped to the service-type
    - `value` : dictionary with arguments used when creating a new instance of the
    service (passed to the `__init__` method)

    e.g.:
    ```py
    get_instances_from_dict({
        "github.com": {"token": "abcd"},
        "pagure": {
            "token": "abcd",
            "instance_url": "https://src.fedoraproject.org",
        },
    }) == {
        GithubService(token="abcd"),
        PagureService(token="abcd", instance_url="https://src.fedoraproject.org")
    }
    ```

    When the mapping `key->service-type` is not recognised, you can add a `type`
    key to the dictionary and specify the type of the instance.
    (It can be either name, hostname or url. The used mapping is same as for
    key->service-type.)

    The provided `key` is used as an `instance_url` and passed to the `__init__`
    method as well.

    e.g.:
    ```py
    get_instances_from_dict({
        "https://my.gtlb": {"token": "abcd", "type": "gitlab"},
    }) == {GitlabService(token="abcd", instance_url="https://my.gtlb")}
    ```

    Args:
        instances: Mapping from service name/url/hostname to attributes for the
            service creation.

    Returns:
        Set of the service instances.
    """
    services = set()
    for key, value in instances.items():
        service_kls = get_service_class_or_none(url=key)
        if not service_kls:
            if "type" not in value:
                raise OgrException(
                    f"No matching service was found for url '{key}'. "
                    f"Add the service name as a `type` attribute."
                )
            service_type = value["type"]
            if service_type not in _SERVICE_MAPPING:
                raise OgrException(
                    f"No matching service was found for type '{service_type}'."
                )

            service_kls = _SERVICE_MAPPING[service_type]
            value.setdefault("instance_url", key)
            del value["type"]

        service_instance = service_kls(**value)
        services.add(service_instance)

    return services
def get_project(url, service_mapping_update: Dict[str, Type[GitService]] = None, custom_instances: Iterable[GitService] = None, force_custom_instance: bool = True, **kwargs) ‑> GitProject

Return the project for the given URL.

Args

url
URL of the project, e.g. "https://github.com/packit/ogr".
service_mapping_update

Custom mapping from service url (str) to service class.

Defaults to no mapping.

custom_instances

List of instances that will be used when creating a project instance.

Defaults to None.

force_custom_instance

Force picking a Git service from the custom_instances list, if there is any provided, raise an error if that is not possible.

Defaults to True.

**kwargs
Arguments forwarded to init of the matching service.

Returns

GitProject using the matching implementation.

Expand source code
def get_project(
    url,
    service_mapping_update: Dict[str, Type[GitService]] = None,
    custom_instances: Iterable[GitService] = None,
    force_custom_instance: bool = True,
    **kwargs,
) -> GitProject:
    """
    Return the project for the given URL.

    Args:
        url: URL of the project, e.g. `"https://github.com/packit/ogr"`.
        service_mapping_update: Custom mapping from
            service url (`str`) to service class.

            Defaults to no mapping.
        custom_instances: List of instances that will be
            used when creating a project instance.

            Defaults to `None`.
        force_custom_instance: Force picking a Git service from the
            `custom_instances` list, if there is any provided, raise an error if
            that is not possible.

            Defaults to `True`.
        **kwargs: Arguments forwarded to __init__ of the matching service.

    Returns:
        `GitProject` using the matching implementation.
    """
    mapping = service_mapping_update.copy() if service_mapping_update else {}
    custom_instances = custom_instances or []
    for instance in custom_instances:
        mapping[instance.instance_url] = instance.__class__

    kls = get_service_class(url=url, service_mapping_update=mapping)
    parsed_repo_url = parse_git_repo(url)

    service = None
    if custom_instances:
        for service_inst in custom_instances:
            if (
                isinstance(service_inst, kls)
                and service_inst.hostname == parsed_repo_url.hostname
            ):
                service = service_inst
                break
        else:
            if force_custom_instance:
                raise OgrException(
                    f"Instance of type {kls.__name__} "
                    f"matching instance url '{url}' was not provided."
                )
    if not service:
        service = kls(instance_url=parsed_repo_url.get_instance_url(), **kwargs)
    return service.get_project_from_url(url=url)
def get_service_class(url: str, service_mapping_update: Dict[str, Type[GitService]] = None) ‑> Type[GitService]

Get the matching service class from the URL.

Args

url
URL of the project, e.g. "https://github.com/packit/ogr".
service_mapping_update

Custom mapping from service url (str) to service class.

Defaults to None.

Returns

Matched class (subclass of GitService).

Expand source code
def get_service_class(
    url: str, service_mapping_update: Dict[str, Type[GitService]] = None
) -> Type[GitService]:
    """
    Get the matching service class from the URL.

    Args:
        url: URL of the project, e.g. `"https://github.com/packit/ogr"`.
        service_mapping_update: Custom mapping from service url (str) to service
            class.

            Defaults to `None`.

    Returns:
        Matched class (subclass of `GitService`).
    """
    service_kls = get_service_class_or_none(
        url=url, service_mapping_update=service_mapping_update
    )
    if service_kls:
        return service_kls
    raise OgrException("No matching service was found.")
def get_service_class_or_none(url: str, service_mapping_update: Dict[str, Type[GitService]] = None) ‑> Optional[Type[GitService]]

Get the matching service class from the URL.

Args

url
URL of the project, e.g. "https://github.com/packit/ogr".
service_mapping_update

Custom mapping from service url (str) to service class.

Defaults to None.

Returns

Matched class (subclass of GitService) or None.

Expand source code
def get_service_class_or_none(
    url: str, service_mapping_update: Dict[str, Type[GitService]] = None
) -> Optional[Type[GitService]]:
    """
    Get the matching service class from the URL.

    Args:
        url: URL of the project, e.g. `"https://github.com/packit/ogr"`.
        service_mapping_update: Custom mapping from service url (`str`) to service
            class.

            Defaults to `None`.

    Returns:
        Matched class (subclass of `GitService`) or `None`.
    """
    mapping = {}
    mapping.update(_SERVICE_MAPPING)
    if service_mapping_update:
        mapping.update(service_mapping_update)

    for service, service_kls in mapping.items():
        if service in url:
            return service_kls

    return None

Classes

class GithubService (token=None, read_only=False, github_app_id: str = None, github_app_private_key: str = None, github_app_private_key_path: str = None, tokman_instance_url: str = None, github_authentication: GithubAuthentication = None, max_retries: Union[int, urllib3.util.retry.Retry] = 0, **kwargs)

Attributes

instance_url : str
URL of the git forge instance.

If multiple authentication methods are provided, they are prioritised: 1. Tokman 2. GithubApp 3. TokenAuthentication (which is also default one, that works without specified token)

Expand source code
class GithubService(BaseGitService):
    # class parameter could be used to mock Github class api
    github_class: Type[github.Github]
    instance_url = "https://github.com"

    def __init__(
        self,
        token=None,
        read_only=False,
        github_app_id: str = None,
        github_app_private_key: str = None,
        github_app_private_key_path: str = None,
        tokman_instance_url: str = None,
        github_authentication: GithubAuthentication = None,
        max_retries: Union[int, Retry] = 0,
        **kwargs,
    ):
        """
        If multiple authentication methods are provided, they are prioritised:
            1. Tokman
            2. GithubApp
            3. TokenAuthentication (which is also default one, that works without specified token)
        """
        super().__init__()
        self.read_only = read_only
        self.authentication = github_authentication

        if isinstance(max_retries, Retry):
            self._max_retries = max_retries
        else:
            self._max_retries = Retry(
                total=int(max_retries),
                read=0,
                # Retry mechanism active for these HTTP methods:
                method_whitelist=["DELETE", "GET", "PATCH", "POST", "PUT"],
                # Only retry on following HTTP status codes
                status_forcelist=[500, 503, 403, 401],
                raise_on_status=False,
            )

        if not self.authentication:
            self.__set_authentication(
                token=token,
                github_app_id=github_app_id,
                github_app_private_key=github_app_private_key,
                github_app_private_key_path=github_app_private_key_path,
                tokman_instance_url=tokman_instance_url,
                max_retries=self._max_retries,
            )

        if kwargs:
            logger.warning(f"Ignored keyword arguments: {kwargs}")

    def __set_authentication(self, **kwargs):
        auth_methods = [
            Tokman,
            GithubApp,
            TokenAuthentication,
        ]
        for auth_class in auth_methods:
            self.authentication = auth_class.try_create(**kwargs)
            if self.authentication:
                return

        return TokenAuthentication(None)

    @property
    def github(self):
        return self.authentication.pygithub_instance

    def __str__(self) -> str:
        readonly_str = ", read_only=True" if self.read_only else ""
        arguments = f", github_authentication={str(self.authentication)}{readonly_str}"

        if arguments:
            # remove the first '- '
            arguments = arguments[2:]

        return f"GithubService({arguments})"

    def __eq__(self, o: object) -> bool:
        if not issubclass(o.__class__, GithubService):
            return False

        return (
            self.read_only == o.read_only  # type: ignore
            and self.authentication == o.authentication  # type: ignore
        )

    def __hash__(self) -> int:
        return hash(str(self))

    def get_project(
        self, repo=None, namespace=None, is_fork=False, **kwargs
    ) -> "GithubProject":
        if is_fork:
            namespace = self.user.get_username()
        return GithubProject(
            repo=repo,
            namespace=namespace,
            service=self,
            read_only=self.read_only,
            **kwargs,
        )

    def get_project_from_github_repository(
        self, github_repo: PyGithubRepository.Repository
    ) -> "GithubProject":
        return GithubProject(
            repo=github_repo.name,
            namespace=github_repo.owner.login,
            github_repo=github_repo,
            service=self,
            read_only=self.read_only,
        )

    @property
    def user(self) -> GitUser:
        return GithubUser(service=self)

    def change_token(self, new_token: str) -> None:
        self.authentication = TokenAuthentication(new_token)

    def project_create(
        self,
        repo: str,
        namespace: Optional[str] = None,
        description: Optional[str] = None,
    ) -> "GithubProject":
        if namespace:
            try:
                owner = self.github.get_organization(namespace)
            except UnknownObjectException as ex:
                raise GithubAPIException(f"Group {namespace} not found.") from ex
        else:
            owner = self.github.get_user()

        try:
            new_repo = owner.create_repo(
                name=repo,
                description=description if description else github.GithubObject.NotSet,
            )
        except github.GithubException as ex:
            raise GithubAPIException("Project creation failed") from ex
        return GithubProject(
            repo=repo,
            namespace=namespace or owner.login,
            service=self,
            github_repo=new_repo,
        )

    def get_pygithub_instance(self, namespace: str, repo: str) -> PyGithubInstance:
        token = self.authentication.get_token(namespace, repo)
        return PyGithubInstance(login_or_token=token, retry=self._max_retries)

    def list_projects(
        self,
        namespace: str = None,
        user: str = None,
        search_pattern: str = None,
        language: str = None,
    ) -> List[GitProject]:

        search_query = ""

        if user:
            search_query += f"user:{user}"

        if language:
            search_query += f" language:{language}"

        projects: List[GitProject]
        projects = [
            GithubProject(
                repo=repo.name,
                namespace=repo.owner.login,
                github_repo=repo,
                service=self,
            )
            for repo in self.github.search_repositories(search_query, order="asc")
        ]

        if search_pattern:
            projects = [
                project
                for project in projects
                if re.search(search_pattern, project.repo)
            ]

        return projects

Ancestors

Class variables

var github_class : Type[github.MainClass.Github]
var instance_url : Optional[str]

Instance variables

var github
Expand source code
@property
def github(self):
    return self.authentication.pygithub_instance

Methods

def get_project_from_github_repository(self, github_repo: github.Repository.Repository) ‑> GithubProject
Expand source code
def get_project_from_github_repository(
    self, github_repo: PyGithubRepository.Repository
) -> "GithubProject":
    return GithubProject(
        repo=github_repo.name,
        namespace=github_repo.owner.login,
        github_repo=github_repo,
        service=self,
        read_only=self.read_only,
    )
def get_pygithub_instance(self, namespace: str, repo: str) ‑> github.MainClass.Github
Expand source code
def get_pygithub_instance(self, namespace: str, repo: str) -> PyGithubInstance:
    token = self.authentication.get_token(namespace, repo)
    return PyGithubInstance(login_or_token=token, retry=self._max_retries)

Inherited members

class GitlabService (token=None, instance_url=None, ssl_verify=True, **kwargs)

Attributes

instance_url : str
URL of the git forge instance.
Expand source code
class GitlabService(BaseGitService):
    name = "gitlab"

    def __init__(self, token=None, instance_url=None, ssl_verify=True, **kwargs):
        super().__init__(token=token)
        self.instance_url = instance_url or "https://gitlab.com"
        self.token = token
        self.ssl_verify = ssl_verify
        self._gitlab_instance = None

        if kwargs:
            logger.warning(f"Ignored keyword arguments: {kwargs}")

    @property
    def gitlab_instance(self) -> gitlab.Gitlab:
        if not self._gitlab_instance:
            self._gitlab_instance = gitlab.Gitlab(
                url=self.instance_url,
                private_token=self.token,
                ssl_verify=self.ssl_verify,
            )
            if self.token:
                self._gitlab_instance.auth()
        return self._gitlab_instance

    @property
    def user(self) -> GitUser:
        return GitlabUser(service=self)

    def __str__(self) -> str:
        token_str = (
            f", token='{self.token[:1]}***{self.token[-1:]}'" if self.token else ""
        )
        ssl_str = ", ssl_verify=False" if not self.ssl_verify else ""
        str_result = (
            f"GitlabService(instance_url='{self.instance_url}'"
            f"{token_str}"
            f"{ssl_str})"
        )
        return str_result

    def __eq__(self, o: object) -> bool:
        if not issubclass(o.__class__, GitlabService):
            return False

        return (
            self.token == o.token  # type: ignore
            and self.instance_url == o.instance_url  # type: ignore
            and self.ssl_verify == o.ssl_verify  # type: ignore
        )

    def __hash__(self) -> int:
        return hash(str(self))

    def get_project(
        self, repo=None, namespace=None, is_fork=False, **kwargs
    ) -> "GitlabProject":
        if is_fork:
            namespace = self.user.get_username()
        return GitlabProject(repo=repo, namespace=namespace, service=self, **kwargs)

    def get_project_from_project_id(self, iid: int) -> "GitlabProject":
        gitlab_repo = self.gitlab_instance.projects.get(iid)
        return GitlabProject(
            repo=gitlab_repo.attributes["path"],
            namespace=gitlab_repo.attributes["namespace"]["full_path"],
            service=self,
            gitlab_repo=gitlab_repo,
        )

    def change_token(self, new_token: str) -> None:
        self.token = new_token
        self._gitlab_instance = None

    def project_create(
        self,
        repo: str,
        namespace: Optional[str] = None,
        description: Optional[str] = None,
    ) -> "GitlabProject":
        data = {"name": repo}
        if namespace:
            try:
                group = self.gitlab_instance.groups.get(namespace)
            except gitlab.GitlabGetError as ex:
                raise GitlabAPIException(f"Group {namespace} not found.") from ex
            data["namespace_id"] = group.id

        if description:
            data["description"] = description
        try:
            new_project = self.gitlab_instance.projects.create(data)
        except gitlab.GitlabCreateError as ex:
            raise GitlabAPIException("Project already exists") from ex
        return GitlabProject(
            repo=repo, namespace=namespace, service=self, gitlab_repo=new_project
        )

    def list_projects(
        self,
        namespace: str = None,
        user: str = None,
        search_pattern: str = None,
        language: str = None,
    ) -> List[GitProject]:

        if namespace:
            group = self.gitlab_instance.groups.get(namespace)
            projects = group.projects.list(all=True)
        elif user:
            user_object = self.gitlab_instance.users.list(username=user)[0]
            projects = user_object.projects.list(all=True)
        else:
            raise OperationNotSupported

        gitlab_projects: List[GitProject]

        if language:
            # group.projects.list gives us a GroupProject instance
            # in order to be able to filter by language we need Project instance
            projects_to_convert = [
                self.gitlab_instance.projects.get(item.attributes["id"])
                for item in projects
                if language
                in self.gitlab_instance.projects.get(item.attributes["id"])
                .languages()
                .keys()
            ]
        else:
            projects_to_convert = projects
        gitlab_projects = [
            GitlabProject(
                repo=project.attributes["path"],
                namespace=project.attributes["namespace"]["full_path"],
                service=self,
            )
            for project in projects_to_convert
        ]

        return gitlab_projects

Ancestors

Class variables

var instance_url : Optional[str]
var name

Instance variables

var gitlab_instance : gitlab.client.Gitlab
Expand source code
@property
def gitlab_instance(self) -> gitlab.Gitlab:
    if not self._gitlab_instance:
        self._gitlab_instance = gitlab.Gitlab(
            url=self.instance_url,
            private_token=self.token,
            ssl_verify=self.ssl_verify,
        )
        if self.token:
            self._gitlab_instance.auth()
    return self._gitlab_instance

Methods

def get_project_from_project_id(self, iid: int) ‑> GitlabProject
Expand source code
def get_project_from_project_id(self, iid: int) -> "GitlabProject":
    gitlab_repo = self.gitlab_instance.projects.get(iid)
    return GitlabProject(
        repo=gitlab_repo.attributes["path"],
        namespace=gitlab_repo.attributes["namespace"]["full_path"],
        service=self,
        gitlab_repo=gitlab_repo,
    )

Inherited members

class PagureService (token: str = None, instance_url: str = 'https://src.fedoraproject.org', read_only: bool = False, insecure: bool = False, max_retries: Union[int, urllib3.util.retry.Retry] = 5, **kwargs)

Attributes

instance_url : str
URL of the git forge instance.
Expand source code
class PagureService(BaseGitService):
    def __init__(
        self,
        token: str = None,
        instance_url: str = "https://src.fedoraproject.org",
        read_only: bool = False,
        insecure: bool = False,
        max_retries: Union[int, requests.packages.urllib3.util.Retry] = 5,
        **kwargs,
    ) -> None:
        super().__init__()
        self.instance_url = instance_url
        self._token = token
        self.read_only = read_only

        self.session = requests.session()

        adapter = requests.adapters.HTTPAdapter(max_retries=max_retries)

        self.insecure = insecure
        if self.insecure:
            self.session.mount("http://", adapter)
        else:
            self.session.mount("https://", adapter)

        self.header = {"Authorization": "token " + self._token} if self._token else {}

        if kwargs:
            logger.warning(f"Ignored keyword arguments: {kwargs}")

    def __str__(self) -> str:
        token_str = (
            f", token='{self._token[:1]}***{self._token[-1:]}'" if self._token else ""
        )
        insecure_str = ", insecure=True" if self.insecure else ""
        readonly_str = ", read_only=True" if self.read_only else ""

        str_result = (
            f"PagureService(instance_url='{self.instance_url}'"
            f"{token_str}"
            f"{readonly_str}"
            f"{insecure_str})"
        )
        return str_result

    def __eq__(self, o: object) -> bool:
        if not issubclass(o.__class__, PagureService):
            return False

        return (
            self._token == o._token  # type: ignore
            and self.read_only == o.read_only  # type: ignore
            and self.instance_url == o.instance_url  # type: ignore
            and self.insecure == o.insecure  # type: ignore
            and self.header == o.header  # type: ignore
        )

    def __hash__(self) -> int:
        return hash(str(self))

    def get_project(self, **kwargs) -> "PagureProject":
        if "username" in kwargs:
            return PagureProject(service=self, **kwargs)
        else:
            return PagureProject(
                service=self, username=self.user.get_username(), **kwargs
            )

    def get_project_from_url(self, url: str) -> "PagureProject":
        repo_url = parse_git_repo(potential_url=url)
        if not repo_url.is_fork:
            repo_url.username = None

        project = self.get_project(
            repo=repo_url.repo,
            namespace=repo_url.namespace,
            is_fork=repo_url.is_fork,
            username=repo_url.username,
        )
        return project

    @property
    def user(self) -> "PagureUser":
        return PagureUser(service=self)

    def call_api(
        self, url: str, method: str = None, params: dict = None, data=None
    ) -> dict:
        """
        Call API endpoint.

        Args:
            url: URL to be called.
            method: Method of the HTTP request, e.g. `"GET"`, `"POST"`, etc.
            params: HTTP(S) query parameters in form of a dictionary.
            data: Data to be sent in form of a dictionary.

        Returns:
            Dictionary representing response.

        Raises:
            PagureAPIException, if error occurs.
        """
        response = self.call_api_raw(url=url, method=method, params=params, data=data)

        if response.status_code == 404:
            error_msg = (
                response.json_content["error"]
                if response.json_content and "error" in response.json_content
                else None
            )
            raise PagureAPIException(
                f"Page '{url}' not found when calling Pagure API.",
                pagure_error=error_msg,
            )

        if not response.json_content:
            logger.debug(response.content)
            raise PagureAPIException("Error while decoding JSON: {0}")

        if not response.ok:
            logger.error(response.json_content)
            if "error" in response.json_content:
                error_msg = response.json_content["error"]
                error_msg_ext = response.json_content.get("errors", "")
                msg = f"Pagure API returned an error when calling '{url}': {error_msg}"
                if error_msg_ext:
                    msg += f" - {error_msg_ext}"
                raise PagureAPIException(
                    msg,
                    pagure_error=error_msg,
                    pagure_response=response.json_content,
                )
            raise PagureAPIException(f"Problem with Pagure API when calling '{url}'")

        return response.json_content

    def call_api_raw(
        self, url: str, method: str = None, params: dict = None, data=None
    ):
        """
        Call API endpoint and returns raw response.

        Args:
            url: URL to be called.
            method: Method of the HTTP request, e.g. `"GET"`, `"POST"`, etc.
            params: HTTP(S) query parameters in form of a dictionary.
            data: Data to be sent in form of a dictionary.

        Returns:
            `RequestResponse` object that represents the response from the API
            endpoint.
        """

        method = method or "GET"
        try:
            response = self.get_raw_request(
                method=method, url=url, params=params, data=data
            )

        except requests.exceptions.ConnectionError as er:
            logger.error(er)
            raise OgrNetworkError(f"Cannot connect to url: '{url}'.") from er
        return response

    def get_raw_request(
        self, url, method="GET", params=None, data=None, header=None
    ) -> RequestResponse:
        """
        Call API endpoint and wrap the response in `RequestResponse` type.

        Args:
            url: URL to be called.
            method: Method of the HTTP request, e.g. `"GET"`, `"POST"`, etc.

                Defaults to `"GET"`.
            params: HTTP(S) query parameters in form of a dictionary.
            data: Data to be sent in form of a dictionary.
            header: Header of the HTTP request.

        Returns:
            `RequestResponse` object representing the response.

        Raises:
            ValueError, if JSON cannot be retrieved.
        """

        response = self.session.request(
            method=method,
            url=url,
            params=params,
            headers=header or self.header,
            data=data,
            verify=not self.insecure,
        )

        json_output = None
        try:
            json_output = response.json()
        except ValueError:
            logger.debug(response.text)

        return RequestResponse(
            status_code=response.status_code,
            ok=response.ok,
            content=response.content,
            json=json_output,
            reason=response.reason,
        )

    @property
    def api_url(self):
        """URL to the Pagure API."""
        return f"{self.instance_url}/api/0/"

    def get_api_url(self, *args, add_api_endpoint_part: bool = True) -> str:
        """
        Get a URL from its parts.

        Args:
            *args: String parts of the URL, e.g. `"a", "b"` will call `project/a/b`
            add_api_endpoint_part: Add part with API endpoint (`/api/0/`).

                Defaults to `True`.

        Returns:
            String
        """
        args_list: List[str] = []

        args_list += filter(lambda x: x is not None, args)

        if add_api_endpoint_part:
            return self.api_url + "/".join(args_list)
        return f"{self.instance_url}/" + "/".join(args_list)

    def get_api_version(self) -> str:
        """
        Returns:
            Version of the Pagure API.
        """
        request_url = self.get_api_url("version")
        return_value = self.call_api(request_url)
        return return_value["version"]

    def get_error_codes(self):
        """
        Returns:
            Dictionary with all error codes.
        """
        request_url = self.get_api_url("error_codes")
        return_value = self.call_api(request_url)
        return return_value

    def change_token(self, token: str):
        self._token = token
        self.header = {"Authorization": "token " + self._token}

    def __handle_project_create_fail(
        self, exception: PagureAPIException, namespace: str
    ) -> None:
        if (
            exception.pagure_response
            and exception.pagure_response["errors"]["namespace"][0]
            == "Not a valid choice"
        ):
            request_url = self.get_api_url("group", namespace)

            try:
                self.call_api(request_url, data={"projects": False})
            except PagureAPIException as ex:
                raise OgrException(f"Namespace doesn't exist ({namespace}).") from ex

            raise OgrException(
                "Cannot create project in given namespace (permissions)."
            )

        raise exception

    def project_create(
        self,
        repo: str,
        namespace: Optional[str] = None,
        description: Optional[str] = None,
    ) -> PagureProject:
        request_url = self.get_api_url("new")

        parameters = {"name": repo, "description": description, "wait": True}
        if not description:
            parameters["description"] = repo
        if namespace:
            parameters["namespace"] = namespace

        try:
            self.call_api(request_url, "POST", data=parameters)
        except PagureAPIException as ex:
            self.__handle_project_create_fail(ex, namespace)
        return PagureProject(repo=repo, namespace=namespace, service=self)

    def list_projects(
        self,
        namespace: str = None,
        user: str = None,
        search_pattern: str = None,
        language: str = None,
    ) -> List[GitProject]:
        raise OperationNotSupported

Ancestors

Class variables

var instance_url : Optional[str]

Instance variables

var api_url

URL to the Pagure API.

Expand source code
@property
def api_url(self):
    """URL to the Pagure API."""
    return f"{self.instance_url}/api/0/"

Methods

def call_api(self, url: str, method: str = None, params: dict = None, data=None) ‑> dict

Call API endpoint.

Args

url
URL to be called.
method
Method of the HTTP request, e.g. "GET", "POST", etc.
params
HTTP(S) query parameters in form of a dictionary.
data
Data to be sent in form of a dictionary.

Returns

Dictionary representing response.

Raises

PagureAPIException, if error occurs.

Expand source code
def call_api(
    self, url: str, method: str = None, params: dict = None, data=None
) -> dict:
    """
    Call API endpoint.

    Args:
        url: URL to be called.
        method: Method of the HTTP request, e.g. `"GET"`, `"POST"`, etc.
        params: HTTP(S) query parameters in form of a dictionary.
        data: Data to be sent in form of a dictionary.

    Returns:
        Dictionary representing response.

    Raises:
        PagureAPIException, if error occurs.
    """
    response = self.call_api_raw(url=url, method=method, params=params, data=data)

    if response.status_code == 404:
        error_msg = (
            response.json_content["error"]
            if response.json_content and "error" in response.json_content
            else None
        )
        raise PagureAPIException(
            f"Page '{url}' not found when calling Pagure API.",
            pagure_error=error_msg,
        )

    if not response.json_content:
        logger.debug(response.content)
        raise PagureAPIException("Error while decoding JSON: {0}")

    if not response.ok:
        logger.error(response.json_content)
        if "error" in response.json_content:
            error_msg = response.json_content["error"]
            error_msg_ext = response.json_content.get("errors", "")
            msg = f"Pagure API returned an error when calling '{url}': {error_msg}"
            if error_msg_ext:
                msg += f" - {error_msg_ext}"
            raise PagureAPIException(
                msg,
                pagure_error=error_msg,
                pagure_response=response.json_content,
            )
        raise PagureAPIException(f"Problem with Pagure API when calling '{url}'")

    return response.json_content
def call_api_raw(self, url: str, method: str = None, params: dict = None, data=None)

Call API endpoint and returns raw response.

Args

url
URL to be called.
method
Method of the HTTP request, e.g. "GET", "POST", etc.
params
HTTP(S) query parameters in form of a dictionary.
data
Data to be sent in form of a dictionary.

Returns

RequestResponse object that represents the response from the API endpoint.

Expand source code
def call_api_raw(
    self, url: str, method: str = None, params: dict = None, data=None
):
    """
    Call API endpoint and returns raw response.

    Args:
        url: URL to be called.
        method: Method of the HTTP request, e.g. `"GET"`, `"POST"`, etc.
        params: HTTP(S) query parameters in form of a dictionary.
        data: Data to be sent in form of a dictionary.

    Returns:
        `RequestResponse` object that represents the response from the API
        endpoint.
    """

    method = method or "GET"
    try:
        response = self.get_raw_request(
            method=method, url=url, params=params, data=data
        )

    except requests.exceptions.ConnectionError as er:
        logger.error(er)
        raise OgrNetworkError(f"Cannot connect to url: '{url}'.") from er
    return response
def get_api_url(self, *args, add_api_endpoint_part: bool = True) ‑> str

Get a URL from its parts.

Args

*args
String parts of the URL, e.g. "a", "b" will call project/a/b
add_api_endpoint_part

Add part with API endpoint (/api/0/).

Defaults to True.

Returns

String

Expand source code
def get_api_url(self, *args, add_api_endpoint_part: bool = True) -> str:
    """
    Get a URL from its parts.

    Args:
        *args: String parts of the URL, e.g. `"a", "b"` will call `project/a/b`
        add_api_endpoint_part: Add part with API endpoint (`/api/0/`).

            Defaults to `True`.

    Returns:
        String
    """
    args_list: List[str] = []

    args_list += filter(lambda x: x is not None, args)

    if add_api_endpoint_part:
        return self.api_url + "/".join(args_list)
    return f"{self.instance_url}/" + "/".join(args_list)
def get_api_version(self) ‑> str

Returns

Version of the Pagure API.

Expand source code
def get_api_version(self) -> str:
    """
    Returns:
        Version of the Pagure API.
    """
    request_url = self.get_api_url("version")
    return_value = self.call_api(request_url)
    return return_value["version"]
def get_error_codes(self)

Returns

Dictionary with all error codes.

Expand source code
def get_error_codes(self):
    """
    Returns:
        Dictionary with all error codes.
    """
    request_url = self.get_api_url("error_codes")
    return_value = self.call_api(request_url)
    return return_value
def get_raw_request(self, url, method='GET', params=None, data=None, header=None) ‑> RequestResponse

Call API endpoint and wrap the response in RequestResponse type.

Args

url
URL to be called.
method

Method of the HTTP request, e.g. "GET", "POST", etc.

Defaults to "GET".

params
HTTP(S) query parameters in form of a dictionary.
data
Data to be sent in form of a dictionary.
header
Header of the HTTP request.

Returns

RequestResponse object representing the response.

Raises

ValueError, if JSON cannot be retrieved.

Expand source code
def get_raw_request(
    self, url, method="GET", params=None, data=None, header=None
) -> RequestResponse:
    """
    Call API endpoint and wrap the response in `RequestResponse` type.

    Args:
        url: URL to be called.
        method: Method of the HTTP request, e.g. `"GET"`, `"POST"`, etc.

            Defaults to `"GET"`.
        params: HTTP(S) query parameters in form of a dictionary.
        data: Data to be sent in form of a dictionary.
        header: Header of the HTTP request.

    Returns:
        `RequestResponse` object representing the response.

    Raises:
        ValueError, if JSON cannot be retrieved.
    """

    response = self.session.request(
        method=method,
        url=url,
        params=params,
        headers=header or self.header,
        data=data,
        verify=not self.insecure,
    )

    json_output = None
    try:
        json_output = response.json()
    except ValueError:
        logger.debug(response.text)

    return RequestResponse(
        status_code=response.status_code,
        ok=response.ok,
        content=response.content,
        json=json_output,
        reason=response.reason,
    )

Inherited members