path.py
407 lines
| 12.1 KiB
| text/x-python
|
PythonLexer
|
r2498 | # encoding: utf-8 | ||
""" | ||||
Utilities for path handling. | ||||
""" | ||||
|
r16486 | # Copyright (c) IPython Development Team. | ||
# Distributed under the terms of the Modified BSD License. | ||||
|
r2498 | |||
import os | ||||
import sys | ||||
|
r11647 | import errno | ||
import shutil | ||||
import random | ||||
|
r8066 | import glob | ||
|
r28664 | import warnings | ||
|
r2498 | |||
|
r2908 | from IPython.utils.process import system | ||
|
r21039 | |||
|
r2498 | #----------------------------------------------------------------------------- | ||
# Code | ||||
#----------------------------------------------------------------------------- | ||||
|
r3808 | fs_encoding = sys.getfilesystemencoding() | ||
|
r4473 | def _writable_dir(path): | ||
"""Whether `path` is a directory, to which the user has write access.""" | ||||
return os.path.isdir(path) and os.access(path, os.W_OK) | ||||
|
r2498 | if sys.platform == 'win32': | ||
def _get_long_path_name(path): | ||||
"""Get a long path name (expand ~) on Windows using ctypes. | ||||
Examples | ||||
-------- | ||||
|
r26934 | >>> get_long_path_name('c:\\\\docume~1') | ||
|
r23213 | 'c:\\\\Documents and Settings' | ||
|
r2498 | |||
""" | ||||
try: | ||||
import ctypes | ||||
|
r25833 | except ImportError as e: | ||
raise ImportError('you need to have ctypes installed for this to work') from e | ||||
|
r2498 | _GetLongPathName = ctypes.windll.kernel32.GetLongPathNameW | ||
_GetLongPathName.argtypes = [ctypes.c_wchar_p, ctypes.c_wchar_p, | ||||
ctypes.c_uint ] | ||||
buf = ctypes.create_unicode_buffer(260) | ||||
rv = _GetLongPathName(path, buf, 260) | ||||
if rv == 0 or rv > 260: | ||||
return path | ||||
else: | ||||
return buf.value | ||||
|
r19739 | else: | ||
def _get_long_path_name(path): | ||||
"""Dummy no-op.""" | ||||
return path | ||||
|
r2498 | |||
def get_long_path_name(path): | ||||
"""Expand a path into its long form. | ||||
On Windows this expands any ~ in the paths. On other platforms, it is | ||||
a null operation. | ||||
""" | ||||
return _get_long_path_name(path) | ||||
|
r13173 | def compress_user(path): | ||
"""Reverse of :func:`os.path.expanduser` | ||||
|
r22438 | """ | ||
|
r13173 | home = os.path.expanduser('~') | ||
if path.startswith(home): | ||||
path = "~" + path[len(home):] | ||||
return path | ||||
|
r4696 | |||
|
r27272 | def get_py_filename(name): | ||
|
r2498 | """Return a valid python filename in the current directory. | ||
If the given name is not a file, it adds '.py' and searches again. | ||||
|
r4688 | Raises IOError with an informative message if the file isn't found. | ||
""" | ||||
|
r2498 | |||
name = os.path.expanduser(name) | ||||
if os.path.isfile(name): | ||||
return name | ||||
|
r27744 | if not name.endswith(".py"): | ||
py_name = name + ".py" | ||||
|
r27743 | if os.path.isfile(py_name): | ||
return py_name | ||||
|
r27744 | raise IOError("File `%r` not found." % name) | ||
|
r2498 | |||
|
r26419 | def filefind(filename: str, path_dirs=None) -> str: | ||
|
r2498 | """Find a file by looking through a sequence of paths. | ||
This iterates through a sequence of paths looking for a file and returns | ||||
|
r24084 | the full, absolute path of the first occurrence of the file. If no set of | ||
|
r2498 | path dirs is given, the filename is tested as is, after running through | ||
:func:`expandvars` and :func:`expanduser`. Thus a simple call:: | ||||
filefind('myfile.txt') | ||||
will find the file in the current working dir, but:: | ||||
filefind('~/myfile.txt') | ||||
Will find the file in the users home directory. This function does not | ||||
automatically try any paths, such as the cwd or the user's home directory. | ||||
|
r4872 | |||
|
r2498 | Parameters | ||
---------- | ||||
filename : str | ||||
The filename to look for. | ||||
path_dirs : str, None or sequence of str | ||||
The sequence of paths to look for the file in. If None, the filename | ||||
need to be absolute or be in the cwd. If a string, the string is | ||||
put into a sequence and the searched. If a sequence, walk through | ||||
each element and join with ``filename``, calling :func:`expandvars` | ||||
and :func:`expanduser` before testing for existence. | ||||
|
r4872 | |||
|
r2498 | Returns | ||
------- | ||||
|
r26419 | path : str | ||
returns absolute path to file. | ||||
Raises | ||||
------ | ||||
IOError | ||||
|
r2498 | """ | ||
|
r4872 | |||
|
r2498 | # If paths are quoted, abspath gets confused, strip them... | ||
filename = filename.strip('"').strip("'") | ||||
# If the input is an absolute path, just check it exists | ||||
if os.path.isabs(filename) and os.path.isfile(filename): | ||||
return filename | ||||
|
r4872 | |||
|
r2498 | if path_dirs is None: | ||
path_dirs = ("",) | ||||
|
r23037 | elif isinstance(path_dirs, str): | ||
|
r2498 | path_dirs = (path_dirs,) | ||
|
r4872 | |||
|
r2498 | for path in path_dirs: | ||
|
r23045 | if path == '.': path = os.getcwd() | ||
|
r2498 | testname = expand_path(os.path.join(path, filename)) | ||
if os.path.isfile(testname): | ||||
return os.path.abspath(testname) | ||||
|
r4872 | |||
raise IOError("File %r does not exist in any of the search paths: %r" % | ||||
|
r2498 | (filename, path_dirs) ) | ||
class HomeDirError(Exception): | ||||
pass | ||||
|
r25290 | def get_home_dir(require_writable=False) -> str: | ||
|
r5384 | """Return the 'home' directory, as a unicode string. | ||
|
r2498 | |||
|
r11499 | Uses os.path.expanduser('~'), and checks for writability. | ||
|
r21517 | |||
|
r5383 | See stdlib docs for how this is determined. | ||
|
r25223 | For Python <3.8, $HOME is first priority on *ALL* platforms. | ||
For Python >=3.8 on Windows, %HOME% is no longer considered. | ||||
|
r21517 | |||
|
r5384 | Parameters | ||
---------- | ||||
require_writable : bool [default: False] | ||||
if True: | ||||
guarantees the return value is a writable directory, otherwise | ||||
raises HomeDirError | ||||
if False: | ||||
The path is resolved, but it is not guaranteed to exist or be writable. | ||||
|
r2498 | """ | ||
|
r5383 | homedir = os.path.expanduser('~') | ||
|
r6112 | # Next line will make things work even when /home/ is a symlink to | ||
# /usr/home as it is on FreeBSD, for example | ||||
homedir = os.path.realpath(homedir) | ||||
|
r21517 | |||
|
r5385 | if not _writable_dir(homedir) and os.name == 'nt': | ||
# expanduser failed, use the registry to get the 'My Documents' folder. | ||||
|
r2498 | try: | ||
|
r25290 | import winreg as wreg | ||
|
r25293 | with wreg.OpenKey( | ||
|
r2498 | wreg.HKEY_CURRENT_USER, | ||
|
r24451 | r"Software\Microsoft\Windows\CurrentVersion\Explorer\Shell Folders" | ||
|
r25293 | ) as key: | ||
homedir = wreg.QueryValueEx(key,'Personal')[0] | ||||
|
r2498 | except: | ||
pass | ||||
|
r21517 | |||
|
r5384 | if (not require_writable) or _writable_dir(homedir): | ||
|
r26875 | assert isinstance(homedir, str), "Homedir should be unicode not bytes" | ||
|
r25290 | return homedir | ||
|
r2498 | else: | ||
|
r5385 | raise HomeDirError('%s is not a writable dir, ' | ||
'set $HOME environment variable to override' % homedir) | ||||
|
r2498 | |||
|
r3347 | def get_xdg_dir(): | ||
"""Return the XDG_CONFIG_HOME, if it is defined and exists, else None. | ||||
|
r4872 | |||
|
r7086 | This is only for non-OS X posix (Linux,Unix,etc.) systems. | ||
|
r3347 | """ | ||
env = os.environ | ||||
|
r4872 | |||
|
r26954 | if os.name == "posix": | ||
|
r7086 | # Linux, Unix, AIX, etc. | ||
|
r5384 | # use ~/.config if empty OR not set | ||
|
r3347 | xdg = env.get("XDG_CONFIG_HOME", None) or os.path.join(get_home_dir(), '.config') | ||
|
r4474 | if xdg and _writable_dir(xdg): | ||
|
r25353 | assert isinstance(xdg, str) | ||
return xdg | ||||
|
r4872 | |||
|
r3347 | return None | ||
|
r4872 | |||
|
r2498 | |||
|
r10230 | def get_xdg_cache_dir(): | ||
"""Return the XDG_CACHE_HOME, if it is defined and exists, else None. | ||||
This is only for non-OS X posix (Linux,Unix,etc.) systems. | ||||
""" | ||||
env = os.environ | ||||
|
r26954 | if os.name == "posix": | ||
|
r10230 | # Linux, Unix, AIX, etc. | ||
# use ~/.cache if empty OR not set | ||||
xdg = env.get("XDG_CACHE_HOME", None) or os.path.join(get_home_dir(), '.cache') | ||||
if xdg and _writable_dir(xdg): | ||||
|
r25353 | assert isinstance(xdg, str) | ||
return xdg | ||||
|
r10230 | |||
return None | ||||
|
r2498 | def expand_path(s): | ||
"""Expand $VARS and ~names in a string, like a shell | ||||
:Examples: | ||||
|
r4872 | |||
|
r2498 | In [2]: os.environ['FOO']='test' | ||
In [3]: expand_path('variable FOO is $FOO') | ||||
Out[3]: 'variable FOO is test' | ||||
""" | ||||
# This is a pretty subtle hack. When expand user is given a UNC path | ||||
# on Windows (\\server\share$\%username%), os.path.expandvars, removes | ||||
# the $ to get (\\server\share\%username%). I think it considered $ | ||||
# alone an empty var. But, we need the $ to remains there (it indicates | ||||
# a hidden share). | ||||
if os.name=='nt': | ||||
s = s.replace('$\\', 'IPYTHON_TEMP') | ||||
s = os.path.expandvars(os.path.expanduser(s)) | ||||
if os.name=='nt': | ||||
s = s.replace('IPYTHON_TEMP', '$\\') | ||||
return s | ||||
|
r8119 | def unescape_glob(string): | ||
"""Unescape glob pattern in `string`.""" | ||||
|
r8122 | def unescape(s): | ||
for pattern in '*[]!?': | ||||
s = s.replace(r'\{0}'.format(pattern), pattern) | ||||
return s | ||||
return '\\'.join(map(unescape, string.split('\\\\'))) | ||||
|
r8119 | |||
|
r8067 | def shellglob(args): | ||
|
r8014 | """ | ||
Do glob expansion for each element in `args` and return a flattened list. | ||||
Unmatched glob pattern will remain as-is in the returned list. | ||||
""" | ||||
expanded = [] | ||||
|
r8646 | # Do not unescape backslash in Windows as it is interpreted as | ||
# path separator: | ||||
unescape = unescape_glob if sys.platform != 'win32' else lambda x: x | ||||
|
r8014 | for a in args: | ||
|
r8646 | expanded.extend(glob.glob(a) or [unescape(a)]) | ||
|
r8014 | return expanded | ||
|
r2498 | def target_outdated(target,deps): | ||
"""Determine whether a target is out of date. | ||||
target_outdated(target,deps) -> 1/0 | ||||
deps: list of filenames which MUST exist. | ||||
target: single filename which may or may not exist. | ||||
If target doesn't exist or is older than any file listed in deps, return | ||||
true, otherwise return false. | ||||
|
r28657 | |||
.. deprecated:: 8.22 | ||||
|
r2498 | """ | ||
|
r28664 | warnings.warn( | ||
"`target_outdated` is deprecated since IPython 8.22 and will be removed in future versions", | ||||
DeprecationWarning, | ||||
stacklevel=2, | ||||
) | ||||
|
r2498 | try: | ||
target_time = os.path.getmtime(target) | ||||
except os.error: | ||||
return 1 | ||||
for dep in deps: | ||||
dep_time = os.path.getmtime(dep) | ||||
if dep_time > target_time: | ||||
|
r28756 | # print("For target",target,"Dep failed:",dep) # dbg | ||
# print("times (dep,tar):",dep_time,target_time) # dbg | ||||
|
r2498 | return 1 | ||
return 0 | ||||
def target_update(target,deps,cmd): | ||||
"""Update a target with a given command given a list of dependencies. | ||||
target_update(target,deps,cmd) -> runs cmd if target is outdated. | ||||
This is just a wrapper around target_outdated() which calls the given | ||||
|
r28657 | command if target is outdated. | ||
.. deprecated:: 8.22 | ||||
""" | ||||
|
r2498 | |||
|
r28664 | warnings.warn( | ||
"`target_update` is deprecated since IPython 8.22 and will be removed in future versions", | ||||
DeprecationWarning, | ||||
stacklevel=2, | ||||
) | ||||
if target_outdated(target, deps): | ||||
|
r2908 | system(cmd) | ||
|
r2498 | |||
|
r4177 | |||
|
r11701 | ENOLINK = 1998 | ||
|
r11647 | def link(src, dst): | ||
|
r11650 | """Hard links ``src`` to ``dst``, returning 0 or errno. | ||
|
r11647 | |||
|
r11701 | Note that the special errno ``ENOLINK`` will be returned if ``os.link`` isn't | ||
|
r11647 | supported by the operating system. | ||
""" | ||||
if not hasattr(os, "link"): | ||||
|
r11701 | return ENOLINK | ||
|
r11647 | link_errno = 0 | ||
try: | ||||
os.link(src, dst) | ||||
except OSError as e: | ||||
link_errno = e.errno | ||||
return link_errno | ||||
def link_or_copy(src, dst): | ||||
"""Attempts to hardlink ``src`` to ``dst``, copying if the link fails. | ||||
Attempts to maintain the semantics of ``shutil.copy``. | ||||
Because ``os.link`` does not overwrite files, a unique temporary file | ||||
will be used if the target already exists, then that file will be moved | ||||
into place. | ||||
""" | ||||
if os.path.isdir(dst): | ||||
dst = os.path.join(dst, os.path.basename(src)) | ||||
link_errno = link(src, dst) | ||||
if link_errno == errno.EEXIST: | ||||
|
r20038 | if os.stat(src).st_ino == os.stat(dst).st_ino: | ||
# dst is already a hard link to the correct file, so we don't need | ||||
# to do anything else. If we try to link and rename the file | ||||
# anyway, we get duplicate files - see http://bugs.python.org/issue21876 | ||||
return | ||||
|
r11647 | new_dst = dst + "-temp-%04X" %(random.randint(1, 16**4), ) | ||
try: | ||||
link_or_copy(src, new_dst) | ||||
except: | ||||
try: | ||||
os.remove(new_dst) | ||||
except OSError: | ||||
pass | ||||
raise | ||||
os.rename(new_dst, dst) | ||||
elif link_errno != 0: | ||||
# Either link isn't supported, or the filesystem doesn't support | ||||
# linking, or 'src' and 'dst' are on different filesystems. | ||||
shutil.copy(src, dst) | ||||
|
r16486 | |||
|
r16488 | def ensure_dir_exists(path, mode=0o755): | ||
|
r16486 | """ensure that a directory exists | ||
|
r21517 | |||
|
r16486 | If it doesn't exist, try to create it and protect against a race condition | ||
if another process is doing the same. | ||||
|
r21517 | |||
|
r16488 | The default permissions are 755, which differ from os.makedirs default of 777. | ||
|
r16486 | """ | ||
if not os.path.exists(path): | ||||
try: | ||||
os.makedirs(path, mode=mode) | ||||
except OSError as e: | ||||
if e.errno != errno.EEXIST: | ||||
raise | ||||
|
r16487 | elif not os.path.isdir(path): | ||
raise IOError("%r exists but is not a directory" % path) | ||||