Package ogr

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

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

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

import contextlib
from importlib.metadata import PackageNotFoundError, distribution

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

with contextlib.suppress(PackageNotFoundError):
    __version__ = distribution(__name__).version

__all__ = [
    GithubService.__name__,
    PagureService.__name__,
    GitlabService.__name__,
    AuthMethod.__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) ‑> 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: Optional[dict[str, type[GitService]]] = None, custom_instances: Optional[collections.abc.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/hostname (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: Optional[dict[str, type[GitService]]] = None,
    custom_instances: Optional[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/hostname (`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.hostname] = 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: Optional[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/hostname (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: Optional[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/hostname (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: Optional[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/hostname (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: Optional[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/hostname (`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)

    parsed_url = parse_git_repo(url)
    for service, service_kls in mapping.items():
        if parse_git_repo(service).hostname in parsed_url.hostname:
            return service_kls

    return None

Classes

class AuthMethod (value, names=None, *, module=None, qualname=None, type=None, start=1)

An enumeration.

Expand source code

class AuthMethod(str, Enum):
    tokman = "tokman"
    github_app = "github_app"
    token = "token"

Ancestors

• builtins.str
• enum.Enum

Class variables

var github_app

var token

var tokman

class GithubService (token=None, read_only=False, github_app_id: Optional[str] = None, github_app_private_key: Optional[str] = None, github_app_private_key_path: Optional[str] = None, tokman_instance_url: Optional[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

@use_for_service("github.com")
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: Optional[str] = None,
        github_app_private_key: Optional[str] = None,
        github_app_private_key_path: Optional[str] = None,
        tokman_instance_url: Optional[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._default_auth_method = github_authentication
        self._other_auth_method: GithubAuthentication = None
        self._auth_methods: dict[AuthMethod, GithubAuthentication] = {}

        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:
                allowed_methods=["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._default_auth_method:
            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, AuthMethod.tokman),
            (GithubApp, AuthMethod.github_app),
            (TokenAuthentication, AuthMethod.token),
        ]
        for auth_class, auth_name in auth_methods:
            auth_inst = auth_class.try_create(**kwargs)
            self._auth_methods[auth_name] = auth_inst
            if not self._default_auth_method:
                self._default_auth_method = auth_inst

        return None if self._default_auth_method else TokenAuthentication(None)

    def set_auth_method(self, method: AuthMethod):
        if self._auth_methods[method]:
            logger.info("Forced Github auth method to %s", method)
            self._other_auth_method = self._auth_methods[method]
        else:
            raise GithubAPIException(
                f"Choosen authentication method ({method}) is not available",
            )

    def reset_auth_method(self):
        logger.info("Reset Github auth method to the default")
        self._other_auth_method = None

    @property
    def authentication(self):
        return self._other_auth_method or self._default_auth_method

    @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={self.authentication!s}{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._default_auth_method = 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: Optional[str] = None,
        user: Optional[str] = None,
        search_pattern: Optional[str] = None,
        language: Optional[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

• BaseGitService
• GitService
• OgrAbstractClass

Class variables

var github_class : type[github.MainClass.Github]

var instance_url : Optional[str]

Instance variables

var authentication

Expand source code

@property
def authentication(self):
    return self._other_auth_method or self._default_auth_method

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

• BaseGitService:
  • change_token
  • get_group
  • get_project
  • list_projects
  • project_create
  • reset_auth_method
  • set_auth_method
  • user
• GitService:
  • get_project_from_url
  • hostname

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

Attributes

instance_url : str

URL of the git forge instance.

Expand source code

@use_for_service("gitlab")  # anything containing a gitlab word in hostname
# + list of community-hosted instances based on the following list
# https://wiki.p2pfoundation.net/List_of_Community-Hosted_GitLab_Instances
@use_for_service("salsa.debian.org")
@use_for_service("git.fosscommunity.in")
@use_for_service("framagit.org")
@use_for_service("dev.gajim.org")
@use_for_service("git.coop")
@use_for_service("lab.libreho.st")
@use_for_service("git.linux-kernel.at")
@use_for_service("git.pleroma.social")
@use_for_service("git.silence.dev")
@use_for_service("code.videolan.org")
@use_for_service("source.puri.sm")
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 ""
        return (
            f"GitlabService(instance_url='{self.instance_url}'"
            f"{token_str}"
            f"{ssl_str})"
        )

    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: Optional[str] = None,
        user: Optional[str] = None,
        search_pattern: Optional[str] = None,
        language: Optional[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

        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()
            ]
        else:
            projects_to_convert = projects
        return [
            GitlabProject(
                repo=project.attributes["path"],
                namespace=project.attributes["namespace"]["full_path"],
                service=self,
            )
            for project in projects_to_convert
        ]

Ancestors

• BaseGitService
• GitService
• OgrAbstractClass

Class variables

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

• BaseGitService:
  • change_token
  • get_group
  • get_project
  • list_projects
  • project_create
  • reset_auth_method
  • set_auth_method
  • user
• GitService:
  • get_project_from_url
  • hostname

class PagureService (token: Optional[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

@use_for_service("pagure")
@use_for_service("src.fedoraproject.org")
@use_for_service("src.stg.fedoraproject.org")
@use_for_service("pkgs.fedoraproject.org")
@use_for_service("pkgs.stg.fedoraproject.org")
@use_for_service("git.centos.org")
@use_for_service("git.stg.centos.org")
class PagureService(BaseGitService):
    def __init__(
        self,
        token: Optional[str] = None,
        instance_url: str = "https://src.fedoraproject.org",
        read_only: bool = False,
        insecure: bool = False,
        max_retries: Union[int, 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 ""

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

    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)

        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:
            raise OgrException(f"Cannot parse project url: '{url}'")

        if not repo_url.is_fork:
            repo_url.username = None

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

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

    def call_api(
        self,
        url: str,
        method: Optional[str] = None,
        params: Optional[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,
                response_code=response.status_code,
            )

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

        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,
                    response_code=response.status_code,
                )
            raise PagureAPIException(
                f"Problem with Pagure API when calling '{url}'",
                response_code=response.status_code,
            )

        return response.json_content

    def call_api_raw(
        self,
        url: str,
        method: Optional[str] = None,
        params: Optional[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

        if response.status_code >= 500:
            raise GitForgeInternalError(
                f"Pagure API returned {response.status_code} status for `{url}`"
                f" with reason: `{response.reason}`",
            )

        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 self.call_api(request_url)

    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: Optional[str] = None,
        user: Optional[str] = None,
        search_pattern: Optional[str] = None,
        language: Optional[str] = None,
    ) -> list[GitProject]:
        raise OperationNotSupported

    def get_group(self, group_name: str) -> PagureGroup:
        """
        Get a Pagure group by name.
        """
        url = self.get_api_url("group", group_name)
        return PagureGroup(group_name, self.call_api(url))

Ancestors

• BaseGitService
• GitService
• OgrAbstractClass

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: Optional[str] = None, params: Optional[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: Optional[str] = None,
    params: Optional[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,
            response_code=response.status_code,
        )

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

    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,
                response_code=response.status_code,
            )
        raise PagureAPIException(
            f"Problem with Pagure API when calling '{url}'",
            response_code=response.status_code,
        )

    return response.json_content

def call_api_raw(self, url: str, method: Optional[str] = None, params: Optional[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: Optional[str] = None,
    params: Optional[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

    if response.status_code >= 500:
        raise GitForgeInternalError(
            f"Pagure API returned {response.status_code} status for `{url}`"
            f" with reason: `{response.reason}`",
        )

    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 self.call_api(request_url)

def get_group(self, group_name: str) ‑> PagureGroup

Get a Pagure group by name.

Expand source code

def get_group(self, group_name: str) -> PagureGroup:
    """
    Get a Pagure group by name.
    """
    url = self.get_api_url("group", group_name)
    return PagureGroup(group_name, self.call_api(url))

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

• BaseGitService:
  • get_project
• BaseGitService:
  • change_token
  • list_projects
  • project_create
  • reset_auth_method
  • set_auth_method
  • user
• GitService:
  • get_project_from_url
• GitService:
  • hostname