30713661660

Committed 01 Aug 2026 06:56PM UTC coverage: 70.42% (+14.6%) from 55.824%

Build # 30713661660

Build Type

push

github

Committed by

1ucian0

Commit Message

refactor

Coverage Stats

1926 of 2735 relevant lines covered (70.42%)

0.7 hits per line

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

85.71

/ecosystem/github.py

# This code is part of Qiskit.
#
# (C) Copyright IBM 2023.
#
# This code is licensed under the Apache License, Version 2.0. You may
# obtain a copy of this license in the LICENSE.txt file in the root directory
# of this source tree or at https://www.apache.org/licenses/LICENSE-2.0.
#
# Any modifications or derivative works of this code must retain this
# copyright notice, and modified files need to carry a notice indicating
# that they have been altered from the originals.

"""GitHub section."""

from re import match
from functools import reduce
from jsonpath import findall

from .license import License
from .serializable import JsonSerializable, parse_date
from .error_handling import EcosystemError, logger
from .request import (
    request_json,
    parse_github_package_ids,
    parse_github_dependants,
    parse_github_contributors_sidebar,
    URL,
)


class GitHubData(JsonSerializable):
    """
    The GitHub data related to a project
    """

    dict_keys = [
        "url",
        "owner",
        "repo",
        "tree",
        "stars",
        "homepage",
        "license",
        "description",
        "estimated_contributors",
        "total_dependent_repositories",
        "total_dependent_packages",
        "private",
        "archived",
        "disabled",
        "last_commit",
        "last_activity",
    ]
    aliases = {
        "stars": "stargazers_count",
        "last_commit": "pushed_at",
        "url": "html_url",
    }
    json_types = {
        "homepage": lambda x: x or None,
        "private": lambda x: x or None,
        "archived": lambda x: x or None,
        "disabled": lambda x: x or None,
        "description": lambda x: x[:131] + "..." if len(str(x)) > 135 else x,
        "pushed_at": parse_date,
    }
    reduce = {}

    def __init__(self, owner: str, repo: str, tree: str = None, **kwargs):
        self.owner = owner
        self.repo = repo
        self.tree = tree
        self._kwargs = kwargs or {}
        self._json_repo = None
        self._json_events = None
        self._json_package_ids = None
        self._json_dependants = None
        self._json_contributors_sidebar = None

    def to_dict(self, keys=None) -> dict:
        return super().to_dict(keys=keys or GitHubData.dict_keys)

    @classmethod
    def from_dict(cls, dictionary: dict):
        if "license" in dictionary and dictionary["license"] is not None:
            dictionary["license"] = License(dictionary["license"], where="github")
        return super().from_dict(dictionary)

    @classmethod
    def from_url(cls, github_project_url: URL):
        """
        Builds a GitHubSection from an url. Returns None
        if the given url is not a GitHub url
        """
        if "github.com" not in github_project_url.hostname:
            # github_project_url is not a GitHub URL
            return None

        tree_path = None
        url_path = github_project_url.path.replace("github.com", "")
        if "/tree/" in url_path:
            new_path, tree_path = url_path.split("/tree/")
            logger.debug(
                "%s includes a branch or a subdirectories: %s | %s",
                url_path,
                new_path,
                tree_path,
            )
            url_path = new_path
        try:
            owner, repo = [
                c for c in url_path.split("/") if match(r"^[A-Za-z0-9_.-]+$", c)
            ]
        except ValueError as exc:
            raise EcosystemError(f"invalid GitHub url: {github_project_url}") from exc

        return GitHubData(owner=owner, repo=repo, tree=tree_path)

    def update_json(self):
        """
        Fetches remote data from:
          - api.github.com/repos/{self.owner}/{self.repo}
          - github.com/{self.owner}/{self.repo}/network/dependents
          - github.com/{self.owner}/{self.repo}
          - api.github.com/networks/{self.owner}/{self.repo}/events

        """
        self._json_repo = request_json(f"api.github.com/repos/{self.owner}/{self.repo}")
        self._json_events = request_json(
            f"api.github.com/networks/{self.owner}/{self.repo}/events"
        )
        self._json_contributors_sidebar = request_json(
            f"github.com/{self.owner}/{self.repo}/contributors_list?deferred=true",
            parser=parse_github_contributors_sidebar,
        )
        self._json_package_ids = request_json(
            f"github.com/{self.owner}/{self.repo}/network/dependents?dependent_type=REPOSITORY",
            parser=parse_github_package_ids,
        )
        self._json_package_ids = {
            v: k
            for v, k in self._json_package_ids.items()
            if isinstance(v, str) and not v.startswith("__")
        }
        self._json_dependants = {}
        for package, package_id in self._json_package_ids.items():
            try:
                self._json_dependants[package] = request_json(
                    f"github.com/{self.owner}/{self.repo}/network/dependents?"
                    f"dependent_type=REPOSITORY&package_id={package_id}",
                    parser=parse_github_dependants,
                )
            except EcosystemError:
                logger.warning("json_dependants could not be updated")

    def __getattr__(self, item):
        if self._json_repo:
            if item in GitHubData.aliases:
                item = GitHubData.aliases[item]

            json_elements = findall(item, self._json_repo)
            if item in GitHubData.json_types:
                json_elements = [GitHubData.json_types[item](e) for e in json_elements]

            if len(json_elements) == 1:
                return json_elements[0]

            if len(json_elements) >= 2:
                return reduce(GitHubData.reduce[item], json_elements)

            raise AttributeError(
                f"'{type(self).__name__}' object has no " f"attribute '{item}'"
            )

        if item in self._kwargs:
            return self._kwargs[item]

        raise AttributeError(
            f"'{type(self).__name__}' object has no attribute '{item}'"
        )

    def update_owner_repo(self):
        """
        Updates GitHub page when the repo was moved or renamed
        """
        if self._json_repo is None:
            self.update_json()
        owner = self._json_repo["owner"]["login"]
        repo = self._json_repo["name"]
        if self.owner != owner or self.repo != repo:
            logger.info("%s/%s moved to %s/%s", self.owner, self.repo, owner, repo)
            self.owner = owner
            self.repo = repo

    def dependants(self, refresh=False):
        """get the dependant data from (cached) JSON"""
        if refresh:
            self.update_json()
        return self._json_dependants

    def contributors_sidebar_data(self, refresh=False):
        """get the front page data from (cached) JSON"""
        if refresh:
            self.update_json()
        return self._json_contributors_sidebar

    @property
    def estimated_contributors(self):
        """..."""
        if self.contributors_sidebar_data():
            return self.contributors_sidebar_data()["estimated_contributors"]
        return self._kwargs.get("estimated_contributors")

    @property
    def total_dependent_repositories(self):
        """Sum of repository dependants"""
        if self.dependants():
            return sum(r.get("repositories", 0) for r in self.dependants().values())
        return self._kwargs.get("total_dependent_repositories")

    @property
    def total_dependent_packages(self):
        """Sum of package dependants"""
        if self.dependants():
            return sum(r.get("packages", 0) for r in self.dependants().values())
        return self._kwargs.get("total_dependent_packages")

    @property
    def last_activity(self):
        """The creation of the last event"""
        if self._json_events and self._json_events["data"]:
            return parse_date(self._json_events["data"][0]["created_at"])
        return parse_date(self._kwargs.get("last_activity"))

    @property
    def license(self):
        """The creation of the last event"""
        json_license_name = self._json_repo.get("license", {}).get("name")
        if json_license_name:
            return License(json_license_name, "github")
        if "license" in self._kwargs:
            if isinstance(self._kwargs["license"], License):
                return self._kwargs["license"]
            return License(self._kwargs["license"], "github")
        return None

1	# This code is part of Qiskit.
2	#
3	# (C) Copyright IBM 2023.
4	#
5	# This code is licensed under the Apache License, Version 2.0. You may
6	# obtain a copy of this license in the LICENSE.txt file in the root directory
7	# of this source tree or at https://www.apache.org/licenses/LICENSE-2.0.
8	#
9	# Any modifications or derivative works of this code must retain this
10	# copyright notice, and modified files need to carry a notice indicating
11	# that they have been altered from the originals.
12
13	"""GitHub section."""	1✔
14
15	from re import match	1✔
16	from functools import reduce	1✔
17	from jsonpath import findall	1✔
18
19	from .license import License	1✔
20	from .serializable import JsonSerializable, parse_date	1✔
21	from .error_handling import EcosystemError, logger	1✔
22	from .request import (	1✔
23	request_json,
24	parse_github_package_ids,
25	parse_github_dependants,
26	parse_github_contributors_sidebar,
27	URL,
28	)
29
30
31	class GitHubData(JsonSerializable):	1✔
32	"""
33	The GitHub data related to a project
34	"""
35
36	dict_keys = [	1✔
37	"url",
38	"owner",
39	"repo",
40	"tree",
41	"stars",
42	"homepage",
43	"license",
44	"description",
45	"estimated_contributors",
46	"total_dependent_repositories",
47	"total_dependent_packages",
48	"private",
49	"archived",
50	"disabled",
51	"last_commit",
52	"last_activity",
53	]
54	aliases = {	1✔
55	"stars": "stargazers_count",
56	"last_commit": "pushed_at",
57	"url": "html_url",
58	}
59	json_types = {	1✔
60	"homepage": lambda x: x or None,
61	"private": lambda x: x or None,
62	"archived": lambda x: x or None,
63	"disabled": lambda x: x or None,
64	"description": lambda x: x[:131] + "..." if len(str(x)) > 135 else x,
65	"pushed_at": parse_date,
66	}
67	reduce = {}	1✔
68
69	def __init__(self, owner: str, repo: str, tree: str = None, **kwargs):	1✔
70	self.owner = owner	1✔
71	self.repo = repo	1✔
72	self.tree = tree	1✔
73	self._kwargs = kwargs or {}	1✔
74	self._json_repo = None	1✔
75	self._json_events = None	1✔
76	self._json_package_ids = None	1✔
77	self._json_dependants = None	1✔
78	self._json_contributors_sidebar = None	1✔
79
80	def to_dict(self, keys=None) -> dict:	1✔
81	return super().to_dict(keys=keys or GitHubData.dict_keys)	1✔
82
83	@classmethod	1✔
84	def from_dict(cls, dictionary: dict):	1✔
85	if "license" in dictionary and dictionary["license"] is not None:	×
86	dictionary["license"] = License(dictionary["license"], where="github")	×
87	return super().from_dict(dictionary)	×
88
89	@classmethod	1✔
90	def from_url(cls, github_project_url: URL):	1✔
91	"""
92	Builds a GitHubSection from an url. Returns None
93	if the given url is not a GitHub url
94	"""
95	if "github.com" not in github_project_url.hostname:	1✔
96	# github_project_url is not a GitHub URL
97	return None	1✔
98
99	tree_path = None	1✔
100	url_path = github_project_url.path.replace("github.com", "")	1✔
101	if "/tree/" in url_path:	1✔
102	new_path, tree_path = url_path.split("/tree/")	1✔
103	logger.debug(	1✔
104	"%s includes a branch or a subdirectories: %s \| %s",
105	url_path,
106	new_path,
107	tree_path,
108	)
109	url_path = new_path	1✔
110	try:	1✔
111	owner, repo = [	1✔
112	c for c in url_path.split("/") if match(r"^[A-Za-z0-9_.-]+$", c)
113	]
114	except ValueError as exc:	1✔
115	raise EcosystemError(f"invalid GitHub url: {github_project_url}") from exc	1✔
116
117	return GitHubData(owner=owner, repo=repo, tree=tree_path)	1✔
118
119	def update_json(self):	1✔
120	"""
121	Fetches remote data from:
122	- api.github.com/repos/{self.owner}/{self.repo}
123	- github.com/{self.owner}/{self.repo}/network/dependents
124	- github.com/{self.owner}/{self.repo}
125	- api.github.com/networks/{self.owner}/{self.repo}/events
126
127	"""
128	self._json_repo = request_json(f"api.github.com/repos/{self.owner}/{self.repo}")	1✔
129	self._json_events = request_json(	1✔
130	f"api.github.com/networks/{self.owner}/{self.repo}/events"
131	)
132	self._json_contributors_sidebar = request_json(	1✔
133	f"github.com/{self.owner}/{self.repo}/contributors_list?deferred=true",
134	parser=parse_github_contributors_sidebar,
135	)
136	self._json_package_ids = request_json(	1✔
137	f"github.com/{self.owner}/{self.repo}/network/dependents?dependent_type=REPOSITORY",
138	parser=parse_github_package_ids,
139	)
140	self._json_package_ids = {	1✔
141	v: k
142	for v, k in self._json_package_ids.items()
143	if isinstance(v, str) and not v.startswith("__")
144	}
145	self._json_dependants = {}	1✔
146	for package, package_id in self._json_package_ids.items():	1✔
147	try:	1✔
148	self._json_dependants[package] = request_json(	1✔
149	f"github.com/{self.owner}/{self.repo}/network/dependents?"
150	f"dependent_type=REPOSITORY&package_id={package_id}",
151	parser=parse_github_dependants,
152	)
153	except EcosystemError:	×
154	logger.warning("json_dependants could not be updated")	×
155
156	def __getattr__(self, item):	1✔
157	if self._json_repo:	1✔
158	if item in GitHubData.aliases:	1✔
159	item = GitHubData.aliases[item]	1✔
160
161	json_elements = findall(item, self._json_repo)	1✔
162	if item in GitHubData.json_types:	1✔
163	json_elements = [GitHubData.json_types[item](e) for e in json_elements]	1✔
164
165	if len(json_elements) == 1:	1✔
166	return json_elements[0]	1✔
167
168	if len(json_elements) >= 2:	1✔
169	return reduce(GitHubData.reduce[item], json_elements)	×
170
171	raise AttributeError(	1✔
172	f"'{type(self).__name__}' object has no " f"attribute '{item}'"
173	)
174
175	if item in self._kwargs:	1✔
176	return self._kwargs[item]	1✔
177
178	raise AttributeError(	1✔
179	f"'{type(self).__name__}' object has no attribute '{item}'"
180	)
181
182	def update_owner_repo(self):	1✔
183	"""
184	Updates GitHub page when the repo was moved or renamed
185	"""
186	if self._json_repo is None:	1✔
187	self.update_json()	×
188	owner = self._json_repo["owner"]["login"]	1✔
189	repo = self._json_repo["name"]	1✔
190	if self.owner != owner or self.repo != repo:	1✔
191	logger.info("%s/%s moved to %s/%s", self.owner, self.repo, owner, repo)	1✔
192	self.owner = owner	1✔
193	self.repo = repo	1✔
194
195	def dependants(self, refresh=False):	1✔
196	"""get the dependant data from (cached) JSON"""
197	if refresh:	1✔
198	self.update_json()	1✔
199	return self._json_dependants	1✔
200
201	def contributors_sidebar_data(self, refresh=False):	1✔
202	"""get the front page data from (cached) JSON"""
203	if refresh:	1✔
204	self.update_json()	×
205	return self._json_contributors_sidebar	1✔
206
207	@property	1✔
208	def estimated_contributors(self):	1✔
209	"""..."""
210	if self.contributors_sidebar_data():	1✔
211	return self.contributors_sidebar_data()["estimated_contributors"]	×
212	return self._kwargs.get("estimated_contributors")	1✔
213
214	@property	1✔
215	def total_dependent_repositories(self):	1✔
216	"""Sum of repository dependants"""
217	if self.dependants():	1✔
218	return sum(r.get("repositories", 0) for r in self.dependants().values())	1✔
219	return self._kwargs.get("total_dependent_repositories")	1✔
220
221	@property	1✔
222	def total_dependent_packages(self):	1✔
223	"""Sum of package dependants"""
224	if self.dependants():	1✔
225	return sum(r.get("packages", 0) for r in self.dependants().values())	1✔
226	return self._kwargs.get("total_dependent_packages")	1✔
227
228	@property	1✔
229	def last_activity(self):	1✔
230	"""The creation of the last event"""
231	if self._json_events and self._json_events["data"]:	1✔
232	return parse_date(self._json_events["data"][0]["created_at"])	×
233	return parse_date(self._kwargs.get("last_activity"))	1✔
234
235	@property	1✔
236	def license(self):	1✔
237	"""The creation of the last event"""
238	json_license_name = self._json_repo.get("license", {}).get("name")	1✔
239	if json_license_name:	×
240	return License(json_license_name, "github")	×
241	if "license" in self._kwargs:	×
242	if isinstance(self._kwargs["license"], License):	×
243	return self._kwargs["license"]	×
244	return License(self._kwargs["license"], "github")	×
245	return None	×

Qiskit / ecosystem / 30713661660

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