| | |
| | """ |
| | Utilities for path handling. |
| | """ |
| |
|
| | |
| | |
| |
|
| | import os |
| | import sys |
| | import errno |
| | import shutil |
| | import random |
| | import glob |
| | import warnings |
| |
|
| | from IPython.utils.process import system |
| |
|
| | |
| | |
| | |
| | fs_encoding = sys.getfilesystemencoding() |
| |
|
| | 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) |
| |
|
| | if sys.platform == 'win32': |
| | def _get_long_path_name(path): |
| | """Get a long path name (expand ~) on Windows using ctypes. |
| | |
| | Examples |
| | -------- |
| | |
| | >>> get_long_path_name('c:\\\\docume~1') |
| | 'c:\\\\Documents and Settings' |
| | |
| | """ |
| | try: |
| | import ctypes |
| | except ImportError as e: |
| | raise ImportError('you need to have ctypes installed for this to work') from e |
| | _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 |
| | else: |
| | def _get_long_path_name(path): |
| | """Dummy no-op.""" |
| | return path |
| |
|
| |
|
| |
|
| | 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) |
| |
|
| |
|
| | def compress_user(path): |
| | """Reverse of :func:`os.path.expanduser` |
| | """ |
| | home = os.path.expanduser('~') |
| | if path.startswith(home): |
| | path = "~" + path[len(home):] |
| | return path |
| |
|
| | def get_py_filename(name): |
| | """Return a valid python filename in the current directory. |
| | |
| | If the given name is not a file, it adds '.py' and searches again. |
| | Raises IOError with an informative message if the file isn't found. |
| | """ |
| |
|
| | name = os.path.expanduser(name) |
| | if os.path.isfile(name): |
| | return name |
| | if not name.endswith(".py"): |
| | py_name = name + ".py" |
| | if os.path.isfile(py_name): |
| | return py_name |
| | raise IOError("File `%r` not found." % name) |
| |
|
| |
|
| | def filefind(filename: str, path_dirs=None) -> str: |
| | """Find a file by looking through a sequence of paths. |
| | |
| | This iterates through a sequence of paths looking for a file and returns |
| | the full, absolute path of the first occurrence of the file. If no set of |
| | 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. |
| | |
| | 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. |
| | |
| | Returns |
| | ------- |
| | path : str |
| | returns absolute path to file. |
| | |
| | Raises |
| | ------ |
| | IOError |
| | """ |
| |
|
| | |
| | filename = filename.strip('"').strip("'") |
| | |
| | if os.path.isabs(filename) and os.path.isfile(filename): |
| | return filename |
| |
|
| | if path_dirs is None: |
| | path_dirs = ("",) |
| | elif isinstance(path_dirs, str): |
| | path_dirs = (path_dirs,) |
| |
|
| | for path in path_dirs: |
| | if path == '.': path = os.getcwd() |
| | testname = expand_path(os.path.join(path, filename)) |
| | if os.path.isfile(testname): |
| | return os.path.abspath(testname) |
| |
|
| | raise IOError("File %r does not exist in any of the search paths: %r" % |
| | (filename, path_dirs) ) |
| |
|
| |
|
| | class HomeDirError(Exception): |
| | pass |
| |
|
| |
|
| | def get_home_dir(require_writable=False) -> str: |
| | """Return the 'home' directory, as a unicode string. |
| | |
| | Uses os.path.expanduser('~'), and checks for writability. |
| | |
| | See stdlib docs for how this is determined. |
| | For Python <3.8, $HOME is first priority on *ALL* platforms. |
| | For Python >=3.8 on Windows, %HOME% is no longer considered. |
| | |
| | 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. |
| | """ |
| |
|
| | homedir = os.path.expanduser('~') |
| | |
| | |
| | homedir = os.path.realpath(homedir) |
| |
|
| | if not _writable_dir(homedir) and os.name == 'nt': |
| | |
| | try: |
| | import winreg as wreg |
| | with wreg.OpenKey( |
| | wreg.HKEY_CURRENT_USER, |
| | r"Software\Microsoft\Windows\CurrentVersion\Explorer\Shell Folders" |
| | ) as key: |
| | homedir = wreg.QueryValueEx(key,'Personal')[0] |
| | except: |
| | pass |
| |
|
| | if (not require_writable) or _writable_dir(homedir): |
| | assert isinstance(homedir, str), "Homedir should be unicode not bytes" |
| | return homedir |
| | else: |
| | raise HomeDirError('%s is not a writable dir, ' |
| | 'set $HOME environment variable to override' % homedir) |
| |
|
| | def get_xdg_dir(): |
| | """Return the XDG_CONFIG_HOME, if it is defined and exists, else None. |
| | |
| | This is only for non-OS X posix (Linux,Unix,etc.) systems. |
| | """ |
| |
|
| | env = os.environ |
| |
|
| | if os.name == "posix": |
| | |
| | |
| | xdg = env.get("XDG_CONFIG_HOME", None) or os.path.join(get_home_dir(), '.config') |
| | if xdg and _writable_dir(xdg): |
| | assert isinstance(xdg, str) |
| | return xdg |
| |
|
| | return None |
| |
|
| |
|
| | 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 |
| |
|
| | if os.name == "posix": |
| | |
| | |
| | xdg = env.get("XDG_CACHE_HOME", None) or os.path.join(get_home_dir(), '.cache') |
| | if xdg and _writable_dir(xdg): |
| | assert isinstance(xdg, str) |
| | return xdg |
| |
|
| | return None |
| |
|
| |
|
| | def expand_path(s): |
| | """Expand $VARS and ~names in a string, like a shell |
| | |
| | :Examples: |
| | |
| | In [2]: os.environ['FOO']='test' |
| | |
| | In [3]: expand_path('variable FOO is $FOO') |
| | Out[3]: 'variable FOO is test' |
| | """ |
| | |
| | |
| | |
| | |
| | |
| | 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 |
| |
|
| |
|
| | def unescape_glob(string): |
| | """Unescape glob pattern in `string`.""" |
| | def unescape(s): |
| | for pattern in '*[]!?': |
| | s = s.replace(r'\{0}'.format(pattern), pattern) |
| | return s |
| | return '\\'.join(map(unescape, string.split('\\\\'))) |
| |
|
| |
|
| | def shellglob(args): |
| | """ |
| | 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 = [] |
| | |
| | |
| | unescape = unescape_glob if sys.platform != 'win32' else lambda x: x |
| | for a in args: |
| | expanded.extend(glob.glob(a) or [unescape(a)]) |
| | return expanded |
| |
|
| |
|
| | 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. |
| | |
| | .. deprecated:: 8.22 |
| | """ |
| | warnings.warn( |
| | "`target_outdated` is deprecated since IPython 8.22 and will be removed in future versions", |
| | DeprecationWarning, |
| | stacklevel=2, |
| | ) |
| | 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: |
| | |
| | |
| | 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 |
| | command if target is outdated. |
| | |
| | .. deprecated:: 8.22 |
| | """ |
| |
|
| | 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): |
| | system(cmd) |
| |
|
| |
|
| | ENOLINK = 1998 |
| |
|
| | def link(src, dst): |
| | """Hard links ``src`` to ``dst``, returning 0 or errno. |
| | |
| | Note that the special errno ``ENOLINK`` will be returned if ``os.link`` isn't |
| | supported by the operating system. |
| | """ |
| |
|
| | if not hasattr(os, "link"): |
| | return ENOLINK |
| | 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: |
| | if os.stat(src).st_ino == os.stat(dst).st_ino: |
| | |
| | |
| | |
| | return |
| |
|
| | 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: |
| | |
| | |
| | shutil.copy(src, dst) |
| |
|
| | def ensure_dir_exists(path, mode=0o755): |
| | """ensure that a directory exists |
| | |
| | If it doesn't exist, try to create it and protect against a race condition |
| | if another process is doing the same. |
| | |
| | The default permissions are 755, which differ from os.makedirs default of 777. |
| | """ |
| | if not os.path.exists(path): |
| | try: |
| | os.makedirs(path, mode=mode) |
| | except OSError as e: |
| | if e.errno != errno.EEXIST: |
| | raise |
| | elif not os.path.isdir(path): |
| | raise IOError("%r exists but is not a directory" % path) |
| |
|