14727141442

Committed 29 Apr 2025 08:47AM UTC coverage: 58.801% (-0.04%) from 58.84%

Build # 14727141442

Build Type

Pull #1331

github

Committed by

web-flow

Commit Message

Merge 84ae26013 into f5f06fd65

Pull Request Pull Request #1331: Add mypy, autoflake and ruff pre-commit checks

Run Details

173 of 231 new or added lines in 21 files covered. (74.89%)

11 existing lines in 7 files now uncovered.

1236 of 2102 relevant lines covered (58.8%)

0.59 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

61.11

/snapshot_manager/snapshot_manager/github_graphql.py

"""
GithubGraphQL
"""

import pathlib
from types import TracebackType
from typing import Any

import fnc
from requests import Session


class GithubGraphQL:
    """A lightweight Github GraphQL API client.

    In order to properly close the session, use this class as a context manager:

        with GithubGraphQL(token="<GITHUB_API_TOKEN>") as g:
            g.query_from_file(filename="query.graphql", variables=None)

    or call the close() method manually

        g = GithubGraphQL(token="<GITHUB_API_TOKEN>")
        g.close()
    """

    def __init__(
        self,
        token: str = "",
        endpoint: str = "https://api.github.com/graphql",
        raise_on_error: bool = False,
    ):
        """
        Creates a session with the given bearer `token` and `endpoint`.

        Args:
            token (str): Your personal access token in Github (see https://github.com/settings/tokens)
            endpoint (str): The endpoint to query GraphQL from
            raise_on_error (bool): If you want to raise an exception in case of an error
        """
        self.__endpoint = endpoint
        self.__token = token
        self.__encoding = "utf-8"
        self.__raise_on_error = raise_on_error
        self.__session = Session()
        self.__session.headers.update(
            {
                "Authorization": f"Bearer {self.__token}",
                # See https://graphql.org/learn/best-practices/#json-with-gzip
                "Accept-Encoding": "gzip",
                # See #
                # https://github.blog/2021-11-16-graphql-global-id-migration-update/
                "X-Github-Next-Global-ID": "1",
            }
        )

    @property
    def token(self) -> str:
        """Returns the bearer token."""
        return self.__token

    @property
    def encoding(self) -> str:
        """Returns the default encoding to be expected from query files."""
        return self.__encoding

    def run_from_file(
        self,
        filename: pathlib.Path | str,
        variables: dict[str, str | int] = dict(),
        raise_on_error: bool = False,
    ) -> Any:
        """
        Read the query/mutation from the given file and execute it with the variables
        applied. If not requested otherwise the plain result is returned.

        See also:
        https://docs.github.com/en/graphql/guides/forming-calls-with-graphql
        https://docs.github.com/en/graphql/overview/explorer

        Args:
            filename (str): The filename of the query/mutation file.
            variables (dict): The variables to be applied to the query/mutation.
            raise_on_error (bool): If you want to raise an exception in case of an error
        """
        with open(file=filename, encoding=self.encoding) as file_handle:
            query = file_handle.read()
        return self.run(
            query,
            variables,
        )

    def __enter__(self) -> "GithubGraphQL":
        return self

    # @property
    # def session_headers(self) -> CaseInsensitiveDict:
    #     """Returns the HTTP headers used for the session."""
    #     return self.__session.headers

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc_value: BaseException | None,
        exc_traceback: TracebackType | None,
    ) -> None:
        self.close()

    def close(self) -> None:
        """Closes the session."""
        self.__session.close()

    def run(
        self,
        query: str,
        variables: dict[str, str | int] = dict(),
        raise_on_error: bool = False,
    ) -> dict[Any, Any]:
        """
        Execute the query with the variables applied. If not requested otherwise
        the plain result is returned. If you want to raise an exception in case
        of an error you can set `raise_on_error` to `True`.

        Args:
            query (str): The GraphQL query.
            variables (dict): The variables to be applied to the query.
            raise_on_error (bool): If you want to raise an exception in case of an error

        Raises:
            RuntimeError: In case of an error when `raise_on_error` is `True`.

        Returns:
            Result: The result of the query. Inspect the result for errors!
        """
        req = self.__session.post(
            url=self.__endpoint, json={"query": query, "variables": variables}
        )
        req.raise_for_status()
        res = dict(req.json())
        if "errors" in res and (raise_on_error or self.__raise_on_error):
            raise RuntimeError(
                str(fnc.get("errors[0].message", res, default="GraphQL Error"))
            )
        return res

1	"""
2	GithubGraphQL
3	"""
4
5	import pathlib	1✔
6	from types import TracebackType	1✔
7	from typing import Any	1✔
8
9	import fnc	1✔
10	from requests import Session	1✔
11
12
13	class GithubGraphQL:	1✔
14	"""A lightweight Github GraphQL API client.
15
16	In order to properly close the session, use this class as a context manager:
17
18	with GithubGraphQL(token="<GITHUB_API_TOKEN>") as g:
19	g.query_from_file(filename="query.graphql", variables=None)
20
21	or call the close() method manually
22
23	g = GithubGraphQL(token="<GITHUB_API_TOKEN>")
24	g.close()
25	"""
26
27	def __init__(	1✔
28	self,
29	token: str = "",
30	endpoint: str = "https://api.github.com/graphql",
31	raise_on_error: bool = False,
32	):
33	"""
34	Creates a session with the given bearer `token` and `endpoint`.
35
36	Args:
37	token (str): Your personal access token in Github (see https://github.com/settings/tokens)
38	endpoint (str): The endpoint to query GraphQL from
39	raise_on_error (bool): If you want to raise an exception in case of an error
40	"""
41	self.__endpoint = endpoint	1✔
42	self.__token = token	1✔
43	self.__encoding = "utf-8"	1✔
44	self.__raise_on_error = raise_on_error	1✔
45	self.__session = Session()	1✔
46	self.__session.headers.update(	1✔
47	{
48	"Authorization": f"Bearer {self.__token}",
49	# See https://graphql.org/learn/best-practices/#json-with-gzip
50	"Accept-Encoding": "gzip",
51	# See #
52	# https://github.blog/2021-11-16-graphql-global-id-migration-update/
53	"X-Github-Next-Global-ID": "1",
54	}
55	)
56
57	@property	1✔
58	def token(self) -> str:	1✔
59	"""Returns the bearer token."""
60	return self.__token	×
61
62	@property	1✔
63	def encoding(self) -> str:	1✔
64	"""Returns the default encoding to be expected from query files."""
65	return self.__encoding	×
66
67	def run_from_file(	1✔
68	self,
69	filename: pathlib.Path \| str,
70	variables: dict[str, str \| int] = dict(),
71	raise_on_error: bool = False,
72	) -> Any:
73	"""
74	Read the query/mutation from the given file and execute it with the variables
75	applied. If not requested otherwise the plain result is returned.
76
77	See also:
78	https://docs.github.com/en/graphql/guides/forming-calls-with-graphql
79	https://docs.github.com/en/graphql/overview/explorer
80
81	Args:
82	filename (str): The filename of the query/mutation file.
83	variables (dict): The variables to be applied to the query/mutation.
84	raise_on_error (bool): If you want to raise an exception in case of an error
85	"""
86	with open(file=filename, encoding=self.encoding) as file_handle:	×
87	query = file_handle.read()	×
NEW 88	return self.run(	×
89	query,
90	variables,
91	)
92
93	def __enter__(self) -> "GithubGraphQL":	1✔
94	return self	×
95
96	# @property
97	# def session_headers(self) -> CaseInsensitiveDict:
98	# """Returns the HTTP headers used for the session."""
99	# return self.__session.headers
100
101	def __exit__(	1✔
102	self,
103	exc_type: type[BaseException] \| None,
104	exc_value: BaseException \| None,
105	exc_traceback: TracebackType \| None,
106	) -> None:
UNCOV 107	self.close()	×
108
109	def close(self) -> None:	1✔
110	"""Closes the session."""
111	self.__session.close()	×
112
113	def run(	1✔
114	self,
115	query: str,
116	variables: dict[str, str \| int] = dict(),
117	raise_on_error: bool = False,
118	) -> dict[Any, Any]:
119	"""
120	Execute the query with the variables applied. If not requested otherwise
121	the plain result is returned. If you want to raise an exception in case
122	of an error you can set `raise_on_error` to `True`.
123
124	Args:
125	query (str): The GraphQL query.
126	variables (dict): The variables to be applied to the query.
127	raise_on_error (bool): If you want to raise an exception in case of an error
128
129	Raises:
130	RuntimeError: In case of an error when `raise_on_error` is `True`.
131
132	Returns:
133	Result: The result of the query. Inspect the result for errors!
134	"""
135	req = self.__session.post(	×
136	url=self.__endpoint, json={"query": query, "variables": variables}
137	)
138	req.raise_for_status()	×
139	res = dict(req.json())	×
NEW 140	if "errors" in res and (raise_on_error or self.__raise_on_error):	×
141	raise RuntimeError(	×
142	str(fnc.get("errors[0].message", res, default="GraphQL Error"))
143	)
144	return res	×

fedora-llvm-team / llvm-snapshots / 14727141442

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous