onnxruntime
80 строк · 2.8 Кб
1# Copyright (c) Microsoft Corporation. All rights reserved.
2# Licensed under the MIT License.
3
4import contextlib
5import logging
6import os
7import platform
8import re
9import shutil
10import stat
11import subprocess
12import tempfile
13import urllib.parse
14import urllib.request
15
16AZCOPY_VERSION = "10.4.3"
17
18# See here for instructions on getting stable download links:
19# https://docs.microsoft.com/en-us/azure/storage/common/storage-use-azcopy-v10#obtain-a-static-download-link
20_AZCOPY_DOWNLOAD_URLS = {
21"Linux": "https://azcopyvnext.azureedge.net/release20200501/azcopy_linux_amd64_10.4.3.tar.gz",
22"Darwin": "https://azcopyvnext.azureedge.net/release20200501/azcopy_darwin_amd64_10.4.3.zip",
23"Windows": "https://azcopyvnext.azureedge.net/release20200501/azcopy_windows_amd64_10.4.3.zip",
24}
25
26_log = logging.getLogger("util.get_azcopy")
27
28
29def _check_version(azcopy_path):
30proc = subprocess.run([azcopy_path, "--version"], stdout=subprocess.PIPE, text=True) # noqa: PLW1510
31match = re.search(r"\d+(?:\.\d+)+", proc.stdout)
32
33if not match:
34raise RuntimeError("Failed to determine azcopy version.")
35
36return match.group(0) == AZCOPY_VERSION
37
38
39def _find_azcopy(start_dir):
40for root, _, file_names in os.walk(start_dir):
41for file_name in file_names:
42if file_name == "azcopy" or file_name == "azcopy.exe":
43return os.path.join(root, file_name)
44raise RuntimeError(f"Failed to azcopy in '{start_dir}'.")
45
46
47@contextlib.contextmanager
48def get_azcopy(local_azcopy_path="azcopy"):
49"""
50Creates a context manager that returns a path to a particular version of
51azcopy (specified in AZCOPY_VERSION). Downloads a temporary copy if needed.
52
53:param local_azcopy_path: Path to a local azcopy to try first.
54
55Example usage:
56with get_azcopy() as azcopy_path:
57subprocess.run([azcopy_path, "--version"])
58"""
59with contextlib.ExitStack() as context_stack:
60azcopy_path = shutil.which(local_azcopy_path)
61
62if azcopy_path is None or not _check_version(azcopy_path):
63temp_dir = context_stack.enter_context(tempfile.TemporaryDirectory())
64
65download_url = _AZCOPY_DOWNLOAD_URLS[platform.system()]
66download_basename = urllib.parse.urlsplit(download_url).path.rsplit("/", 1)[-1]
67assert len(download_basename) > 0
68downloaded_path = os.path.join(temp_dir, download_basename)
69
70_log.info(f"Downloading azcopy from '{download_url}'...")
71urllib.request.urlretrieve(download_url, downloaded_path)
72
73extracted_path = os.path.join(temp_dir, "azcopy")
74shutil.unpack_archive(downloaded_path, extracted_path)
75
76azcopy_path = _find_azcopy(extracted_path)
77
78os.chmod(azcopy_path, stat.S_IXUSR)
79
80yield azcopy_path
81