nix-community/dream2nix
1
1
Fork 0
mirror of https://github.com/nix-community/dream2nix.git synced 2024-12-25 15:33:20 +03:00
Code Issues Projects Releases Wiki Activity

fetch-python-requirements: make more robust...

Browse Source 
...and handle packages which might not even contain
a PKG_INFO. They could still include requirements.txt, otherwise
no dependencies are assumed.
This commit is contained in:
phaer 2023-02-24 14:44:15 +01:00 committed by DavHau
parent 724ed96702
commit 5427ea668e
1 changed files with 41 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

46
v1/nix/modules/drv-parts/mach-nix-xs/python_requirements_from_dists.py
View File

@ -33,18 +33,31 @@ from pathlib import Path
from pkginfo import SDist, Wheel from pkginfo import SDist, Wheel
from packaging.requirements import Requirement from packaging.requirements import Requirement
from packaging.utils import parse_sdist_filename, canonicalize_name from packaging.utils import parse_sdist_filename, parse_wheel_filename, canonicalize_name
def _is_source_dist(pkg_file): def _is_source_dist(pkg_file):
return pkg_file.suffixes[-2:] == ['.tar', '.gz'] return pkg_file.suffixes[-2:] == ['.tar', '.gz']
def _get_name_version(pkg_file):
if _is_source_dist(pkg_file):
name, *_ = parse_sdist_filename(pkg_file.name)
else:
name, *_ = parse_wheel_filename(pkg_file.name)
return canonicalize_name(name)
def get_pkg_info(pkg_file): def get_pkg_info(pkg_file):
try:
if pkg_file.suffix == '.whl': if pkg_file.suffix == '.whl':
return Wheel(str(pkg_file)) return Wheel(str(pkg_file))
elif _is_source_dist(pkg_file): elif _is_source_dist(pkg_file):
return SDist(str(pkg_file)) return SDist(str(pkg_file))
else:
raise NotImplemented(f"Unknown file format: {pkg_file}")
except ValueError:
pass
def _is_required_dependency(requirement): def _is_required_dependency(requirement):
@ -53,21 +66,17 @@ def _is_required_dependency(requirement):
return not requirement.marker or requirement.marker.evaluate({'extra': ""}) return not requirement.marker or requirement.marker.evaluate({'extra': ""})
def get_requirements(pkg_file): def parse_requirements_txt(pkg_file):
requirements = [Requirement(req) for req in info.requires_dist] requirements = []
# For source distributions which do *not* specify requires_dist, if requirements_txt := read_requirements_txt(pkg_file):
# we fallback to parsing requirements.txt
if not requirements and _is_source_dist(pkg_file):
if requirements_txt := get_requirements_txt(pkg_file):
requirements = [ requirements = [
Requirement(req) Requirement(req)
for req in requirements_txt.split("\n") for req in requirements_txt.split("\n")
if req and not req.startswith("#")] if req and not req.startswith("#")]
requirements = filter(_is_required_dependency, requirements)
return requirements return requirements
def get_requirements_txt(source_dist_file): def read_requirements_txt(source_dist_file):
name, version = parse_sdist_filename(source_dist_file.name) name, version = parse_sdist_filename(source_dist_file.name)
with tarfile.open(source_dist_file) as tar: with tarfile.open(source_dist_file) as tar:
try: try:
@ -92,9 +101,20 @@ if __name__ == '__main__':
dependencies = [] dependencies = []
for pkg_file in pkgs_path.iterdir(): for pkg_file in pkgs_path.iterdir():
info = get_pkg_info(pkg_file) info = get_pkg_info(pkg_file)
if not info: name = _get_name_version(pkg_file)
continue if info:
name = canonicalize_name(info.name) requirements = [Requirement(req) for req in info.requires_dist]
dependencies.append((name, [canonicalize_name(req.name) for req in get_requirements(pkg_file)])) else:
requirements = []
# For source distributions which do *not* specify requires_dist,
# we fallback to parsing requirements.txt
if not requirements and _is_source_dist(pkg_file):
requirements = parse_requirements_txt(pkg_file)
requirements = filter(_is_required_dependency, requirements)
dependencies.append((name, [canonicalize_name(req.name) for req in requirements]))
dependencies = sorted(dependencies, key=lambda d: len(d[1])) dependencies = sorted(dependencies, key=lambda d: len(d[1]))
print(json.dumps(dependencies, indent=2)) print(json.dumps(dependencies, indent=2))