-
Notifications
You must be signed in to change notification settings - Fork 104
/
Copy pathconda_lock.py
2062 lines (1879 loc) · 66.7 KB
/
conda_lock.py
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
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
"""
Somewhat hacky solution to create conda lock files.
"""
import datetime
import importlib.util
import itertools
import logging
import os
import pathlib
import posixpath
import re
import sys
import tempfile
from contextlib import contextmanager
from functools import partial
from importlib.metadata import distribution
from types import TracebackType
from typing import (
AbstractSet,
Any,
Dict,
Iterator,
List,
Optional,
Sequence,
Set,
Tuple,
Type,
Union,
)
from urllib.parse import urlsplit
import click
import yaml
from ensureconda.api import ensureconda
from ensureconda.resolve import platform_subdir
from typing_extensions import Literal
from conda_lock.click_helpers import OrderedGroup
from conda_lock.common import (
read_file,
read_json,
relative_path,
temporary_file_with_contents,
warn,
write_file,
)
from conda_lock.conda_solver import solve_conda
from conda_lock.errors import MissingEnvVarError, PlatformValidationError
from conda_lock.export_lock_spec import EditableDependency, render_pixi_toml
from conda_lock.invoke_conda import (
PathLike,
_invoke_conda,
determine_conda_executable,
is_micromamba,
)
from conda_lock.lockfile import (
parse_conda_lock_file,
write_conda_lock_file,
)
from conda_lock.lockfile.v2prelim.models import (
GitMeta,
InputMeta,
LockedDependency,
Lockfile,
LockMeta,
MetadataOption,
TimeMeta,
UpdateSpecification,
)
from conda_lock.lookup import DEFAULT_MAPPING_URL
from conda_lock.models.channel import Channel
from conda_lock.models.lock_spec import LockSpecification
from conda_lock.models.pip_repository import PipRepository
from conda_lock.pypi_solver import solve_pypi
from conda_lock.src_parser import make_lock_spec
from conda_lock.virtual_package import (
FakeRepoData,
default_virtual_package_repodata,
virtual_package_repo_from_specification,
)
logger = logging.getLogger(__name__)
DEFAULT_FILES = [pathlib.Path("environment.yml"), pathlib.Path("environment.yaml")]
# Captures basic auth credentials, if they exists, in the third capture group.
AUTH_PATTERN = re.compile(r"^(# pip .* @ )?(https?:\/\/)(.*:.*@)?(.*)")
# Do not substitute in comments, but do substitute in pip installable packages
# with the pattern: # pip package @ url.
PKG_PATTERN = re.compile(r"(^[^#@].*|^# pip .*)")
# Captures the domain in the third group.
DOMAIN_PATTERN = re.compile(r"^(# pip .* @ )?(https?:\/\/)?([^\/]+)(.*)")
# Captures the platform in the first group.
PLATFORM_PATTERN = re.compile(r"^# platform: (.*)$")
INPUT_HASH_PATTERN = re.compile(r"^# input_hash: (.*)$")
HAVE_MAMBA = (
ensureconda(
mamba=True, micromamba=False, conda=False, conda_exe=False, no_install=True
)
is not None
)
if not (sys.version_info.major >= 3 and sys.version_info.minor >= 6):
print("conda_lock needs to run under python >=3.6")
sys.exit(1)
KIND_EXPLICIT: Literal["explicit"] = "explicit"
KIND_LOCK: Literal["lock"] = "lock"
KIND_ENV: Literal["env"] = "env"
TKindAll = Union[Literal["explicit"], Literal["lock"], Literal["env"]]
TKindRendarable = Union[Literal["explicit"], Literal["lock"], Literal["env"]]
DEFAULT_KINDS: List[Union[Literal["explicit"], Literal["lock"]]] = [
KIND_EXPLICIT,
KIND_LOCK,
]
DEFAULT_LOCKFILE_NAME = "conda-lock.yml"
KIND_FILE_EXT = {
KIND_EXPLICIT: "",
KIND_ENV: ".yml",
KIND_LOCK: "." + DEFAULT_LOCKFILE_NAME,
}
KIND_USE_TEXT = {
KIND_EXPLICIT: "conda create --name YOURENV --file {lockfile}",
KIND_ENV: "conda env create --name YOURENV --file {lockfile}",
KIND_LOCK: "conda-lock install --name YOURENV {lockfile}",
}
_implicit_cuda_message = """
'cudatoolkit' package added implicitly without specifying that cuda packages
should be accepted.
Specify a cuda version via `--with-cuda VERSION` or via virtual packages
to suppress this warning,
or pass `--without-cuda` to explicitly exclude cuda packages.
"""
class UnknownLockfileKind(ValueError):
pass
def _extract_platform(line: str) -> Optional[str]:
search = PLATFORM_PATTERN.search(line)
if search:
return search.group(1)
return None
def _extract_spec_hash(line: str) -> Optional[str]:
search = INPUT_HASH_PATTERN.search(line)
if search:
return search.group(1)
return None
def extract_platform(lockfile: str) -> str:
for line in lockfile.strip().split("\n"):
platform = _extract_platform(line)
if platform:
return platform
raise RuntimeError("Cannot find platform in lockfile.")
def extract_input_hash(lockfile_contents: str) -> Optional[str]:
for line in lockfile_contents.strip().split("\n"):
platform = _extract_spec_hash(line)
if platform:
return platform
return None
def _do_validate_platform(platform: str) -> Tuple[bool, str]:
determined_subdir = platform_subdir()
return platform == determined_subdir, determined_subdir
def do_validate_platform(lockfile: str) -> None:
platform_lockfile = extract_platform(lockfile)
try:
success, platform_sys = _do_validate_platform(platform_lockfile)
except KeyError:
raise RuntimeError(f"Unknown platform type in lockfile '{platform_lockfile}'.")
if not success:
raise PlatformValidationError(
f"Platform in lockfile '{platform_lockfile}' is not compatible with system platform '{platform_sys}'."
)
def do_conda_install(
conda: PathLike,
prefix: "str | None",
name: "str | None",
file: pathlib.Path,
copy: bool,
) -> None:
_conda = partial(_invoke_conda, conda, prefix, name, check_call=True)
kind = "env" if file.name.endswith(".yml") else "explicit"
if kind == "explicit":
with open(file) as explicit_env:
pip_requirements = [
line.split("# pip ")[1]
for line in explicit_env
if line.startswith("# pip ")
]
else:
pip_requirements = []
env_prefix = ["env"] if kind == "env" and not is_micromamba(conda) else []
copy_arg = ["--copy"] if kind != "env" and copy else []
yes_arg = ["--yes"] if kind != "env" else []
_conda(
[
*env_prefix,
"create",
"--quiet",
*copy_arg,
"--file",
str(file),
*yes_arg,
],
)
if not pip_requirements:
return
with temporary_file_with_contents("\n".join(pip_requirements)) as requirements_path:
_conda(["run"], ["pip", "install", "--no-deps", "-r", str(requirements_path)])
def fn_to_dist_name(fn: str) -> str:
if fn.endswith(".conda"):
fn, _, _ = fn.partition(".conda")
elif fn.endswith(".tar.bz2"):
fn, _, _ = fn.partition(".tar.bz2")
else:
raise RuntimeError(f"unexpected file type {fn}", fn)
return fn
def make_lock_files( # noqa: C901
*,
conda: PathLike,
src_files: List[pathlib.Path],
kinds: Sequence[TKindAll],
lockfile_path: Optional[pathlib.Path] = None,
platform_overrides: Optional[Sequence[str]] = None,
channel_overrides: Optional[Sequence[str]] = None,
virtual_package_spec: Optional[pathlib.Path] = None,
update: Optional[Sequence[str]] = None,
include_dev_dependencies: bool = True,
filename_template: Optional[str] = None,
filter_categories: bool = False,
extras: Optional[AbstractSet[str]] = None,
check_input_hash: bool = False,
metadata_choices: AbstractSet[MetadataOption] = frozenset(),
metadata_yamls: Sequence[pathlib.Path] = (),
with_cuda: Optional[str] = None,
strip_auth: bool = False,
mapping_url: str,
) -> None:
"""
Generate a lock file from the src files provided
Parameters
----------
conda :
Path to conda, mamba, or micromamba
src_files :
Files to parse requirements from
kinds :
Lockfile formats to output
lockfile_path :
Path to a conda-lock.yml to create or update
platform_overrides :
Platforms to solve for. Takes precedence over platforms found in src_files.
channel_overrides :
Channels to use. Takes precedence over channels found in src_files.
virtual_package_spec :
Path to a virtual package repository that defines each platform.
update :
Names of dependencies to update to their latest versions, regardless
of whether the constraint in src_files has changed.
include_dev_dependencies :
Include development dependencies in explicit or env output
filename_template :
Format for names of rendered explicit or env files. Must include {platform}.
extras :
Include the given extras in explicit or env output
filter_categories :
Filter out unused categories prior to solving
check_input_hash :
Do not re-solve for each target platform for which specifications are unchanged
metadata_choices:
Set of selected metadata fields to generate for this lockfile.
with_cuda:
The version of cuda requested.
'' means no cuda.
None will pick a default version and warn if cuda packages are installed.
metadata_yamls:
YAML or JSON file(s) containing structured metadata to add to metadata section of the lockfile.
"""
# Compute lock specification
required_categories = {"main"}
if include_dev_dependencies:
required_categories.add("dev")
if extras is not None:
required_categories.update(extras)
lock_spec = make_lock_spec(
src_files=src_files,
channel_overrides=channel_overrides,
platform_overrides=platform_overrides,
required_categories=required_categories if filter_categories else None,
mapping_url=mapping_url,
)
# Load existing lockfile if it exists
original_lock_content: Optional[Lockfile] = None
if lockfile_path is None:
lockfile_path = pathlib.Path(DEFAULT_LOCKFILE_NAME)
if lockfile_path.exists():
try:
original_lock_content = parse_conda_lock_file(lockfile_path)
except (yaml.error.YAMLError, FileNotFoundError):
logger.warning("Failed to parse existing lock. Regenerating from scratch")
original_lock_content = None
else:
original_lock_content = None
# initialize virtual packages
if virtual_package_spec and virtual_package_spec.exists():
virtual_package_repo = virtual_package_repo_from_specification(
virtual_package_spec
)
cuda_specified = True
else:
if with_cuda is None:
cuda_specified = False
with_cuda = "11.4"
else:
cuda_specified = True
virtual_package_repo = default_virtual_package_repodata(cuda_version=with_cuda)
with virtual_package_repo:
platforms_to_lock: List[str] = []
platforms_already_locked: List[str] = []
if original_lock_content is not None:
platforms_already_locked = list(original_lock_content.metadata.platforms)
if update is not None:
# Narrow `update` sequence to list for mypy
update = list(update)
update_spec = UpdateSpecification(
locked=original_lock_content.package, update=update
)
for platform in lock_spec.platforms:
if (
update
or platform not in platforms_already_locked
or not check_input_hash
or lock_spec.content_hash_for_platform(
platform, virtual_package_repo
)
!= original_lock_content.metadata.content_hash[platform]
):
platforms_to_lock.append(platform)
if platform in platforms_already_locked:
platforms_already_locked.remove(platform)
else:
platforms_to_lock = lock_spec.platforms
update_spec = UpdateSpecification()
if platforms_already_locked:
print(
f"Spec hash already locked for {sorted(platforms_already_locked)}. Skipping solve.",
file=sys.stderr,
)
platforms_to_lock = sorted(set(platforms_to_lock))
if not platforms_to_lock:
new_lock_content = original_lock_content
else:
print(f"Locking dependencies for {platforms_to_lock}...", file=sys.stderr)
fresh_lock_content = create_lockfile_from_spec(
conda=conda,
spec=lock_spec,
platforms=platforms_to_lock,
lockfile_path=lockfile_path,
update_spec=update_spec,
metadata_choices=metadata_choices,
metadata_yamls=metadata_yamls,
strip_auth=strip_auth,
virtual_package_repo=virtual_package_repo,
mapping_url=mapping_url,
)
if not original_lock_content:
new_lock_content = fresh_lock_content
else:
# Persist packages from original lockfile for platforms not requested for lock
packages_not_to_lock = [
dep
for dep in original_lock_content.package
if dep.platform not in platforms_to_lock
]
lock_content_to_persist = original_lock_content.model_copy(
deep=True,
update={"package": packages_not_to_lock},
)
new_lock_content = lock_content_to_persist.merge(fresh_lock_content)
if "lock" in kinds:
write_conda_lock_file(
new_lock_content,
lockfile_path,
metadata_choices=metadata_choices,
)
print(
" - Install lock using:",
KIND_USE_TEXT["lock"].format(lockfile=str(lockfile_path)),
file=sys.stderr,
)
# After this point, we're working with `new_lock_content`, never
# `original_lock_content` or `fresh_lock_content`.
assert new_lock_content is not None
# check for implicit inclusion of cudatoolkit
# warn if it was pulled in, but not requested explicitly
if not cuda_specified:
# asking for 'cudatoolkit' is explicit enough
cudatoolkit_requested = any(
pkg.name == "cudatoolkit"
for pkg in itertools.chain(*lock_spec.dependencies.values())
)
if not cudatoolkit_requested:
for package in new_lock_content.package:
if package.name == "cudatoolkit":
logger.warning(_implicit_cuda_message)
break
do_render(
new_lock_content,
kinds=[k for k in kinds if k != "lock"],
include_dev_dependencies=include_dev_dependencies,
filename_template=filename_template,
extras=extras,
check_input_hash=check_input_hash,
)
def do_render(
lockfile: Lockfile,
kinds: Sequence[Union[Literal["env"], Literal["explicit"]]],
include_dev_dependencies: bool = True,
filename_template: Optional[str] = None,
extras: Optional[AbstractSet[str]] = None,
check_input_hash: bool = False,
override_platform: Optional[Sequence[str]] = None,
) -> None:
"""Render the lock content for each platform in lockfile
Parameters
----------
lockfile :
Lock content
kinds :
Lockfile formats to render
include_dev_dependencies :
Include development dependencies in output
filename_template :
Format for the lock file names. Must include {platform}.
extras :
Include the given extras in output
check_input_hash :
Do not re-render if specifications are unchanged
override_platform :
Generate only this subset of the platform files
"""
platforms = lockfile.metadata.platforms
if override_platform is not None and len(override_platform) > 0:
platforms = list(sorted(set(platforms) & set(override_platform)))
if filename_template:
if "{platform}" not in filename_template and len(platforms) > 1:
print(
"{platform} must be in filename template when locking"
f" more than one platform: {', '.join(platforms)}",
file=sys.stderr,
)
sys.exit(1)
for kind, file_ext in KIND_FILE_EXT.items():
if file_ext and filename_template.endswith(file_ext):
print(
f"Filename template must not end with '{file_ext}', as this "
f"is reserved for '{kind}' lock files, in which case it is "
f"automatically added."
)
sys.exit(1)
for plat in platforms:
for kind in kinds:
if filename_template:
context = {
"platform": plat,
"dev-dependencies": str(include_dev_dependencies).lower(),
"input-hash": lockfile.metadata.content_hash,
"version": distribution("conda_lock").version,
"timestamp": datetime.datetime.now(datetime.timezone.utc).strftime(
"%Y%m%dT%H%M%SZ"
),
}
filename = filename_template.format(**context)
else:
filename = f"conda-{plat}.lock"
if pathlib.Path(filename).exists() and check_input_hash:
with open(filename) as f:
previous_hash = extract_input_hash(f.read())
if previous_hash == lockfile.metadata.content_hash.get(plat):
print(
f"Lock content already rendered for {plat}. Skipping render of {filename}.",
file=sys.stderr,
)
continue
print(f"Rendering lockfile(s) for {plat}...", file=sys.stderr)
lockfile_contents = render_lockfile_for_platform(
lockfile=lockfile,
include_dev_dependencies=include_dev_dependencies,
extras=extras,
kind=kind,
platform=plat,
)
filename += KIND_FILE_EXT[kind]
with open(filename, "w") as fo:
fo.write("\n".join(lockfile_contents) + "\n")
print(
f" - Install lock using {'(see warning below)' if kind == 'env' else ''}:",
KIND_USE_TEXT[kind].format(lockfile=filename),
file=sys.stderr,
)
if "env" in kinds:
print(
"\nWARNING: Using environment lock files (*.yml) does NOT guarantee "
"that generated environments will be identical over time, since the "
"dependency resolver is re-run every time and changes in repository "
"metadata or resolver logic may cause variation. Conversely, since "
"the resolver is run every time, the resulting packages ARE "
"guaranteed to be seen by conda as being in a consistent state. This "
"makes them useful when updating existing environments.",
file=sys.stderr,
)
def render_lockfile_for_platform( # noqa: C901
*,
lockfile: Lockfile,
include_dev_dependencies: bool,
extras: Optional[AbstractSet[str]],
kind: Union[Literal["env"], Literal["explicit"]],
platform: str,
suppress_warning_for_pip_and_explicit: bool = False,
) -> List[str]:
"""
Render lock content into a single-platform lockfile that can be installed
with conda.
Parameters
----------
lockfile :
Locked package versions
include_dev_dependencies :
Include development dependencies in output
extras :
Optional dependency groups to include in output
kind :
Lockfile format (explicit or env)
platform :
Target platform
suppress_warning_for_pip_and_explicit :
When rendering internally for `conda-lock install`, we should suppress
the warning about pip dependencies not being supported by all tools.
"""
lockfile_contents = [
"# Generated by conda-lock.",
f"# platform: {platform}",
f"# input_hash: {lockfile.metadata.content_hash.get(platform)}\n",
]
categories_to_install: Set[str] = {
"main",
*(extras or []),
*(["dev"] if include_dev_dependencies else []),
}
conda_deps: List[LockedDependency] = []
pip_deps: List[LockedDependency] = []
# ensure consistent ordering of generated file
# topographic for explicit files and alphabetical otherwise (see gh #554)
if kind == "explicit":
lockfile.toposort_inplace()
else:
lockfile.alphasort_inplace()
lockfile.filter_virtual_packages_inplace()
for p in lockfile.package:
if p.platform == platform and len(p.categories & categories_to_install) > 0:
if p.manager == "pip":
pip_deps.append(p)
elif p.manager == "conda":
# exclude virtual packages
if not p.name.startswith("__"):
conda_deps.append(p)
def format_pip_requirement(
spec: LockedDependency, platform: str, direct: bool = False
) -> str:
if spec.source and spec.source.type == "url":
return f"{spec.name} @ {spec.source.url}"
elif direct:
s = f"{spec.name} @ {spec.url}"
if spec.hash.sha256:
s += f"#sha256={spec.hash.sha256}"
return s
else:
s = f"{spec.name} == {spec.version}"
if spec.hash.sha256:
s += f" --hash=sha256:{spec.hash.sha256}"
return s
def format_conda_requirement(
spec: LockedDependency, platform: str, direct: bool = False
) -> str:
if direct:
# inject the environment variables in here
return posixpath.expandvars(f"{spec.url}#{spec.hash.md5}")
else:
path = pathlib.Path(urlsplit(spec.url).path)
while path.suffix in {".tar", ".bz2", ".gz", ".conda"}:
path = path.with_suffix("")
build_string = path.name.split("-")[-1]
return f"{spec.name}={spec.version}={build_string}"
if kind == "env":
lockfile_contents.extend(
[
"channels:",
*(
f" - {channel.env_replaced_url()}"
for channel in lockfile.metadata.channels
),
"dependencies:",
*(
f" - {format_conda_requirement(dep, platform, direct=False)}"
for dep in conda_deps
),
]
)
lockfile_contents.extend(
[
" - pip:",
*(
f" - {format_pip_requirement(dep, platform, direct=False)}"
for dep in pip_deps
),
]
if pip_deps
else []
)
elif kind == "explicit":
lockfile_contents.append("@EXPLICIT\n")
lockfile_contents.extend(
[format_conda_requirement(dep, platform, direct=True) for dep in conda_deps]
)
def sanitize_lockfile_line(line: str) -> str:
line = line.strip()
if line == "":
return "#"
else:
return line
lockfile_contents = [sanitize_lockfile_line(line) for line in lockfile_contents]
# emit an explicit requirements.txt, prefixed with '# pip '
lockfile_contents.extend(
[
f"# pip {format_pip_requirement(dep, platform, direct=True)}"
for dep in pip_deps
]
)
if len(pip_deps) > 0 and not suppress_warning_for_pip_and_explicit:
logger.warning(
"WARNING: installation of pip dependencies from explicit lockfiles "
"is only supported by the "
"'conda-lock install' and 'micromamba install' commands. Other tools "
"may silently ignore them. For portability, we recommend using the "
"newer unified lockfile format (i.e. removing the --kind=explicit "
"argument."
)
else:
raise ValueError(f"Unrecognised lock kind {kind}.")
logging.debug("lockfile_contents:\n%s\n", lockfile_contents)
return lockfile_contents
def _solve_for_arch(
*,
conda: PathLike,
spec: LockSpecification,
platform: str,
channels: List[Channel],
pip_repositories: List[PipRepository],
virtual_package_repo: FakeRepoData,
update_spec: Optional[UpdateSpecification] = None,
strip_auth: bool = False,
mapping_url: str,
) -> List[LockedDependency]:
"""
Solve specification for a single platform
"""
if update_spec is None:
update_spec = UpdateSpecification()
dependencies = spec.dependencies[platform]
locked = [dep for dep in update_spec.locked if dep.platform == platform]
requested_deps_by_name = {
manager: {dep.name: dep for dep in dependencies if dep.manager == manager}
for manager in ("conda", "pip")
}
locked_deps_by_name = {
manager: {dep.name: dep for dep in locked if dep.manager == manager}
for manager in ("conda", "pip")
}
conda_deps = solve_conda(
conda,
specs=requested_deps_by_name["conda"],
locked=locked_deps_by_name["conda"],
update=update_spec.update,
platform=platform,
channels=channels,
mapping_url=mapping_url,
)
if requested_deps_by_name["pip"]:
if "python" not in conda_deps:
raise ValueError("Got pip specs without Python")
pip_deps = solve_pypi(
pip_specs=requested_deps_by_name["pip"],
use_latest=update_spec.update,
pip_locked={
dep.name: dep for dep in update_spec.locked if dep.manager == "pip"
},
conda_locked={dep.name: dep for dep in conda_deps.values()},
python_version=conda_deps["python"].version,
platform=platform,
platform_virtual_packages=(
virtual_package_repo.all_repodata.get(platform, {"packages": None})[
"packages"
]
if virtual_package_repo
else None
),
pip_repositories=pip_repositories,
allow_pypi_requests=spec.allow_pypi_requests,
strip_auth=strip_auth,
mapping_url=mapping_url,
)
else:
pip_deps = {}
return list(conda_deps.values()) + list(pip_deps.values())
def convert_structured_metadata_yaml(in_path: pathlib.Path) -> Dict[str, Any]:
with in_path.open("r") as infile:
metadata = yaml.safe_load(infile)
return metadata
def update_metadata(to_change: Dict[str, Any], change_source: Dict[str, Any]) -> None:
for key in change_source:
if key in to_change:
logger.warning(
f"Custom metadata field {key} provided twice, overwriting value "
+ f"{to_change[key]} with {change_source[key]}"
)
to_change.update(change_source)
def get_custom_metadata(
metadata_yamls: Sequence[pathlib.Path],
) -> Optional[Dict[str, str]]:
custom_metadata_dict: Dict[str, str] = {}
for yaml_path in metadata_yamls:
new_metadata = convert_structured_metadata_yaml(yaml_path)
update_metadata(custom_metadata_dict, new_metadata)
if custom_metadata_dict:
return custom_metadata_dict
return None
def create_lockfile_from_spec(
*,
conda: PathLike,
spec: LockSpecification,
platforms: Optional[List[str]] = None,
lockfile_path: pathlib.Path,
update_spec: Optional[UpdateSpecification] = None,
metadata_choices: AbstractSet[MetadataOption] = frozenset(),
metadata_yamls: Sequence[pathlib.Path] = (),
strip_auth: bool = False,
virtual_package_repo: FakeRepoData,
mapping_url: str,
) -> Lockfile:
"""
Solve or update specification
"""
if platforms is None:
platforms = []
locked: Dict[Tuple[str, str, str], LockedDependency] = {}
for platform in platforms or spec.platforms:
deps = _solve_for_arch(
conda=conda,
spec=spec,
platform=platform,
channels=[*spec.channels, virtual_package_repo.channel],
pip_repositories=spec.pip_repositories,
virtual_package_repo=virtual_package_repo,
update_spec=update_spec,
strip_auth=strip_auth,
mapping_url=mapping_url,
)
for dep in deps:
locked[(dep.manager, dep.name, dep.platform)] = dep
meta_sources: Dict[str, pathlib.Path] = {}
for source in spec.sources:
try:
path = relative_path(lockfile_path.parent, source)
except ValueError as e:
if "Paths don't have the same drive" not in str(e):
raise e
path = str(source.resolve())
meta_sources[path] = source
if MetadataOption.TimeStamp in metadata_choices:
time_metadata = TimeMeta.create()
else:
time_metadata = None
if metadata_choices & {
MetadataOption.GitUserEmail,
MetadataOption.GitUserName,
MetadataOption.GitSha,
}:
if not importlib.util.find_spec("git"):
raise RuntimeError(
"The GitPython package is required to read Git metadata."
)
git_metadata = GitMeta.create(
metadata_choices=metadata_choices,
src_files=spec.sources,
)
else:
git_metadata = None
if metadata_choices & {MetadataOption.InputSha, MetadataOption.InputMd5}:
inputs_metadata: Optional[Dict[str, InputMeta]] = {
meta_src: InputMeta.create(
metadata_choices=metadata_choices, src_file=src_file
)
for meta_src, src_file in meta_sources.items()
}
else:
inputs_metadata = None
custom_metadata = get_custom_metadata(metadata_yamls=metadata_yamls)
content_hash = spec.content_hash(virtual_package_repo)
return Lockfile(
package=[locked[k] for k in locked],
metadata=LockMeta(
content_hash=content_hash,
channels=[c for c in spec.channels],
platforms=spec.platforms,
sources=list(meta_sources.keys()),
git_metadata=git_metadata,
time_metadata=time_metadata,
inputs_metadata=inputs_metadata,
custom_metadata=custom_metadata,
),
)
def _add_auth_to_line(line: str, auth: Dict[str, str]) -> str:
matching_auths = [a for a in auth if a in line]
if not matching_auths:
return line
# If we have multiple matching auths, we choose the longest one.
matching_auth = max(matching_auths, key=len)
replacement = f"{auth[matching_auth]}@{matching_auth}"
return line.replace(matching_auth, replacement)
def _add_auth_to_lockfile(lockfile: str, auth: Dict[str, str]) -> str:
lockfile_with_auth = "\n".join(
_add_auth_to_line(line, auth) if PKG_PATTERN.match(line) else line
for line in lockfile.strip().split("\n")
)
if lockfile.endswith("\n"):
return lockfile_with_auth + "\n"
return lockfile_with_auth
@contextmanager
def _add_auth(lockfile: str, auth: Dict[str, str]) -> Iterator[pathlib.Path]:
lockfile_with_auth = _add_auth_to_lockfile(lockfile, auth)
with temporary_file_with_contents(lockfile_with_auth) as path:
yield path
def _strip_auth_from_line(line: str) -> str:
return AUTH_PATTERN.sub(r"\1\2\4", line)
def _extract_domain(line: str) -> str:
return DOMAIN_PATTERN.sub(r"\3", line)
def _strip_auth_from_lockfile(lockfile: str) -> str:
lockfile_lines = lockfile.strip().split("\n")
stripped_lockfile_lines = tuple(
_strip_auth_from_line(line) if PKG_PATTERN.match(line) else line
for line in lockfile_lines
)
stripped_domains = sorted(
{
_extract_domain(stripped_line)
for line, stripped_line in zip(lockfile_lines, stripped_lockfile_lines)
if line != stripped_line
}
)
stripped_lockfile = "\n".join(stripped_lockfile_lines)
if lockfile.endswith("\n"):
stripped_lockfile += "\n"
if stripped_domains:
stripped_domains_doc = "\n".join(f"# - {domain}" for domain in stripped_domains)
return f"# The following domains require authentication:\n{stripped_domains_doc}\n{stripped_lockfile}"
return stripped_lockfile
@contextmanager
def _render_lockfile_for_install(
filename: pathlib.Path,
include_dev_dependencies: bool = True,
extras: Optional[AbstractSet[str]] = None,
force_platform: Optional[str] = None,
) -> Iterator[pathlib.Path]:
"""
Render lock content into a temporary, explicit lockfile for the current platform
Parameters
----------
filename :
Path to conda-lock.yml
include_dev_dependencies :
Include development dependencies in output
extras :
Optional dependency groups to include in output