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
|
from typing import List, Any, Optional
import os
import bs4
import urllib3
from ci_tools.variables import str_to_bool
http = urllib3.PoolManager()
# arguments: |
# -c "${{ replace(convertToJson(parameters.CondaArtifacts), '"', '\"') }}"
# -w "$(Build.SourcesDirectory)/conda/conda-recipes"
#
# # Sample configuration yaml. This is converted to json before being passed a blob to the invoking script. (see directly above for conversion)
# - name: uamqp
# common_root: uamqp
# in_batch: true
# conda_py_versions:
# - "38"
# - "39"
# - "310"
# - "311"
# checkout:
# - package: uamqp
# download_uri: https://files.pythonhosted.org/packages/0b/d8/fc24d95e6f6c80851ae6738c78da081cd535c924b02c5a4928b108b9ed42/uamqp-1.6.5.tar.gz
# - name: azure-core
# common_root: azure
# in_batch: ${{ parameters.release_azure_core }}
# channels: "conda-forge"
# checkout:
# - package: azure-core
# version: 1.24.0
# - package: azure-core
# version: 1.24.0
# - name: azure-storage
# common_root: azure
# in_batch: ${{ parameters.release_azure_storage }}
# checkout:
# - package: azure-storage-blob
# checkout_path: sdk/storage
# version: 12.12.0
# - package: azure-storage-queue
# checkout_path: sdk/storage
# version: 12.3.0
# - package: azure-storage-file-share
# checkout_path: sdk/storage
# version: 12.8.0
# - package: azure-storage-file-datalake
# checkout_path: sdk/storage
# version: 12.7.0
def get_package_sdist_url(package: str, version: str) -> str:
url = f"https://pypi.org/pypi/{package}/{version}/json"
response = http.request("GET", url)
if response.status != 200:
raise RuntimeError(f"Failed to fetch metadata for {package}@{version} from PyPI.")
data = response.json()
for file_info in data.get("urls", []):
if file_info.get("packagetype") == "sdist":
return file_info["url"]
raise ValueError(f"Unable to find a source distribution for {package}@{version}.")
class CheckoutConfiguration:
def __init__(self, raw_json: dict):
# we should always have a package name
if "package" in raw_json:
self.package = raw_json["package"]
else:
raise ValueError("A checkout configuration MUST have a package name defined in key 'package'.")
self.checkout_path = raw_json.get("checkout_path", None)
self.version = raw_json.get("version", None)
self.download_uri = raw_json.get("download_uri", None)
if self.version and self.checkout_path is None:
self.download_uri = get_package_sdist_url(self.package, self.version)
if not self.checkout_path and not self.download_uri:
raise ValueError(
"When defining a checkout configuration, one must either have a valid PyPI download url"
" (download_uri) or a path and version in the repo (checkout_path, version)."
)
def __str__(self) -> str:
if self.download_uri:
return f"""- {self.package} downloaded from pypi
{self.download_uri}"""
else:
return f"""- {self.checkout_path}/{self.package} from git @ {self.version}"""
def parse_checkout_config(checkout_configs: List[Any]) -> List[CheckoutConfiguration]:
configs = []
for checkout_config in checkout_configs:
configs.append(CheckoutConfiguration(checkout_config))
return configs
class CondaConfiguration:
def __init__(
self,
name: str,
common_root: str,
in_batch: bool,
checkout: List[CheckoutConfiguration],
created_sdist_path: Optional[str] = None,
service: str = "",
conda_py_versions: List[str] = [],
channels: List[str] = [],
):
self.name: str = name
self.common_root: str = common_root
self.in_batch: bool = in_batch
self.checkout: List[CheckoutConfiguration] = checkout
self.created_sdist_path: Optional[str] = created_sdist_path
self.service: str = service
self.conda_py_versions = conda_py_versions
self.channels = channels
@classmethod
def from_json(cls, raw_json_blob: dict):
name = raw_json_blob.get("name", None)
common_root = raw_json_blob.get("common_root", None)
in_batch = str_to_bool(raw_json_blob["in_batch"])
checkout_config = parse_checkout_config(raw_json_blob.get("checkout", []))
conda_py_versions = raw_json_blob.get("conda_py_versions", [])
service = raw_json_blob.get("service", None)
channels = raw_json_blob.get("channels", [])
# default the service
if any([a.checkout_path for a in checkout_config]) and not service:
valid_checkout_config = next((x for x in checkout_config if x.checkout_path is not None), None)
if valid_checkout_config:
service = valid_checkout_config.checkout_path.split("/")[1].strip()
if not service and name.startswith("azure"):
raise ValueError(
f"Tooling cannot auto-detect targeted service for conda package {name}, nor is there a checkout_path that we can parse the service from. Please correct and retry."
)
return cls(name, common_root, in_batch, checkout_config, None, service, conda_py_versions, channels)
def __str__(self) -> str:
checkout = f"{os.linesep}".join([str(c_config) for c_config in self.checkout])
return f"""====================================
\"{self.name}\" generated from:
{checkout}
====================================
"""
def prepare_local_folder(self) -> None:
pass
|
