/
core.py
2877 lines (2639 loc) · 104 KB
/
core.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
# -*- coding=utf-8 -*-
from __future__ import absolute_import, print_function
import io
import json as simplejson
import logging
import os
import sys
import time
import warnings
import click
import six
import delegator
import dotenv
import pipfile
import vistir
from click_completion import init as init_completion
from . import environments, exceptions, pep508checker, progress
from ._compat import decode_for_output, fix_utf8
from .cmdparse import Script
from .environments import (
PIP_EXISTS_ACTION, PIPENV_CACHE_DIR, PIPENV_COLORBLIND,
PIPENV_DEFAULT_PYTHON_VERSION, PIPENV_DONT_USE_PYENV, PIPENV_HIDE_EMOJIS,
PIPENV_MAX_SUBPROCESS, PIPENV_PYUP_API_KEY, PIPENV_RESOLVE_VCS,
PIPENV_SHELL_FANCY, PIPENV_SKIP_VALIDATION, PIPENV_YES,
SESSION_IS_INTERACTIVE, is_type_checking
)
from .patched import crayons
from .project import Project
from .utils import (
convert_deps_to_pip, create_spinner, download_file,
escape_grouped_arguments, find_python, find_windows_executable,
get_canonical_names, get_source_list, interrupt_handled_subprocess,
is_pinned, is_python_command, is_required_version, is_star, is_valid_url,
parse_indexes, pep423_name, prepare_pip_source_args, proper_case,
python_version, run_command, venv_resolve_deps
)
if is_type_checking():
from typing import Dict, List, Optional, Union, Text
from pipenv.vendor.requirementslib.models.requirements import Requirement
TSourceDict = Dict[Text, Union[Text, bool]]
# Packages that should be ignored later.
BAD_PACKAGES = (
"distribute",
"packaging",
"pip",
"pkg-resources",
"setuptools",
"wheel",
)
FIRST_PACKAGES = ("cython",)
# Are we using the default Python?
USING_DEFAULT_PYTHON = True
if not PIPENV_HIDE_EMOJIS:
now = time.localtime()
# Halloween easter-egg.
if ((now.tm_mon == 10) and (now.tm_mday == 30)) or (
(now.tm_mon == 10) and (now.tm_mday == 31)
):
INSTALL_LABEL = "🎃 "
# Christmas easter-egg.
elif ((now.tm_mon == 12) and (now.tm_mday == 24)) or (
(now.tm_mon == 12) and (now.tm_mday == 25)
):
INSTALL_LABEL = "🎅 "
else:
INSTALL_LABEL = "🐍 "
INSTALL_LABEL2 = crayons.normal("☤ ", bold=True)
STARTING_LABEL = " "
else:
INSTALL_LABEL = " "
INSTALL_LABEL2 = " "
STARTING_LABEL = " "
# Enable shell completion.
init_completion()
# Disable colors, for the color blind and others who do not prefer colors.
if PIPENV_COLORBLIND:
crayons.disable()
def which(command, location=None, allow_global=False):
if not allow_global and location is None:
if project.virtualenv_exists:
location = project.virtualenv_location
else:
location = os.environ.get("VIRTUAL_ENV", None)
if not (location and os.path.exists(location)) and not allow_global:
raise RuntimeError("location not created nor specified")
version_str = "python{0}".format(".".join([str(v) for v in sys.version_info[:2]]))
is_python = command in ("python", os.path.basename(sys.executable), version_str)
if not allow_global:
if os.name == "nt":
p = find_windows_executable(os.path.join(location, "Scripts"), command)
else:
p = os.path.join(location, "bin", command)
else:
if is_python:
p = sys.executable
if not os.path.exists(p):
if is_python:
p = sys.executable or system_which("python")
else:
p = system_which(command)
return p
project = Project(which=which)
def do_clear():
click.echo(crayons.white(fix_utf8("Clearing caches…"), bold=True))
try:
from pip._internal import locations
except ImportError: # pip 9.
from pip import locations
try:
vistir.path.rmtree(PIPENV_CACHE_DIR)
vistir.path.rmtree(locations.USER_CACHE_DIR)
except OSError as e:
# Ignore FileNotFoundError. This is needed for Python 2.7.
import errno
if e.errno == errno.ENOENT:
pass
raise
def load_dot_env():
"""Loads .env file into sys.environ."""
if not environments.PIPENV_DONT_LOAD_ENV:
# If the project doesn't exist yet, check current directory for a .env file
project_directory = project.project_directory or "."
dotenv_file = environments.PIPENV_DOTENV_LOCATION or os.sep.join(
[project_directory, ".env"]
)
if os.path.isfile(dotenv_file):
click.echo(
crayons.normal(fix_utf8("Loading .env environment variables…"), bold=True),
err=True,
)
else:
if environments.PIPENV_DOTENV_LOCATION:
click.echo(
"{0}: file {1}={2} does not exist!!\n{3}".format(
crayons.red("Warning", bold=True),
crayons.normal("PIPENV_DOTENV_LOCATION", bold=True),
crayons.normal(environments.PIPENV_DOTENV_LOCATION, bold=True),
crayons.red("Not loading environment variables.", bold=True),
),
err=True,
)
dotenv.load_dotenv(dotenv_file, override=True)
def add_to_path(p):
"""Adds a given path to the PATH."""
if p not in os.environ["PATH"]:
os.environ["PATH"] = "{0}{1}{2}".format(p, os.pathsep, os.environ["PATH"])
def cleanup_virtualenv(bare=True):
"""Removes the virtualenv directory from the system."""
if not bare:
click.echo(crayons.red("Environment creation aborted."))
try:
# Delete the virtualenv.
vistir.path.rmtree(project.virtualenv_location)
except OSError as e:
click.echo(
"{0} An error occurred while removing {1}!".format(
crayons.red("Error: ", bold=True),
crayons.green(project.virtualenv_location),
),
err=True,
)
click.echo(crayons.blue(e), err=True)
def import_requirements(r=None, dev=False):
from .patched.notpip._vendor import requests as pip_requests
from .vendor.pip_shims.shims import parse_requirements
# Parse requirements.txt file with Pip's parser.
# Pip requires a `PipSession` which is a subclass of requests.Session.
# Since we're not making any network calls, it's initialized to nothing.
if r:
assert os.path.isfile(r)
# Default path, if none is provided.
if r is None:
r = project.requirements_location
with open(r, "r") as f:
contents = f.read()
indexes = []
trusted_hosts = []
# Find and add extra indexes.
for line in contents.split("\n"):
line_indexes, _trusted_hosts, _ = parse_indexes(line.strip())
indexes.extend(line_indexes)
trusted_hosts.extend(_trusted_hosts)
indexes = sorted(set(indexes))
trusted_hosts = sorted(set(trusted_hosts))
reqs = [f for f in parse_requirements(r, session=pip_requests)]
for package in reqs:
if package.name not in BAD_PACKAGES:
if package.link is not None:
package_string = (
"-e {0}".format(package.link)
if package.editable
else str(package.link)
)
project.add_package_to_pipfile(package_string, dev=dev)
else:
project.add_package_to_pipfile(str(package.req), dev=dev)
for index in indexes:
trusted = index in trusted_hosts
project.add_index_to_pipfile(index, verify_ssl=trusted)
project.recase_pipfile()
def ensure_environment():
# Skip this on Windows…
if os.name != "nt":
if "LANG" not in os.environ:
click.echo(
"{0}: the environment variable {1} is not set!"
"\nWe recommend setting this in {2} (or equivalent) for "
"proper expected behavior.".format(
crayons.red("Warning", bold=True),
crayons.normal("LANG", bold=True),
crayons.green("~/.profile"),
),
err=True,
)
def import_from_code(path="."):
from pipreqs import pipreqs
rs = []
try:
for r in pipreqs.get_all_imports(
path, encoding="utf-8", extra_ignore_dirs=[".venv"]
):
if r not in BAD_PACKAGES:
rs.append(r)
pkg_names = pipreqs.get_pkg_names(rs)
return [proper_case(r) for r in pkg_names]
except Exception:
return []
def ensure_pipfile(validate=True, skip_requirements=False, system=False):
"""Creates a Pipfile for the project, if it doesn't exist."""
from .environments import PIPENV_VIRTUALENV
# Assert Pipfile exists.
python = which("python") if not (USING_DEFAULT_PYTHON or system) else None
if project.pipfile_is_empty:
# Show an error message and exit if system is passed and no pipfile exists
if system and not PIPENV_VIRTUALENV:
raise exceptions.PipenvOptionsError(
"--system",
"--system is intended to be used for pre-existing Pipfile "
"installation, not installation of specific packages. Aborting."
)
# If there's a requirements file, but no Pipfile…
if project.requirements_exists and not skip_requirements:
click.echo(
crayons.normal(
fix_utf8("requirements.txt found, instead of Pipfile! Converting…"),
bold=True,
)
)
# Create a Pipfile…
project.create_pipfile(python=python)
with create_spinner("Importing requirements...") as sp:
# Import requirements.txt.
try:
import_requirements()
except Exception:
sp.fail(environments.PIPENV_SPINNER_FAIL_TEXT.format("Failed..."))
else:
sp.ok(environments.PIPENV_SPINNER_OK_TEXT.format("Success!"))
# Warn the user of side-effects.
click.echo(
u"{0}: Your {1} now contains pinned versions, if your {2} did. \n"
"We recommend updating your {1} to specify the {3} version, instead."
"".format(
crayons.red("Warning", bold=True),
crayons.normal("Pipfile", bold=True),
crayons.normal("requirements.txt", bold=True),
crayons.normal('"*"', bold=True),
)
)
else:
click.echo(
crayons.normal(fix_utf8("Creating a Pipfile for this project…"), bold=True),
err=True,
)
# Create the pipfile if it doesn't exist.
project.create_pipfile(python=python)
# Validate the Pipfile's contents.
if validate and project.virtualenv_exists and not PIPENV_SKIP_VALIDATION:
# Ensure that Pipfile is using proper casing.
p = project.parsed_pipfile
changed = project.ensure_proper_casing()
# Write changes out to disk.
if changed:
click.echo(
crayons.normal(u"Fixing package names in Pipfile…", bold=True), err=True
)
project.write_toml(p)
def find_a_system_python(line):
"""Find a Python installation from a given line.
This tries to parse the line in various of ways:
* Looks like an absolute path? Use it directly.
* Looks like a py.exe call? Use py.exe to get the executable.
* Starts with "py" something? Looks like a python command. Try to find it
in PATH, and use it directly.
* Search for "python" and "pythonX.Y" executables in PATH to find a match.
* Nothing fits, return None.
"""
from .vendor.pythonfinder import Finder
finder = Finder(system=False, global_search=True)
if not line:
return next(iter(finder.find_all_python_versions()), None)
# Use the windows finder executable
if (line.startswith("py ") or line.startswith("py.exe ")) and os.name == "nt":
line = line.split(" ", 1)[1].lstrip("-")
python_entry = find_python(finder, line)
return python_entry
def ensure_python(three=None, python=None):
# Support for the PIPENV_PYTHON environment variable.
from .environments import PIPENV_PYTHON
if PIPENV_PYTHON and python is False and three is None:
python = PIPENV_PYTHON
def abort():
click.echo(
"You can specify specific versions of Python with:\n {0}".format(
crayons.red(
"$ pipenv --python {0}".format(
os.sep.join(("path", "to", "python"))
)
)
),
err=True,
)
sys.exit(1)
global USING_DEFAULT_PYTHON
USING_DEFAULT_PYTHON = three is None and not python
# Find out which python is desired.
if not python:
python = convert_three_to_python(three, python)
if not python:
python = project.required_python_version
if not python:
python = PIPENV_DEFAULT_PYTHON_VERSION
path_to_python = find_a_system_python(python)
if environments.is_verbose():
click.echo(u"Using python: {0}".format(python), err=True)
click.echo(u"Path to python: {0}".format(path_to_python), err=True)
if not path_to_python and python is not None:
# We need to install Python.
click.echo(
u"{0}: Python {1} {2}".format(
crayons.red("Warning", bold=True),
crayons.blue(python),
fix_utf8("was not found on your system…"),
),
err=True,
)
# Pyenv is installed
from .vendor.pythonfinder.environment import PYENV_INSTALLED
if not PYENV_INSTALLED:
abort()
else:
if (not PIPENV_DONT_USE_PYENV) and (SESSION_IS_INTERACTIVE or PIPENV_YES):
from .pyenv import Runner, PyenvError
pyenv = Runner("pyenv")
try:
version = pyenv.find_version_to_install(python)
except ValueError:
abort()
except PyenvError as e:
click.echo(fix_utf8("Something went wrong…"))
click.echo(crayons.blue(e.err), err=True)
abort()
s = "{0} {1} {2}".format(
"Would you like us to install",
crayons.green("CPython {0}".format(version)),
"with pyenv?",
)
# Prompt the user to continue…
if not (PIPENV_YES or click.confirm(s, default=True)):
abort()
else:
# Tell the user we're installing Python.
click.echo(
u"{0} {1} {2} {3}{4}".format(
crayons.normal(u"Installing", bold=True),
crayons.green(u"CPython {0}".format(version), bold=True),
crayons.normal(u"with pyenv", bold=True),
crayons.normal(u"(this may take a few minutes)"),
crayons.normal(fix_utf8("…"), bold=True),
)
)
with create_spinner("Installing python...") as sp:
try:
c = pyenv.install(version)
except PyenvError as e:
sp.fail(environments.PIPENV_SPINNER_FAIL_TEXT.format(
"Failed...")
)
click.echo(fix_utf8("Something went wrong…"), err=True)
click.echo(crayons.blue(e.err), err=True)
else:
sp.ok(environments.PIPENV_SPINNER_OK_TEXT.format("Success!"))
# Print the results, in a beautiful blue…
click.echo(crayons.blue(c.out), err=True)
# Clear the pythonfinder caches
from .vendor.pythonfinder import Finder
finder = Finder(system=False, global_search=True)
finder.find_python_version.cache_clear()
finder.find_all_python_versions.cache_clear()
# Find the newly installed Python, hopefully.
version = str(version)
path_to_python = find_a_system_python(version)
try:
assert python_version(path_to_python) == version
except AssertionError:
click.echo(
"{0}: The Python you just installed is not available on your {1}, apparently."
"".format(
crayons.red("Warning", bold=True),
crayons.normal("PATH", bold=True),
),
err=True,
)
sys.exit(1)
return path_to_python
def ensure_virtualenv(three=None, python=None, site_packages=None, pypi_mirror=None):
"""Creates a virtualenv, if one doesn't exist."""
from .environments import PIPENV_USE_SYSTEM
def abort():
sys.exit(1)
global USING_DEFAULT_PYTHON
if not project.virtualenv_exists:
try:
# Ensure environment variables are set properly.
ensure_environment()
# Ensure Python is available.
python = ensure_python(three=three, python=python)
if python is not None and not isinstance(python, six.string_types):
python = python.path.as_posix()
# Create the virtualenv.
# Abort if --system (or running in a virtualenv).
if PIPENV_USE_SYSTEM:
click.echo(
crayons.red(
"You are attempting to re–create a virtualenv that "
"Pipenv did not create. Aborting."
)
)
sys.exit(1)
do_create_virtualenv(
python=python, site_packages=site_packages, pypi_mirror=pypi_mirror
)
except KeyboardInterrupt:
# If interrupted, cleanup the virtualenv.
cleanup_virtualenv(bare=False)
sys.exit(1)
# If --three, --two, or --python were passed…
elif (python) or (three is not None) or (site_packages is not None):
USING_DEFAULT_PYTHON = False
# Ensure python is installed before deleting existing virtual env
python = ensure_python(three=three, python=python)
if python is not None and not isinstance(python, six.string_types):
python = python.path.as_posix()
click.echo(crayons.red("Virtualenv already exists!"), err=True)
# If VIRTUAL_ENV is set, there is a possibility that we are
# going to remove the active virtualenv that the user cares
# about, so confirm first.
if "VIRTUAL_ENV" in os.environ:
if not (
PIPENV_YES or click.confirm("Remove existing virtualenv?", default=True)
):
abort()
click.echo(
crayons.normal(fix_utf8("Removing existing virtualenv…"), bold=True), err=True
)
# Remove the virtualenv.
cleanup_virtualenv(bare=True)
# Call this function again.
ensure_virtualenv(
three=three,
python=python,
site_packages=site_packages,
pypi_mirror=pypi_mirror,
)
def ensure_project(
three=None,
python=None,
validate=True,
system=False,
warn=True,
site_packages=None,
deploy=False,
skip_requirements=False,
pypi_mirror=None,
clear=False,
):
"""Ensures both Pipfile and virtualenv exist for the project."""
from .environments import PIPENV_USE_SYSTEM
# Clear the caches, if appropriate.
if clear:
print("clearing")
sys.exit(1)
# Automatically use an activated virtualenv.
if PIPENV_USE_SYSTEM:
system = True
if not project.pipfile_exists and deploy:
raise exceptions.PipfileNotFound
# Fail if working under /
if not project.name:
click.echo(
"{0}: Pipenv is not intended to work under the root directory, "
"please choose another path.".format(crayons.red("ERROR")),
err=True
)
sys.exit(1)
# Skip virtualenv creation when --system was used.
if not system:
ensure_virtualenv(
three=three,
python=python,
site_packages=site_packages,
pypi_mirror=pypi_mirror,
)
if warn:
# Warn users if they are using the wrong version of Python.
if project.required_python_version:
path_to_python = which("python") or which("py")
if path_to_python and project.required_python_version not in (
python_version(path_to_python) or ""
):
click.echo(
"{0}: Your Pipfile requires {1} {2}, "
"but you are using {3} ({4}).".format(
crayons.red("Warning", bold=True),
crayons.normal("python_version", bold=True),
crayons.blue(project.required_python_version),
crayons.blue(python_version(path_to_python) or "unknown"),
crayons.green(shorten_path(path_to_python)),
),
err=True,
)
click.echo(
" {0} and rebuilding the virtual environment "
"may resolve the issue.".format(crayons.green("$ pipenv --rm")),
err=True,
)
if not deploy:
click.echo(
" {0} will surely fail."
"".format(crayons.red("$ pipenv check")),
err=True,
)
else:
raise exceptions.DeployException
# Ensure the Pipfile exists.
ensure_pipfile(
validate=validate, skip_requirements=skip_requirements, system=system
)
def shorten_path(location, bold=False):
"""Returns a visually shorter representation of a given system path."""
original = location
short = os.sep.join(
[s[0] if len(s) > (len("2long4")) else s for s in location.split(os.sep)]
)
short = short.split(os.sep)
short[-1] = original.split(os.sep)[-1]
if bold:
short[-1] = str(crayons.normal(short[-1], bold=True))
return os.sep.join(short)
# return short
def do_where(virtualenv=False, bare=True):
"""Executes the where functionality."""
if not virtualenv:
if not project.pipfile_exists:
click.echo(
"No Pipfile present at project home. Consider running "
"{0} first to automatically generate a Pipfile for you."
"".format(crayons.green("`pipenv install`")),
err=True,
)
return
location = project.pipfile_location
# Shorten the virtual display of the path to the virtualenv.
if not bare:
location = shorten_path(location)
click.echo(
"Pipfile found at {0}.\n Considering this to be the project home."
"".format(crayons.green(location)),
err=True,
)
else:
click.echo(project.project_directory)
else:
location = project.virtualenv_location
if not bare:
click.echo(
"Virtualenv location: {0}".format(crayons.green(location)), err=True
)
else:
click.echo(location)
def _cleanup_procs(procs, failed_deps_queue, retry=True):
while not procs.empty():
c = procs.get()
if not c.blocking:
c.block()
failed = False
if c.return_code != 0:
failed = True
if "Ignoring" in c.out:
click.echo(crayons.yellow(c.out.strip()))
elif environments.is_verbose():
click.echo(crayons.blue(c.out.strip() or c.err.strip()))
# The Installation failed…
if failed:
if not retry:
# The Installation failed…
# We echo both c.out and c.err because pip returns error details on out.
err = c.err.strip().splitlines() if c.err else []
out = c.out.strip().splitlines() if c.out else []
err_lines = [line for message in [out, err] for line in message]
# Return the subprocess' return code.
raise exceptions.InstallError(c.dep.name, extra=err_lines)
# Save the Failed Dependency for later.
dep = c.dep.copy()
failed_deps_queue.put(dep)
# Alert the user.
click.echo(
"{0} {1}! Will try again.".format(
crayons.red("An error occurred while installing"),
crayons.green(dep.as_line()),
), err=True
)
def batch_install(deps_list, procs, failed_deps_queue,
requirements_dir, no_deps=True, ignore_hashes=False,
allow_global=False, blocking=False, pypi_mirror=None,
retry=True, sequential_deps=None):
from .vendor.requirementslib.models.utils import strip_extras_markers_from_requirement
if sequential_deps is None:
sequential_deps = []
failed = (not retry)
install_deps = not no_deps
if not failed:
label = INSTALL_LABEL if not PIPENV_HIDE_EMOJIS else ""
else:
label = INSTALL_LABEL2
deps_to_install = deps_list[:]
deps_to_install.extend(sequential_deps)
sequential_dep_names = [d.name for d in sequential_deps]
deps_list_bar = progress.bar(
deps_to_install, width=32,
label=label
)
trusted_hosts = []
# Install these because
for dep in deps_list_bar:
extra_indexes = []
if dep.req.req:
dep.req.req = strip_extras_markers_from_requirement(dep.req.req)
if dep.markers:
dep.markers = str(strip_extras_markers_from_requirement(dep.get_markers()))
# Install the module.
is_artifact = False
if dep.is_file_or_url and (dep.is_direct_url or any(
dep.req.uri.endswith(ext) for ext in ["zip", "tar.gz"]
)):
is_artifact = True
elif dep.is_vcs:
is_artifact = True
if not PIPENV_RESOLVE_VCS and is_artifact and not dep.editable:
install_deps = True
no_deps = False
with vistir.contextmanagers.temp_environ():
if not allow_global:
os.environ["PIP_USER"] = vistir.compat.fs_str("0")
if "PYTHONHOME" in os.environ:
del os.environ["PYTHONHOME"]
if "GIT_CONFIG" in os.environ and dep.is_vcs:
del os.environ["GIT_CONFIG"]
c = pip_install(
dep,
ignore_hashes=any([ignore_hashes, dep.editable, dep.is_vcs]),
allow_global=allow_global,
no_deps=not install_deps,
block=any([dep.editable, dep.is_vcs, blocking]),
index=dep.index,
requirements_dir=requirements_dir,
pypi_mirror=pypi_mirror,
trusted_hosts=trusted_hosts,
extra_indexes=extra_indexes,
use_pep517=not failed,
)
c.dep = dep
# if dep.is_vcs or dep.editable:
is_sequential = sequential_deps and dep.name in sequential_dep_names
if is_sequential:
c.block()
procs.put(c)
if procs.full() or procs.qsize() == len(deps_list) or is_sequential:
_cleanup_procs(procs, failed_deps_queue, retry=retry)
def do_install_dependencies(
dev=False,
only=False,
bare=False,
requirements=False,
allow_global=False,
ignore_hashes=False,
skip_lock=False,
concurrent=True,
requirements_dir=None,
pypi_mirror=False,
):
""""
Executes the install functionality.
If requirements is True, simply spits out a requirements format to stdout.
"""
from six.moves import queue
if requirements:
bare = True
# Load the lockfile if it exists, or if only is being used (e.g. lock is being used).
if skip_lock or only or not project.lockfile_exists:
if not bare:
click.echo(
crayons.normal(fix_utf8("Installing dependencies from Pipfile…"), bold=True)
)
# skip_lock should completely bypass the lockfile (broken in 4dac1676)
lockfile = project.get_or_create_lockfile(from_pipfile=True)
else:
lockfile = project.get_or_create_lockfile()
if not bare:
click.echo(
crayons.normal(
fix_utf8("Installing dependencies from Pipfile.lock ({0})…".format(
lockfile["_meta"].get("hash", {}).get("sha256")[-6:]
)),
bold=True,
)
)
# Allow pip to resolve dependencies when in skip-lock mode.
no_deps = not skip_lock # skip_lock true, no_deps False, pip resolves deps
deps_list = list(lockfile.get_requirements(dev=dev, only=requirements))
if requirements:
index_args = prepare_pip_source_args(project.sources)
index_args = " ".join(index_args).replace(" -", "\n-")
deps = [
req.as_line(sources=False, include_hashes=False) for req in deps_list
]
# Output only default dependencies
click.echo(index_args)
click.echo(
"\n".join(sorted(deps))
)
sys.exit(0)
if concurrent:
nprocs = PIPENV_MAX_SUBPROCESS
else:
nprocs = 1
procs = queue.Queue(maxsize=nprocs)
failed_deps_queue = queue.Queue()
if skip_lock:
ignore_hashes = True
editable_or_vcs_deps = [dep for dep in deps_list if (dep.editable or dep.vcs)]
normal_deps = [dep for dep in deps_list if not (dep.editable or dep.vcs)]
install_kwargs = {
"no_deps": no_deps, "ignore_hashes": ignore_hashes, "allow_global": allow_global,
"blocking": not concurrent, "pypi_mirror": pypi_mirror,
"sequential_deps": editable_or_vcs_deps
}
batch_install(
normal_deps, procs, failed_deps_queue, requirements_dir, **install_kwargs
)
if not procs.empty():
_cleanup_procs(procs, failed_deps_queue)
# click.echo(crayons.normal(
# decode_for_output("Installing editable and vcs dependencies…"), bold=True
# ))
# install_kwargs.update({"blocking": True})
# # XXX: All failed and editable/vcs deps should be installed in sequential mode!
# procs = queue.Queue(maxsize=1)
# batch_install(
# editable_or_vcs_deps, procs, failed_deps_queue, requirements_dir,
# **install_kwargs
# )
# Iterate over the hopefully-poorly-packaged dependencies…
if not failed_deps_queue.empty():
click.echo(
crayons.normal(fix_utf8("Installing initially failed dependencies…"), bold=True)
)
retry_list = []
while not failed_deps_queue.empty():
failed_dep = failed_deps_queue.get()
retry_list.append(failed_dep)
install_kwargs.update({"retry": False})
batch_install(
retry_list, procs, failed_deps_queue, requirements_dir, **install_kwargs
)
if not procs.empty():
_cleanup_procs(procs, failed_deps_queue, retry=False)
def convert_three_to_python(three, python):
"""Converts a Three flag into a Python flag, and raises customer warnings
in the process, if needed.
"""
if not python:
if three is False:
return "2"
elif three is True:
return "3"
else:
return python
def do_create_virtualenv(python=None, site_packages=None, pypi_mirror=None):
"""Creates a virtualenv."""
click.echo(
crayons.normal(fix_utf8("Creating a virtualenv for this project…"), bold=True), err=True
)
click.echo(
u"Pipfile: {0}".format(crayons.red(project.pipfile_location, bold=True)),
err=True,
)
# Default to using sys.executable, if Python wasn't provided.
using_string = u"Using"
if not python:
python = sys.executable
using_string = "Using default python from"
click.echo(
u"{0} {1} {3} {2}".format(
crayons.normal(using_string, bold=True),
crayons.red(python, bold=True),
crayons.normal(fix_utf8("to create virtualenv…"), bold=True),
crayons.green("({0})".format(python_version(python))),
),
err=True,
)
cmd = [
vistir.compat.Path(sys.executable).absolute().as_posix(),
"-m",
"virtualenv",
"--prompt=({0}) ".format(project.name),
"--python={0}".format(python),
project.get_location_for_virtualenv(),
]
# Pass site-packages flag to virtualenv, if desired…
if site_packages:
click.echo(
crayons.normal(fix_utf8("Making site-packages available…"), bold=True), err=True
)
cmd.append("--system-site-packages")
if pypi_mirror:
pip_config = {"PIP_INDEX_URL": vistir.misc.fs_str(pypi_mirror)}
else:
pip_config = {}
# Actually create the virtualenv.
error = None
with create_spinner(u"Creating virtual environment...") as sp:
with interrupt_handled_subprocess(cmd, combine_stderr=False, env=pip_config) as c:
click.echo(crayons.blue(u"{0}".format(c.out)), err=True)
if c.returncode != 0:
error = c.err if environments.is_verbose() else exceptions.prettify_exc(c.err)
sp.fail(environments.PIPENV_SPINNER_FAIL_TEXT.format(u"Failed creating virtual environment"))
else:
sp.green.ok(environments.PIPENV_SPINNER_OK_TEXT.format(u"Successfully created virtual environment!"))
if error is not None:
raise exceptions.VirtualenvCreationException(
extra=crayons.red("{0}".format(error))
)
# Associate project directory with the environment.
# This mimics Pew's "setproject".
project_file_name = os.path.join(project.virtualenv_location, ".project")
with open(project_file_name, "w") as f:
f.write(vistir.misc.fs_str(project.project_directory))
from .environment import Environment
sources = project.pipfile_sources
project._environment = Environment(
prefix=project.get_location_for_virtualenv(),
is_venv=True,
sources=sources,
pipfile=project.parsed_pipfile,
project=project
)
project._environment.add_dist("pipenv")
# Say where the virtualenv is.
do_where(virtualenv=True, bare=False)
def parse_download_fname(fname, name):
fname, fextension = os.path.splitext(fname)
if fextension == ".whl":
fname = "-".join(fname.split("-")[:-3])
if fname.endswith(".tar"):
fname, _ = os.path.splitext(fname)
# Substring out package name (plus dash) from file name to get version.
version = fname[len(name) + 1 :]
# Ignore implicit post releases in version number.
if "-" in version and version.split("-")[1].isdigit():
version = version.split("-")[0]
return version
def get_downloads_info(names_map, section):
from .vendor.requirementslib.models.requirements import Requirement
info = []
p = project.parsed_pipfile
for fname in os.listdir(project.download_location):
# Get name from filename mapping.
name = Requirement.from_line(names_map[fname]).name
# Get the version info from the filenames.
version = parse_download_fname(fname, name)
# Get the hash of each file.
cmd = '{0} hash "{1}"'.format(
escape_grouped_arguments(which_pip()),
os.sep.join([project.download_location, fname]),
)
c = delegator.run(cmd)
hash = c.out.split("--hash=")[1].strip()
# Verify we're adding the correct version from Pipfile