-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathlib.py
More file actions
272 lines (225 loc) · 10.8 KB
/
lib.py
File metadata and controls
272 lines (225 loc) · 10.8 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
from __future__ import annotations
import importlib.metadata
import logging
import os
from typing import TYPE_CHECKING
from packaging.utils import canonicalize_name
from packaging.version import Version
from variantlib.constants import VARIANT_ABI_DEPENDENCY_NAMESPACE
from variantlib.models.variant import VariantDescription
from variantlib.models.variant import VariantFeature
from variantlib.models.variant import VariantProperty
from variantlib.resolver.filtering import filter_variants_by_features
from variantlib.resolver.filtering import filter_variants_by_namespaces
from variantlib.resolver.filtering import filter_variants_by_property
from variantlib.resolver.filtering import remove_duplicates
from variantlib.resolver.sorting import sort_variant_properties
from variantlib.resolver.sorting import sort_variants_descriptions
from variantlib.validators.base import validate_type
if TYPE_CHECKING:
from collections.abc import Generator
from variantlib.protocols import VariantFeatureName
from variantlib.protocols import VariantFeatureValue
from variantlib.protocols import VariantNamespace
logger = logging.getLogger(__name__)
def _normalize_package_name(name: str) -> str:
# VALIDATION_FEATURE_NAME_REGEX does not accepts "-"
return canonicalize_name(name).replace("-", "_")
def _generate_version_matches(version: str) -> Generator[str]:
vspec = Version(version)
yield f"{vspec.major}"
yield f"{vspec.major}.{vspec.minor}"
yield f"{vspec.major}.{vspec.minor}.{vspec.micro}"
def filter_variants(
vdescs: list[VariantDescription],
allowed_properties: list[VariantProperty],
forbidden_namespaces: list[str] | None = None,
forbidden_features: list[VariantFeature] | None = None,
forbidden_properties: list[VariantProperty] | None = None,
) -> Generator[VariantDescription]:
"""
Filters out a `list` of `VariantDescription` with the following filters:
- Duplicates removed
- Only allowed `variant properties` kept
# Optionally:
- Forbidden `variant namespaces` removed - if `forbidden_namespaces` is not None
- Forbidden `variant features` removed - if `forbidden_features` is not None
- Forbidden `variant properties` removed - if `forbidden_properties` is not None
:param vdescs: list of `VariantDescription` to filter.
:param allowed_properties: List of allowed `VariantProperty`.
:param forbidden_namespaces: List of forbidden variant namespaces as `str`.
:param forbidden_features: List of forbidden `VariantFeature`.
:param forbidden_properties: List of forbidden `VariantProperty`.
:return: Filtered list of `VariantDescription`.
"""
# Input validation
validate_type(vdescs, list[VariantDescription])
validate_type(allowed_properties, list[VariantProperty])
if forbidden_namespaces is not None:
validate_type(forbidden_namespaces, list[str])
if forbidden_features is not None:
validate_type(forbidden_features, list[VariantFeature])
if forbidden_properties is not None:
validate_type(forbidden_properties, list[VariantProperty])
# Step 1
# Remove duplicates - There should never be any duplicates on the index
# - filename collision (same filename & same hash)
# - hash collision inside `variants.json`
# => Added for safety and to avoid any potential bugs
# (Note: In all fairness, even if it was to happen, it would most
# likely not be a problem given that we just pick the best match)
result = remove_duplicates(vdescs)
# Step 2 [Optional]
# Remove any `VariantDescription` which declares any `VariantProperty` with
# a variant namespace explicitly forbidden by the user.
if forbidden_namespaces is not None:
result = filter_variants_by_namespaces(
vdescs=result,
forbidden_namespaces=forbidden_namespaces,
)
# Step 3 [Optional]
# Remove any `VariantDescription` which declares any `VariantProperty` with
# `namespace :: feature` (aka. `VariantFeature`) explicitly forbidden by the user.
if forbidden_features is not None:
result = filter_variants_by_features(
vdescs=result,
forbidden_features=forbidden_features,
)
# Step 4 [Optional]
# Remove any `VariantDescription` which declare any `VariantProperty`
# `namespace :: feature :: value` unsupported on this platform or explicitly
# forbidden by the user.
if allowed_properties is not None:
result = filter_variants_by_property(
vdescs=result,
allowed_properties=allowed_properties,
forbidden_properties=forbidden_properties,
)
yield from result
def inject_abi_dependency(
supported_vprops: list[VariantProperty],
namespace_priorities: list[VariantNamespace],
) -> None:
"""Inject supported vairants for the abi_dependency namespace"""
# 1. Automatically populate from the current python environment
packages = {
_normalize_package_name(dist.name): dist.version
for dist in importlib.metadata.distributions()
}
# 2. Manually fed from environment variable
# Env Var Format: `VARIANT_ABI_DEPENDENCY=packageA==1.2.3,...,packageZ==7.8.9`
if variant_abi_deps_env := os.environ.get("VARIANT_ABI_DEPENDENCY"):
for pkg_spec in variant_abi_deps_env.split(","):
try:
pkg_name, pkg_version = pkg_spec.split("==", maxsplit=1)
except ValueError:
logger.warning(
"`VARIANT_ABI_DEPENDENCY` received an invalid value "
"`%(pkg_spec)s`. It will be ignored.\n"
"Expected format: `packageA==1.2.3,...,packageZ==7.8.9`.",
{"pkg_spec": pkg_spec},
)
continue
pkg_name = _normalize_package_name(pkg_name)
if (old_version := packages.get(pkg_name)) is not None:
logger.warning(
"`VARIANT_ABI_DEPENDENCY` overrides package version: "
"`%(pkg_name)s` from `%(old_ver)s` to `%(new_ver)s`",
{
"pkg_name": pkg_name,
"old_ver": old_version,
"new_ver": pkg_version,
},
)
packages[pkg_name] = pkg_version
for pkg_name, pkg_version in sorted(packages.items()):
supported_vprops.extend(
VariantProperty(
namespace=VARIANT_ABI_DEPENDENCY_NAMESPACE,
feature=pkg_name,
value=_ver,
)
for _ver in _generate_version_matches(pkg_version)
)
# 3. Adding `VARIANT_ABI_DEPENDENCY_NAMESPACE` at the back of`namespace_priorities`
namespace_priorities.append(VARIANT_ABI_DEPENDENCY_NAMESPACE)
def sort_and_filter_supported_variants(
vdescs: list[VariantDescription],
supported_vprops: list[VariantProperty],
namespace_priorities: list[VariantNamespace],
feature_priorities: dict[VariantNamespace, list[VariantFeatureName]] | None = None,
property_priorities: dict[
VariantNamespace, dict[VariantFeatureName, list[VariantFeatureValue]]
]
| None = None,
forbidden_namespaces: list[VariantNamespace] | None = None,
forbidden_features: list[VariantFeature] | None = None,
forbidden_properties: list[VariantProperty] | None = None,
) -> list[VariantDescription]:
"""
Sort and filter a list of `VariantDescription` objects based on their
`VariantProperty`s.
:param vdescs: List of `VariantDescription` objects.
:param supported_vprops: List of `VariantProperty` objects supported on the platform
:param namespace_priorities: Ordered list of `str` objects.
:param feature_priorities: Ordered list of `VariantFeature` objects.
:param property_priorities: Ordered list of `VariantProperty` objects.
:return: Sorted and filtered list of `VariantDescription` objects.
"""
validate_type(vdescs, list[VariantDescription])
validate_type(supported_vprops, list[VariantProperty])
if namespace_priorities is None:
namespace_priorities = []
# Avoiding modification in place
namespace_priorities = namespace_priorities.copy()
supported_vprops = supported_vprops.copy()
# ======================================================================= #
# ABI DEPENDENCY INJECTION #
# ======================================================================= #
inject_abi_dependency(supported_vprops, namespace_priorities)
# ======================================================================= #
# NULL VARIANT #
# ======================================================================= #
# Adding the `null-variant` to the list - always "compatible"
if (null_variant := VariantDescription()) not in vdescs:
"""Add a null variant description to the list."""
# This is needed to ensure that we always consider the null variant
# to fall back on when no other variants are available.
#
# This can be used to provide a different default build when using
# a variant-enabled installer.
vdescs.append(null_variant)
if supported_vprops is None:
"""No supported properties provided, return no variants."""
return []
# ======================================================================= #
# FILTERING #
# ======================================================================= #
# Step 1: we remove any duplicate, or unsupported `VariantDescription` on
# this platform.
filtered_vdescs = list(
filter_variants(
vdescs=vdescs,
allowed_properties=supported_vprops,
forbidden_namespaces=forbidden_namespaces,
forbidden_features=forbidden_features,
forbidden_properties=forbidden_properties,
)
)
# ======================================================================= #
# SORTING #
# ======================================================================= #
# Step 2: we sort the supported `VariantProperty`s based on their respective
# priority.
sorted_supported_vprops = sort_variant_properties(
vprops=supported_vprops,
property_priorities=property_priorities,
feature_priorities=feature_priorities,
namespace_priorities=namespace_priorities,
)
# Step 3: we sort the `VariantDescription` based on the sorted supported properties
# and their respective priority.
return sort_variants_descriptions(
filtered_vdescs,
property_priorities=sorted_supported_vprops,
)