Coverage for pyTooling / Dependency / Python.py: 68%
328 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-02-13 22:36 +0000
1# ==================================================================================================================== #
2# _____ _ _ ____ _ #
3# _ __ _ |_ _|__ ___ | (_)_ __ __ _ | _ \ ___ _ __ ___ _ __ __| | ___ _ __ ___ _ _ #
4# | '_ \| | | || |/ _ \ / _ \| | | '_ \ / _` | | | | |/ _ \ '_ \ / _ \ '_ \ / _` |/ _ \ '_ \ / __| | | | #
5# | |_) | |_| || | (_) | (_) | | | | | | (_| |_| |_| | __/ |_) | __/ | | | (_| | __/ | | | (__| |_| | #
6# | .__/ \__, ||_|\___/ \___/|_|_|_| |_|\__, (_)____/ \___| .__/ \___|_| |_|\__,_|\___|_| |_|\___|\__, | #
7# |_| |___/ |___/ |_| |___/ #
8# ==================================================================================================================== #
9# Authors: #
10# Patrick Lehmann #
11# #
12# License: #
13# ==================================================================================================================== #
14# Copyright 2025-2026 Patrick Lehmann - Bötzingen, Germany #
15# #
16# Licensed under the Apache License, Version 2.0 (the "License"); #
17# you may not use this file except in compliance with the License. #
18# You may obtain a copy of the License at #
19# #
20# http://www.apache.org/licenses/LICENSE-2.0 #
21# #
22# Unless required by applicable law or agreed to in writing, software #
23# distributed under the License is distributed on an "AS IS" BASIS, #
24# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. #
25# See the License for the specific language governing permissions and #
26# limitations under the License. #
27# #
28# SPDX-License-Identifier: Apache-2.0 #
29# ==================================================================================================================== #
30#
31"""
32Implementation of package dependencies.
34.. hint::
36 See :ref:`high-level help <DEPENDENCIES>` for explanations and usage examples.
37"""
38from asyncio import run as asyncio_run, gather as asyncio_gather
39from datetime import datetime
40from enum import IntEnum
41from functools import wraps, update_wrapper
42from threading import RLock
43from typing import Optional as Nullable, List, Dict, Union, Iterable, Mapping
45try:
46 from aiohttp import ClientSession
47except ImportError as ex: # pragma: no cover
48 raise Exception(f"Optional dependency 'aiohttp' not installed. Either install pyTooling with extra dependencies 'pyTooling[pypi]' or install 'aiohttp' directly.") from ex
50try:
51 from packaging.requirements import Requirement
52except ImportError as ex: # pragma: no cover
53 raise Exception(f"Optional dependency 'packaging' not installed. Either install pyTooling with extra dependencies 'pyTooling[pypi]' or install 'packaging' directly.") from ex
55try:
56 from requests import Session, HTTPError
57except ImportError as ex: # pragma: no cover
58 raise Exception(f"Optional dependency 'requests' not installed. Either install pyTooling with extra dependencies 'pyTooling[pypi]' or install 'requests' directly.") from ex
60from pyTooling.Decorators import export, readonly
61from pyTooling.MetaClasses import ExtendedType, abstractmethod
62from pyTooling.Exceptions import ToolingException
63from pyTooling.Common import getFullyQualifiedName, firstValue
64from pyTooling.Dependency import Package, PackageStorage, PackageVersion, PackageDependencyGraph
65from pyTooling.GenericPath.URL import URL
66from pyTooling.Versioning import SemanticVersion, PythonVersion, Parts
69@export
70class LazyLoaderState(IntEnum):
71 Uninitialized = 0 #: No data or minimal data like ID or name.
72 Initialized = 1 #: Initialized by some __init__ parameters.
73 PartiallyLoaded = 2 #: Some additional data was loaded.
74 FullyLoaded = 3 #: All data is loaded.
75 PostProcessed = 4 #: Loaded data triggered further processing.
78@export
79class lazy:
80 """
81 Unified decorator that supports:
82 1. @lazy(state) def method()
83 2. @lazy(state) @property def prop()
84 """
86 def __init__(self, _requiredState: LazyLoaderState = LazyLoaderState.PartiallyLoaded):
87 self._requiredState = _requiredState
88 self._wrapped = None
90 def __call__(self, wrapped):
91 self._wrapped = wrapped
92 # If it's a function, we update metadata.
93 # If it's a property, it doesn't support update_wrapper directly.
94 if hasattr(wrapped, "__name__"):
95 update_wrapper(self, wrapped)
97 return self
99 def __get__(self, obj, objtype=None):
100 if obj is None: 100 ↛ 101line 100 didn't jump to line 101 because the condition on line 100 was never true
101 return self
103 # 1. Thread-safe state check
104 with obj.__lazy_lock__:
105 if obj.__lazy_state__ < self._requiredState:
106 obj.__lazy_loader__(self._requiredState)
108 # 2. Determine if we are wrapping a property or a method
109 if isinstance(self._wrapped, property):
110 # If it's a property, call its __get__ to return the value
111 return self._wrapped.__get__(obj, objtype)
113 # 3. Otherwise, treat as a method and return a bound wrapper
114 @wraps(self._wrapped)
115 def wrapper(*args, **kwargs):
116 return self._wrapped(obj, *args, **kwargs)
118 return wrapper
121@export
122class LazyLoadableMixin(metaclass=ExtendedType, mixin=True):
123 __lazy_state__: LazyLoaderState
124 __lazy_lock__: RLock
126 def __init__(self, targetLevel: LazyLoaderState = LazyLoaderState.Initialized) -> None:
127 self.__lazy_state__ = LazyLoaderState.Initialized
128 self.__lazy_lock__ = RLock()
130 if targetLevel > self.__lazy_state__:
131 with self.__lazy_lock__:
132 self.__lazy_loader__(targetLevel)
134 @abstractmethod
135 def __lazy_loader__(self, targetLevel: LazyLoaderState) -> None:
136 pass
139@export
140class Distribution(metaclass=ExtendedType, slots=True):
141 _filename: str
142 _url: URL
143 _uploadTime: datetime
145 def __init__(self, filename: str, url: Union[str, URL], uploadTime: datetime) -> None:
146 if not isinstance(filename, str): 146 ↛ 147line 146 didn't jump to line 147 because the condition on line 146 was never true
147 ex = TypeError("Parameter 'filename' is not of type 'str'.")
148 ex.add_note(f"Got type '{getFullyQualifiedName(filename)}'.")
149 raise ex
151 self._filename = filename
153 if isinstance(url, str): 153 ↛ 155line 153 didn't jump to line 155 because the condition on line 153 was always true
154 url = URL.Parse(url)
155 elif not isinstance(url, URL):
156 ex = TypeError("Parameter 'url' is not of type 'URL'.")
157 ex.add_note(f"Got type '{getFullyQualifiedName(url)}'.")
158 raise ex
160 self._url = url
162 if not isinstance(uploadTime, datetime): 162 ↛ 163line 162 didn't jump to line 163 because the condition on line 162 was never true
163 ex = TypeError("Parameter 'uploadTime' is not of type 'str'.")
164 ex.add_note(f"Got type '{getFullyQualifiedName(uploadTime)}'.")
165 raise ex
167 self._uploadTime = uploadTime
169 @readonly
170 def Filename(self) -> str:
171 return self._filename
173 @readonly
174 def URL(self) -> URL:
175 return self._url
177 @readonly
178 def UploadTime(self) -> datetime:
179 return self._uploadTime
181 def __repr__(self) -> str:
182 return f"Distribution: {self._filename}"
184 def __str__(self) -> str:
185 return f"{self._filename}"
188@export
189class Release(PackageVersion, LazyLoadableMixin):
190 _files: List[Distribution]
191 _requirements: Dict[Union[str, None], List[Requirement]]
193 _api: Nullable[URL]
194 _session: Nullable[Session]
196 def __init__(
197 self,
198 version: PythonVersion,
199 timestamp: datetime,
200 files: Nullable[Iterable[Distribution]] = None,
201 requirements: Nullable[Mapping[str, List[Requirement]]] = None,
202 project: Nullable["Project"] = None,
203 lazy: LazyLoaderState = LazyLoaderState.Initialized
204 ) -> None:
205 if project is not None and (storage := project._storage) is not None: 205 ↛ 209line 205 didn't jump to line 209 because the condition on line 205 was always true
206 self._api = storage._api
207 self._session = storage._session
208 else:
209 self._api = None
210 self._session = None
212 super().__init__(version, project, timestamp)
213 LazyLoadableMixin.__init__(self, lazy)
215 self._files = [file for file in files] if files is not None else []
216 self._requirements = {k: v for k, v in requirements} if requirements is not None else {None: []}
218 def __lazy_loader__(self, targetLevel: LazyLoaderState) -> None:
219 if targetLevel >= LazyLoaderState.PartiallyLoaded: 219 ↛ 221line 219 didn't jump to line 221 because the condition on line 219 was always true
220 self.DownloadDetails()
221 if targetLevel >= LazyLoaderState.PostProcessed: 221 ↛ 222line 221 didn't jump to line 222 because the condition on line 221 was never true
222 self.PostProcess()
224 @lazy(LazyLoaderState.PostProcessed)
225 @PackageVersion.DependsOn.getter
226 def DependsOn(self) -> Dict["Package", Dict[SemanticVersion, "PackageVersion"]]:
227 return super().DependsOn
229 @readonly
230 def Project(self) -> "Project":
231 return self._package
233 @lazy(LazyLoaderState.PartiallyLoaded)
234 @readonly
235 def Files(self) -> List[Distribution]:
236 return self._files
238 @lazy(LazyLoaderState.PartiallyLoaded)
239 @readonly
240 def Requirements(self) -> Dict[str, List[Requirement]]:
241 return self._requirements
243 def _GetPyPIEndpoint(self) -> str:
244 return f"{self._package._name.lower()}/{self._version}/json"
246 def DownloadDetails(self) -> None:
247 if self._session is None: 247 ↛ 249line 247 didn't jump to line 249 because the condition on line 247 was never true
248 # TODO: NoSessionAvailableException
249 raise ToolingException(f"No session available.")
251 response = self._session.get(url=f"{self._api}{self._GetPyPIEndpoint()}")
252 try:
253 response.raise_for_status()
254 except HTTPError as ex:
255 if ex.response.status_code == 404:
256 # TODO: ReleaseNotFoundException
257 raise ToolingException(f"Release '{self._version}' of package '{self._package._name}' not found.")
259 self.UpdateDetailsFromPyPIJSON(response.json())
261 index: PythonPackageIndex = self._package._storage
262 for requirement in self._requirements[None]:
263 packageName = requirement.name
264 index.DownloadProject(packageName, True)
266 def UpdateDetailsFromPyPIJSON(self, json) -> None:
267 infoNode = json["info"]
268 if (extras := infoNode["provides_extra"]) is not None:
269 self._requirements = {extra: [] for extra in extras}
270 self._requirements[None] = []
272 if (requirements := infoNode["requires_dist"]) is not None: 272 ↛ 293line 272 didn't jump to line 293 because the condition on line 272 was always true
273 brokenRequirements = []
274 for requirement in requirements:
275 req = Requirement(requirement)
277 # Handle requirements without an extra marker
278 if req.marker is None:
279 self._requirements[None].append(req)
280 continue
282 for extra in self._requirements.keys():
283 if extra is not None and req.marker.evaluate({"extra": extra}):
284 self._requirements[extra].append(req)
285 break
286 else:
287 brokenRequirements.append(req)
289 # TODO: raise a warning
290 if len(brokenRequirements) > 0:
291 self._requirements[0] = brokenRequirements
293 self.__lazy_state__ = LazyLoaderState.FullyLoaded
295 def PostProcess(self) -> None:
296 index: PythonPackageIndex = self._package._storage
297 for requirement in self._requirements[None]:
298 package = index.DownloadProject(requirement.name)
300 for release in package:
301 if str(release._version) in requirement.specifier:
302 self.AddDependencyToPackageVersion(release)
304 self.SortDependencies()
305 self.__lazy_state__ = LazyLoaderState.PostProcessed
307 @lazy(LazyLoaderState.PartiallyLoaded)
308 def __repr__(self) -> str:
309 return f"Release: {self._package._name}:{self._version} Files: {len(self._files)}"
311 def __str__(self) -> str:
312 return f"{self._version}"
315@export
316class Project(Package, LazyLoadableMixin):
317 _url: Nullable[URL]
319 _api: Nullable[URL]
320 _session: Nullable[Session]
322 def __init__(
323 self,
324 name: str,
325 url: Union[str, URL],
326 releases: Nullable[Iterable[Release]] = None,
327 index: Nullable["PythonPackageIndex"] = None,
328 lazy: LazyLoaderState = LazyLoaderState.Initialized
329 ) -> None:
330 if index is not None: 330 ↛ 334line 330 didn't jump to line 334 because the condition on line 330 was always true
331 self._api = index._api
332 self._session = index._session
333 else:
334 self._api = None
335 self._session = None
337 super().__init__(name, storage=index)
338 LazyLoadableMixin.__init__(self, lazy)
340 # if isinstance(url, str):
341 # url = URL.Parse(url)
342 # elif not isinstance(url, URL):
343 # ex = TypeError("Parameter 'url' is not of type 'URL'.")
344 # ex.add_note(f"Got type '{getFullyQualifiedName(url)}'.")
345 # raise ex
346 #
347 # self._url = url
348 # self._releases = {release.Version: release for release in sorted(releases, key=lambda r: r.Version)} if releases is not None else {}
350 def __lazy_loader__(self, targetLevel: LazyLoaderState) -> None:
351 if targetLevel >= LazyLoaderState.PartiallyLoaded: 351 ↛ 353line 351 didn't jump to line 353 because the condition on line 351 was always true
352 self.DownloadDetails()
353 if targetLevel >= LazyLoaderState.PostProcessed: 353 ↛ 354line 353 didn't jump to line 354 because the condition on line 353 was never true
354 self.DownloadReleaseDetails()
356 @readonly
357 def PackageIndex(self) -> "PythonPackageIndex":
358 return self._storage
360 @lazy(LazyLoaderState.PartiallyLoaded)
361 @readonly
362 def URL(self) -> URL:
363 return self._url
365 @lazy(LazyLoaderState.PartiallyLoaded)
366 @readonly
367 def Releases(self) -> Dict[PythonVersion, Release]:
368 return self._versions
370 @lazy(LazyLoaderState.PartiallyLoaded)
371 @readonly
372 def ReleaseCount(self) -> int:
373 return len(self._versions)
375 @lazy(LazyLoaderState.PartiallyLoaded)
376 @readonly
377 def LatestRelease(self) -> Release:
378 return firstValue(self._versions)
380 def _GetPyPIEndpoint(self) -> str:
381 return f"{self._name.lower()}/json"
383 def DownloadDetails(self) -> None:
384 if self._session is None: 384 ↛ 386line 384 didn't jump to line 386 because the condition on line 384 was never true
385 # TODO: NoSessionAvailableException
386 raise ToolingException(f"No session available.")
388 response = self._session.get(url=f"{self._api}{self._GetPyPIEndpoint()}")
389 try:
390 response.raise_for_status()
391 except HTTPError as ex:
392 if ex.response.status_code == 404:
393 # TODO: ReleaseNotFoundException
394 raise ToolingException(f"Package '{self._name}' not found.")
396 self.UpdateDetailsFromPyPIJSON(response.json())
398 def UpdateDetailsFromPyPIJSON(self, json) -> None:
399 infoNode = json["info"]
400 releasesNode = json["releases"]
402 # Update project/package URL
403 self._url = URL.Parse(infoNode["project_url"])
405 # Convert key to Version number, skip empty releases
406 convertedReleasesNode = {}
407 for k, v in releasesNode.items():
408 if len(v) == 0: 408 ↛ 409line 408 didn't jump to line 409 because the condition on line 408 was never true
409 continue
411 try:
412 version = PythonVersion.Parse(k)
413 convertedReleasesNode[version] = v
414 except ValueError as ex:
415 print(f"Unsupported version format '{k}' - {ex}")
417 for version, releaseNode in sorted(convertedReleasesNode.items(), key=lambda t: t[0]):
418 if Parts.Postfix in version._parts: 418 ↛ 419line 418 didn't jump to line 419 because the condition on line 418 was never true
419 pass
421 files = [Distribution(file["filename"], file["url"], datetime.fromisoformat(file["upload_time_iso_8601"]), ) for
422 file in releaseNode]
423 lazy = LazyLoaderState.PartiallyLoaded if LazyLoaderState.PartiallyLoaded <= self.__lazy_state__ <= LazyLoaderState.FullyLoaded else LazyLoaderState.Initialized
424 Release(
425 version,
426 files[0]._uploadTime,
427 files,
428 project=self,
429 lazy=lazy
430 )
432 self.SortVersions()
433 self.__lazy_state__ = LazyLoaderState.FullyLoaded
435 def DownloadReleaseDetails(self) -> None:
436 async def ParallelDownloadReleaseDetails():
437 async def routine(session, release: Release):
438 if Parts.Postfix in release._version._parts:
439 pass
441 async with session.get(self._GetPyPIEndpoint()) as response:
442 json = await response.json()
443 response.raise_for_status()
445 release.UpdateDetailsFromPyPIJSON(json)
447 async with ClientSession(base_url=str(self._api), headers={"accept": "application/json"}) as session:
448 tasks = []
449 for release in self._versions.values(): # type: Release
450 tasks.append(routine(session, release))
452 results = await asyncio_gather(*tasks, return_exceptions=True)
453 delList = []
454 for release, result in zip(self.Releases.values(), results):
455 if isinstance(result, Exception):
456 delList.append((release, result))
458 # TODO: raise a warning
459 for release, ex in delList:
460 print(f" Removing {release.Project._name} {release.Version} - {ex}")
461 del self.Releases[release.Version]
463 asyncio_run(ParallelDownloadReleaseDetails())
464 self.__lazy_state__ = LazyLoaderState.PostProcessed
466 def __repr__(self) -> str:
467 return f"Project: {self._name} latest: {self.LatestRelease._version}"
469 def __str__(self) -> str:
470 return f"{self._name}"
473@export
474class PythonPackageIndex(PackageStorage):
475 _url: URL
477 _api: URL
478 _session: Session
480 def __init__(self, name: str, url: Union[str, URL], api: Union[str, URL], graph: "PackageDependencyGraph") -> None:
481 super().__init__(name, graph)
483 if isinstance(url, str): 483 ↛ 485line 483 didn't jump to line 485 because the condition on line 483 was always true
484 url = URL.Parse(url)
485 elif not isinstance(url, URL):
486 ex = TypeError("Parameter 'url' is not of type 'URL'.")
487 ex.add_note(f"Got type '{getFullyQualifiedName(url)}'.")
488 raise ex
490 self._url = url
492 if isinstance(api, str): 492 ↛ 494line 492 didn't jump to line 494 because the condition on line 492 was always true
493 api = URL.Parse(api)
494 elif not isinstance(api, URL):
495 ex = TypeError("Parameter 'api' is not of type 'URL'.")
496 ex.add_note(f"Got type '{getFullyQualifiedName(api)}'.")
497 raise ex
499 self._api = api
501 self._session = Session()
502 self._session.headers["accept"] = "application/json"
504 @readonly
505 def URL(self) -> URL:
506 return self._url
508 @readonly
509 def API(self) -> URL:
510 return self._api
512 @readonly
513 def Projects(self) -> Dict[str, Project]:
514 return self._packages
516 @readonly
517 def ProjectCount(self) -> int:
518 return len(self._packages)
520 def _GetPyPIEndpoint(self, projectName: str) -> str:
521 return f"{self._api}{projectName.lower()}/json"
523 def DownloadProject(self, projectName: str, lazy: LazyLoaderState = LazyLoaderState.PartiallyLoaded) -> Project:
524 project = Project(projectName, "", index=self, lazy=lazy)
526 return project
528 def __repr__(self) -> str:
529 return f"{self._name}"
531 def __str__(self) -> str:
532 return f"{self._name}"
535@export
536class PythonPackageDependencyGraph(PackageDependencyGraph):
537 def __init__(self, name: str) -> None:
538 super().__init__(name)