Module ogr.services.pagure.service

Classes

class PagureService (token: str | None = None,
instance_url: str = 'https://src.fedoraproject.org',
read_only: bool = False,
insecure: bool = False,
max_retries: int | urllib3.util.retry.Retry = 3,
user_agent: str | None = None,
**kwargs)
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] = 3,
        user_agent: Optional[str] = None,
        **kwargs,
    ) -> None:
        super().__init__()
        self.instance_url = instance_url
        self._token = token
        self.read_only = read_only

        self.session = requests.session()

        retry_config = create_retry_config(max_retries)

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

        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 {}
        # By default ogr deals with Pagure API -> json
        self.header["Accept"] = "application/json"

        if user_agent:
            self.header |= {"User-Agent": user_agent}
        else:
            try:
                from ogr import __version__ as ogr_version
            except ImportError:
                ogr_version = "dev"
            self.header |= {"User-Agent": f"ogr/{ogr_version} (hello@packit.dev)"}

        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,
        header: 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.
            header: HTTP request header, dict that will update default headers
            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,
                header=header,
            )

        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.
        """

        headers = self.header | (header or {})

        response = self.session.request(
            method=method,
            url=url,
            params=params,
            headers=headers,
            data=data,
            verify=not self.insecure,
        )
        logger.debug(
            f"Ogr sent request with following headers: {headers | {'Authorization': '<redacted>'}}",
        )

        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))

    def get_rate_limit_remaining(
        self,
        namespace: Optional[str] = None,
        repo: Optional[str] = None,
    ) -> Optional[int]:
        """
        There is no way to check rate limit status from Pagure API.
        """
        return None
Attributes
-----=
instance_url : str
URL of the git forge instance.

Ancestors

Instance variables

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

URL to the Pagure API.

Methods

def call_api(self, url: str, method: str | None = None, params: dict | None = None, data=None) ‑> dict
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

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.

def call_api_raw(self,
url: str,
method: str | None = None,
params: dict | None = None,
header: dict | None = None,
data=None)
Expand source code 
def call_api_raw(
    self,
    url: str,
    method: Optional[str] = None,
    params: Optional[dict] = None,
    header: 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.
        header: HTTP request header, dict that will update default headers
        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,
            header=header,
        )

    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

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.
header
HTTP request header, dict that will update default headers
data
Data to be sent in form of a dictionary.

Returns -----= RequestResponse object that represents the response from the API endpoint.

def get_api_url(self, *args, add_api_endpoint_part: bool = True) ‑> str
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)

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

def get_api_version(self) ‑> str
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"]

Returns -----= Version of the Pagure API.

def get_error_codes(self)
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)

Returns -----= Dictionary with all error codes.

def get_group(self, group_name: str) ‑> PagureGroup
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))

Get a Pagure group by name.

def get_rate_limit_remaining(self, namespace: str | None = None, repo: str | None = None) ‑> int | None
Expand source code 
def get_rate_limit_remaining(
    self,
    namespace: Optional[str] = None,
    repo: Optional[str] = None,
) -> Optional[int]:
    """
    There is no way to check rate limit status from Pagure API.
    """
    return None

There is no way to check rate limit status from Pagure API.

def get_raw_request(self, url, method='GET', params=None, data=None, header=None) ‑> RequestResponse
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.
    """

    headers = self.header | (header or {})

    response = self.session.request(
        method=method,
        url=url,
        params=params,
        headers=headers,
        data=data,
        verify=not self.insecure,
    )
    logger.debug(
        f"Ogr sent request with following headers: {headers | {'Authorization': '<redacted>'}}",
    )

    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,
    )

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.

Inherited members