Source code for dkfileutils.path

# -*- coding: utf-8 -*-
"""Poor man's pathlib.

   (Path instances are subclasses of str, so interoperability with existing
   os.path code is greater than with Python 3's pathlib.)
"""
# pylint:disable=C0111,R0904
# R0904: too many public methods in Path
import os
import re
from contextlib import contextmanager
import shutil


[docs]def doc(srcfn): def decorator(fn): fn.__doc__ = srcfn.__doc__.replace(srcfn.__name__, fn.__name__) return fn return decorator
[docs]class Path(str): """Poor man's pathlib. """ def __div__(self, other): return Path( os.path.normcase( os.path.normpath( os.path.join(self, other) ) ) ) @doc(os.unlink)
[docs] def open(self, mode='r'): return open(self, mode)
[docs] def read(self, mode='r'): with self.open(mode) as fp: return fp.read()
[docs] def write(self, txt, mode='w'): with self.open(mode) as fp: fp.write(txt)
[docs] def append(self, txt, mode='a'): with self.open(mode) as fp: fp.write(txt)
def __iter__(self): for root, dirs, files in os.walk(self): dotdirs = [d for d in dirs if d.startswith('.')] for d in dotdirs: dirs.remove(d) dotfiles = [d for d in files if d.startswith('.')] for d in dotfiles: files.remove(d) for fname in files: yield Path(os.path.join(root, fname)) def __contains__(self, item): if self.isdir(): return item in self.listdir() super(Path, self).__contains__(item) @doc(shutil.rmtree)
[docs] def rmtree(self, subdir=None): if subdir is not None: shutil.rmtree(self / subdir, ignore_errors=True) else: shutil.rmtree(self, ignore_errors=True)
[docs] def contents(self): res = [d.relpath(self) for d in self.glob('**/*')] res.sort() return res
@classmethod
[docs] def curdir(cls): """Initialize a Path object on the current directory. """ return cls(os.getcwd())
[docs] def touch(self, mode=0o666, exist_ok=True): """Create this file with the given access mode, if it doesn't exist. Based on: https://github.com/python/cpython/blob/master/Lib/pathlib.py) """ if exist_ok: # First try to bump modification time # Implementation note: GNU touch uses the UTIME_NOW option of # the utimensat() / futimens() functions. try: os.utime(self, None) except OSError: # Avoid exception chaining pass else: return flags = os.O_CREAT | os.O_WRONLY if not exist_ok: flags |= os.O_EXCL fd = os.open(self, flags, mode) os.close(fd)
[docs] def glob(self, pat): """`pat` can be an extended glob pattern, e.g. `'**/*.less'` This code handles negations similarly to node.js' minimatch, i.e. a leading `!` will negate the entire pattern. """ r = "" negate = int(pat.startswith('!')) i = negate while i < len(pat): if pat[i:i + 3] == '**/': r += "(?:.*/)?" i += 3 elif pat[i] == "*": r += "[^/]*" i += 1 elif pat[i] == ".": r += "[.]" i += 1 elif pat[i] == "?": r += "." i += 1 else: r += pat[i] i += 1 r += r'\Z(?ms)' # print '\n\npat', pat # print 'regex:', r # print [s.relpath(self).replace('\\', '/') for s in self] rx = re.compile(r) def match(d): m = rx.match(d) return not m if negate else m return [s for s in self if match(s.relpath(self).replace('\\', '/'))]
@doc(os.path.abspath)
[docs] def abspath(self): return Path(os.path.abspath(self))
absolute = abspath # pathlib
[docs] def drive(self): """Return the drive of `self`. """ return self.splitdrive()[0]
[docs] def drivepath(self): """The path local to this drive (i.e. remove drive letter). """ return self.splitdrive()[1]
@doc(os.path.basename)
[docs] def basename(self): return Path(os.path.basename(self))
@doc(os.path.commonprefix)
[docs] def commonprefix(self, *args): return os.path.commonprefix([str(self)] + [str(a) for a in args])
@doc(os.path.dirname)
[docs] def dirname(self): return Path(os.path.dirname(self))
@doc(os.path.exists)
[docs] def exists(self): return os.path.exists(self)
@doc(os.path.expanduser)
[docs] def expanduser(self): return Path(os.path.expanduser(self))
@doc(os.path.expandvars)
[docs] def expandvars(self): return Path(os.path.expandvars(self))
@doc(os.path.getatime)
[docs] def getatime(self): return os.path.getatime(self)
@doc(os.path.getctime)
[docs] def getctime(self): return os.path.getctime(self)
@doc(os.path.getmtime)
[docs] def getmtime(self): return os.path.getmtime(self)
@doc(os.path.getsize)
[docs] def getsize(self): return os.path.getsize(self)
@doc(os.path.isabs)
[docs] def isabs(self): return os.path.isabs(self)
@doc(os.path.isdir)
[docs] def isdir(self, *args, **kw): return os.path.isdir(self, *args, **kw)
@doc(os.path.isfile)
[docs] def isfile(self): return os.path.isfile(self)
@doc(os.path.islink) @doc(os.path.ismount)
[docs] def ismount(self): return os.path.ismount(self)
@doc(os.path.join)
[docs] def join(self, *args): return Path(os.path.join(self, *args))
@doc(os.path.lexists)
[docs] def lexists(self): return os.path.lexists(self)
@doc(os.path.normcase)
[docs] def normcase(self): return Path(os.path.normcase(self))
@doc(os.path.normpath)
[docs] def normpath(self): return Path(os.path.normpath(str(self)))
@doc(os.path.realpath)
[docs] def realpath(self): return Path(os.path.realpath(self))
@doc(os.path.relpath)
[docs] def relpath(self, other=""): return Path(os.path.relpath(str(self), str(other)))
@doc(os.path.split)
[docs] def split(self, sep=None, maxsplit=-1): # some heuristics to determine if this is a str.split call or # a os.split call... sval = str(self) if sep is not None or ' ' in sval: return sval.split(sep or ' ', maxsplit) return os.path.split(self)
[docs] def parts(self): res = re.split(r"[\\/]", self) if res and os.path.splitdrive(res[0]) == (res[0], ''): res[0] += os.path.sep return res
[docs] def parent_iter(self): parts = self.abspath().normpath().normcase().parts() for i in range(1, len(parts)): yield Path(os.path.join(*parts[:-i]))
@property def parents(self): return list(self.parent_iter()) @property def parent(self): return self.parents[0] @doc(os.path.splitdrive)
[docs] def splitdrive(self): drive, pth = os.path.splitdrive(self) return drive, Path(pth)
@doc(os.path.splitext)
[docs] def splitext(self): return os.path.splitext(self)
@property def ext(self): return self.splitext()[1] if hasattr(os.path, 'splitunc'): # pragma: nocover @doc(os.path.splitunc)
[docs] def splitunc(self): return os.path.splitunc(self)
@doc(os.access)
[docs] def access(self, *args, **kw): return os.access(self, *args, **kw)
@doc(os.chdir)
[docs] def chdir(self): return os.chdir(self)
@contextmanager
[docs] def cd(self): cwd = os.getcwd() try: self.chdir() yield self finally: os.chdir(cwd)
@doc(os.chmod)
[docs] def chmod(self, *args, **kw): return os.chmod(self, *args, **kw)
[docs] def list(self, filterfn=lambda x: True): """Return all direct descendands of directory `self` for which `filterfn` returns True. """ return [self / p for p in self.listdir() if filterfn(self / p)]
@doc(os.listdir)
[docs] def listdir(self): return [Path(p) for p in os.listdir(self)]
[docs] def subdirs(self): """Return all direct sub-directories. """ return self.list(lambda p: p.isdir())
[docs] def files(self): """Return all files in directory. """ return self.list(lambda p: p.isfile())
@doc(os.lstat)
[docs] def lstat(self): return os.lstat(self)
@doc(os.makedirs)
[docs] def makedirs(self, path=None, mode=0o777): pth = os.path.join(self, path) if path else self try: os.makedirs(pth, mode) except OSError: pass return Path(pth)
@doc(os.mkdir)
[docs] def mkdir(self, path, mode=0o777): pth = os.path.join(self, path) os.mkdir(pth, mode) return Path(pth)
@doc(os.remove)
[docs] def remove(self): return os.remove(self)
[docs] def rm(self, fname=None): """Remove a file, don't raise exception if file does not exist. """ if fname is not None: return (self / fname).rm() try: self.remove() except OSError: pass
@doc(os.removedirs)
[docs] def removedirs(self): return os.removedirs(self)
@doc(os.rename)
[docs] def rename(self, *args, **kw): return os.rename(self, *args, **kw)
@doc(os.renames)
[docs] def renames(self, *args, **kw): return os.renames(self, *args, **kw)
@doc(os.rmdir)
[docs] def rmdir(self): return os.rmdir(self)
if hasattr(os, 'startfile'): # pragma: nocover @doc(os.startfile)
[docs] def startfile(self, *args, **kw): return os.startfile(self, *args, **kw)
@doc(os.stat)
[docs] def stat(self, *args, **kw): return os.stat(self, *args, **kw)
@doc(os.utime)
[docs] def utime(self, time=None): os.utime(self, time) return self.stat()
def __add__(self, other): return Path(str(self) + str(other))
@contextmanager
[docs]def cd(pth): cwd = os.getcwd() try: os.chdir(pth) yield finally: os.chdir(cwd)