in metaflow/plugins/pypi/micromamba.py [0:0]
def _call(self, args, env=None):
if env is None:
env = {}
try:
result = (
subprocess.check_output(
[self.bin] + args,
stderr=subprocess.PIPE,
env={
**os.environ,
# prioritize metaflow-specific env vars
**{k: v for k, v in env.items() if v is not None},
**{
"MAMBA_NO_BANNER": "1",
"MAMBA_JSON": "true",
# play with fire! needed for resolving cross-platform
# environments
"CONDA_SAFETY_CHECKS": "disabled",
# "CONDA_UNSATISFIABLE_HINTS_CHECK_DEPTH": "0",
# Support packages on S3
# "CONDA_ALLOW_NON_CHANNEL_URLS": "1",
"MAMBA_USE_LOCKFILES": "false",
},
},
)
.decode()
.strip()
)
if result:
return json.loads(result)
return {}
except subprocess.CalledProcessError as e:
msg = "command '{cmd}' returned error ({code})\n{stderr}"
try:
output = json.loads(e.output)
err = []
v_pkgs = ["__cuda", "__glibc"]
for error in output.get("solver_problems", []):
# raise a specific error message for virtual package related errors
match = next((p for p in v_pkgs if p in error), None)
if match is not None:
vpkg_name = match[2:]
# try to strip version from error msg which are of the format:
# nothing provides <__vpkg> >=2.17,<3.0.a0 needed by <pkg_name>
try:
vpkg_version = error[
len("nothing provides %s " % match) : error.index(
" needed by"
)
]
except ValueError:
vpkg_version = None
raise MicromambaException(
"{msg}\n\n"
"*Please set the environment variable CONDA_OVERRIDE_{var} to a specific version{version} of {name}.*\n\n"
"Here is an example of supplying environment variables through the command line\n"
"CONDA_OVERRIDE_{var}=<{name}-version> python flow.py <args>".format(
msg=msg.format(
cmd=" ".join(e.cmd),
code=e.returncode,
output=e.output.decode(),
stderr=error,
),
var=vpkg_name.upper(),
version=(
"" if not vpkg_version else f" ({vpkg_version})"
),
name=vpkg_name,
)
)
err.append(error)
raise MicromambaException(
msg.format(
cmd=" ".join(e.cmd),
code=e.returncode,
output=e.output.decode(),
stderr="\n".join(err),
)
)
except (TypeError, ValueError):
pass
raise MicromambaException(
msg.format(
cmd=" ".join(e.cmd),
code=e.returncode,
output=e.output.decode(),
stderr=e.stderr.decode(),
)
)