Skip to content

morgan.ini: detailed selection of *.whl files #65

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
shamilbi wants to merge 2 commits into
base: main
Choose a base branch
from
Open

morgan.ini: detailed selection of *.whl files #65

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
0580c8b
morgan.ini: +whl.tag.*, -platform_tag, -sys_platform, -platform_machine
shamilbi Nov 26, 2025
63abf42
fix: morgan.ini, -parse_interpreter
shamilbi Nov 26, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
95 changes: 25 additions & 70 deletions morgan/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
import argparse
import configparser
import hashlib
import inspect
import json
import os
import os.path
Expand All @@ -10,7 +11,7 @@
import urllib.parse
import urllib.request
import zipfile
from typing import Dict, Iterable, Tuple
from typing import Dict, Iterable

import packaging.requirements
import packaging.specifiers
Expand All @@ -26,7 +27,7 @@
PREFERRED_HASH_ALG = "sha256"


class Mirrorer:
class Mirrorer: # pylint: disable=too-few-public-methods
"""
Mirrorer is a class that implements the mirroring capabilities of Morgan.
A class is used to maintain state, as the mirrorer needs to keep track of
Expand All @@ -48,7 +49,7 @@ def __init__(self, args: argparse.Namespace):
self.config.read(args.config)
self.envs = {}
self._supported_pyversions = []
self._supported_platforms = []
self.whl_tags = [] # list[(interpreter, abi, platform)]
for key in self.config:
m = re.fullmatch(r"env\.(.+)", key)
if m:
Expand All @@ -62,13 +63,15 @@ def __init__(self, args: argparse.Namespace):
self._supported_pyversions.append(env["python_full_version"])
else:
self._supported_pyversions.append(env["python_version"])
if "platform_tag" in env:
self._supported_platforms.append(re.compile(env["platform_tag"]))
self._supported_platforms.append(
re.compile(
r".*" + env["sys_platform"] + r".*" + env["platform_machine"]
)
)
# whl.tag.*
l = [] # list[re.Pattern]
for i in ('interpreter', 'abi', 'platform'):
t = env.get(f'whl.tag.{i}', '').strip()
if t:
l.append(re.compile(t))
else:
l.append(None)
self.whl_tags.append(l)

self._processed_pkgs = Cache()

Expand Down Expand Up @@ -116,8 +119,6 @@ def copy_server(self):
with open(serverpath, "rb") as inp, open(outpath, "wb") as out:
out.write(inp.read())
else:
import inspect

with open(outpath, "w") as out:
out.write(inspect.getsource(server))

Expand Down Expand Up @@ -172,7 +173,7 @@ def _mirror(
# for any of our environments and don't return an error
return None

if len(files) == 0:
if not files:
raise Exception(f"No files match requirement {requirement}")

# download all files
Expand Down Expand Up @@ -255,16 +256,16 @@ def _filter_files(
)
)

if len(files) == 0:
print(f"Skipping {requirement}, no version matches requirement")
if not files:
print(f"\tSkipping {requirement}, no version matches requirement")
return None

# Now we only have files that satisfy the requirement, and we need to
# filter out files that do not match our environments.
files = list(filter(lambda file: self._matches_environments(file), files))
files = list(filter(self._matches_environments, files))

if len(files) == 0:
print(f"Skipping {requirement}, no file matches environments")
if not files:
print(f"\tSkipping {requirement}, no file matches environments")
return None

# Only keep files from the latest version in case the package is a dependency of another
Expand All @@ -276,7 +277,8 @@ def _filter_files(
return files

def _matches_environments(self, fileinfo: dict) -> bool:
if req := fileinfo.get("requires-python", None):
req = fileinfo.get("requires-python", None)
if req:
# The Python versions in all of our environments must be supported
# by this file in order to match.
# Some packages specify their required Python versions with a simple
Expand All @@ -301,37 +303,11 @@ def _matches_environments(self, fileinfo: dict) -> bool:
return False

if fileinfo.get("tags", None):
# At least one of the tags must match ALL of our environments
for tag in fileinfo["tags"]:
(intrp_name, intrp_ver) = parse_interpreter(tag.interpreter)
if intrp_name not in ("py", "cp"):
continue

intrp_set = packaging.specifiers.SpecifierSet(r'>=' + intrp_ver)
# As an example, cp38 seems to indicate CPython 3.8+, so we
# check if the version matches any of the supported Pythons, and
# only skip it if it does not match any.
intrp_ver_matched = any(
map(
lambda supported_python: intrp_set.contains(supported_python),
self._supported_pyversions,
)
)

if (
intrp_ver
and intrp_ver != "3"
and not intrp_ver_matched
):
continue

if tag.platform == "any":
return True
else:
for platformre in self._supported_platforms:
if platformre.fullmatch(tag.platform):
# tag matched, accept this file
return True
for t in self.whl_tags:
t2 = zip(t[:3], [tag.interpreter, tag.abi, tag.platform])
if all(i.match(j) for i, j in t2 if i):
return True

# none of the tags matched, reject this file
return False
Expand Down Expand Up @@ -455,27 +431,6 @@ def _extract_metadata(
return md


def parse_interpreter(inp: str) -> Tuple[str, str]:
"""
Parse interpreter tags in the name of a binary wheel file. Returns a tuple
of interpreter name and optional version, which will either be <major> or
<major>.<minor>.
"""

m = re.fullmatch(r"^([^\d]+)(?:(\d)(?:[._])?(\d+)?)$", inp)
if m is None:
return (inp, None)

intr = m.group(1)
version = None
if m.lastindex > 1:
version = m.group(2)
if m.lastindex > 2:
version = "{}.{}".format(version, m.group(3))

return (intr, version)


def parse_requirement(req_string: str) -> packaging.requirements.Requirement:
"""
Parse a requirement string into a packaging.requirements.Requirement object.
Expand Down
41 changes: 7 additions & 34 deletions tests/test_init.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,35 +6,7 @@
import packaging.requirements
import pytest

from morgan import PYPI_ADDRESS, Mirrorer, parse_interpreter, parse_requirement, server


class TestParseInterpreter:
@pytest.mark.parametrize(
"interpreter_string, expected_name, expected_version",
[
("cp38", "cp", "3.8"),
("cp3", "cp", "3"),
("cp310", "cp", "3.10"),
("cp3_10", "cp", "3.10"),
("py38", "py", "3.8"),
("something_strange", "something_strange", None),
],
ids=[
"typical_cpython",
"cpython_no_minor_version",
"cpython_two_digit_minor",
"cpython_with_underscore",
"generic_python",
"unrecognized_format",
],
)
def test_parse_interpreter_components(
self, interpreter_string, expected_name, expected_version
):
name, version = parse_interpreter(interpreter_string)
assert name == expected_name
assert version == expected_version
from morgan import PYPI_ADDRESS, Mirrorer, parse_requirement, server


class TestParseRequirement:
Expand Down Expand Up @@ -71,9 +43,9 @@ def temp_index_path(self, tmpdir):
"""
[env.test_env]
python_version = 3.10
sys_platform = linux
platform_machine = x86_64
platform_tag = manylinux
whl.tag.interpreter = (cp310|py3)$
whl.tag.abi = (cp310|cp310t|abi3|none)$
whl.tag.platform = (manylinux.*_x86_64|any)$

[requirements]
requests = >=2.0.0
Expand All @@ -95,8 +67,9 @@ def test_mirrorer_initialization(self, temp_index_path):
assert mirrorer.index_url == "https://pypi.org/simple/"
assert "test_env" in mirrorer.envs
assert mirrorer.envs["test_env"]["python_version"] == "3.10"
assert mirrorer.envs["test_env"]["sys_platform"] == "linux"
assert mirrorer.envs["test_env"]["platform_machine"] == "x86_64"
assert mirrorer.envs["test_env"]["whl.tag.interpreter"] == "(cp310|py3)$"
assert mirrorer.envs["test_env"]["whl.tag.abi"] == "(cp310|cp310t|abi3|none)$"
assert mirrorer.envs["test_env"]["whl.tag.platform"] == "(manylinux.*_x86_64|any)$"
assert not mirrorer.mirror_all_versions

def test_server_file_copying(self, temp_index_path):
Expand Down