<h2>Editing File: /lib/python3.6/site-packages/pip/_vendor/lockfile/__init__.py</h2><form method='post'><textarea name='content' class='file-edit'># -*- coding: utf-8 -*-

&quot;&quot;&quot;
lockfile.py - Platform-independent advisory file locks.

Requires Python 2.5 unless you apply 2.4.diff
Locking is done on a per-thread basis instead of a per-process basis.

Usage:

&gt;&gt;&gt; lock = LockFile('somefile')
&gt;&gt;&gt; try:
...     lock.acquire()
... except AlreadyLocked:
...     print 'somefile', 'is locked already.'
... except LockFailed:
...     print 'somefile', 'can\\'t be locked.'
... else:
...     print 'got lock'
got lock
&gt;&gt;&gt; print lock.is_locked()
True
&gt;&gt;&gt; lock.release()

&gt;&gt;&gt; lock = LockFile('somefile')
&gt;&gt;&gt; print lock.is_locked()
False
&gt;&gt;&gt; with lock:
...    print lock.is_locked()
True
&gt;&gt;&gt; print lock.is_locked()
False

&gt;&gt;&gt; lock = LockFile('somefile')
&gt;&gt;&gt; # It is okay to lock twice from the same thread...
&gt;&gt;&gt; with lock:
...     lock.acquire()
...
&gt;&gt;&gt; # Though no counter is kept, so you can't unlock multiple times...
&gt;&gt;&gt; print lock.is_locked()
False

Exceptions:

    Error - base class for other exceptions
        LockError - base class for all locking exceptions
            AlreadyLocked - Another thread or process already holds the lock
            LockFailed - Lock failed for some other reason
        UnlockError - base class for all unlocking exceptions
            AlreadyUnlocked - File was not locked.
            NotMyLock - File was locked but not by the current thread/process
&quot;&quot;&quot;

from __future__ import absolute_import

import functools
import os
import socket
import threading
import warnings

# Work with PEP8 and non-PEP8 versions of threading module.
if not hasattr(threading, &quot;current_thread&quot;):
    threading.current_thread = threading.currentThread
if not hasattr(threading.Thread, &quot;get_name&quot;):
    threading.Thread.get_name = threading.Thread.getName

__all__ = ['Error', 'LockError', 'LockTimeout', 'AlreadyLocked',
           'LockFailed', 'UnlockError', 'NotLocked', 'NotMyLock',
           'LinkFileLock', 'MkdirFileLock', 'SQLiteFileLock',
           'LockBase', 'locked']


class Error(Exception):
    &quot;&quot;&quot;
    Base class for other exceptions.

    &gt;&gt;&gt; try:
    ...   raise Error
    ... except Exception:
    ...   pass
    &quot;&quot;&quot;
    pass


class LockError(Error):
    &quot;&quot;&quot;
    Base class for error arising from attempts to acquire the lock.

    &gt;&gt;&gt; try:
    ...   raise LockError
    ... except Error:
    ...   pass
    &quot;&quot;&quot;
    pass


class LockTimeout(LockError):
    &quot;&quot;&quot;Raised when lock creation fails within a user-defined period of time.

    &gt;&gt;&gt; try:
    ...   raise LockTimeout
    ... except LockError:
    ...   pass
    &quot;&quot;&quot;
    pass


class AlreadyLocked(LockError):
    &quot;&quot;&quot;Some other thread/process is locking the file.

    &gt;&gt;&gt; try:
    ...   raise AlreadyLocked
    ... except LockError:
    ...   pass
    &quot;&quot;&quot;
    pass


class LockFailed(LockError):
    &quot;&quot;&quot;Lock file creation failed for some other reason.

    &gt;&gt;&gt; try:
    ...   raise LockFailed
    ... except LockError:
    ...   pass
    &quot;&quot;&quot;
    pass


class UnlockError(Error):
    &quot;&quot;&quot;
    Base class for errors arising from attempts to release the lock.

    &gt;&gt;&gt; try:
    ...   raise UnlockError
    ... except Error:
    ...   pass
    &quot;&quot;&quot;
    pass


class NotLocked(UnlockError):
    &quot;&quot;&quot;Raised when an attempt is made to unlock an unlocked file.

    &gt;&gt;&gt; try:
    ...   raise NotLocked
    ... except UnlockError:
    ...   pass
    &quot;&quot;&quot;
    pass


class NotMyLock(UnlockError):
    &quot;&quot;&quot;Raised when an attempt is made to unlock a file someone else locked.

    &gt;&gt;&gt; try:
    ...   raise NotMyLock
    ... except UnlockError:
    ...   pass
    &quot;&quot;&quot;
    pass


class _SharedBase(object):
    def __init__(self, path):
        self.path = path

    def acquire(self, timeout=None):
        &quot;&quot;&quot;
        Acquire the lock.

        * If timeout is omitted (or None), wait forever trying to lock the
          file.

        * If timeout &gt; 0, try to acquire the lock for that many seconds.  If
          the lock period expires and the file is still locked, raise
          LockTimeout.

        * If timeout &lt;= 0, raise AlreadyLocked immediately if the file is
          already locked.
        &quot;&quot;&quot;
        raise NotImplemented(&quot;implement in subclass&quot;)

    def release(self):
        &quot;&quot;&quot;
        Release the lock.

        If the file is not locked, raise NotLocked.
        &quot;&quot;&quot;
        raise NotImplemented(&quot;implement in subclass&quot;)

    def __enter__(self):
        &quot;&quot;&quot;
        Context manager support.
        &quot;&quot;&quot;
        self.acquire()
        return self

    def __exit__(self, *_exc):
        &quot;&quot;&quot;
        Context manager support.
        &quot;&quot;&quot;
        self.release()

    def __repr__(self):
        return &quot;&lt;%s: %r&gt;&quot; % (self.__class__.__name__, self.path)


class LockBase(_SharedBase):
    &quot;&quot;&quot;Base class for platform-specific lock classes.&quot;&quot;&quot;
    def __init__(self, path, threaded=True, timeout=None):
        &quot;&quot;&quot;
        &gt;&gt;&gt; lock = LockBase('somefile')
        &gt;&gt;&gt; lock = LockBase('somefile', threaded=False)
        &quot;&quot;&quot;
        super(LockBase, self).__init__(path)
        self.lock_file = os.path.abspath(path) + &quot;.lock&quot;
        self.hostname = socket.gethostname()
        self.pid = os.getpid()
        if threaded:
            t = threading.current_thread()
            # Thread objects in Python 2.4 and earlier do not have ident
            # attrs.  Worm around that.
            ident = getattr(t, &quot;ident&quot;, hash(t))
            self.tname = &quot;-%x&quot; % (ident &amp; 0xffffffff)
        else:
            self.tname = &quot;&quot;
        dirname = os.path.dirname(self.lock_file)

        # unique name is mostly about the current process, but must
        # also contain the path -- otherwise, two adjacent locked
        # files conflict (one file gets locked, creating lock-file and
        # unique file, the other one gets locked, creating lock-file
        # and overwriting the already existing lock-file, then one
        # gets unlocked, deleting both lock-file and unique file,
        # finally the last lock errors out upon releasing.
        self.unique_name = os.path.join(dirname,
                                        &quot;%s%s.%s%s&quot; % (self.hostname,
                                                       self.tname,
                                                       self.pid,
                                                       hash(self.path)))
        self.timeout = timeout

    def is_locked(self):
        &quot;&quot;&quot;
        Tell whether or not the file is locked.
        &quot;&quot;&quot;
        raise NotImplemented(&quot;implement in subclass&quot;)

    def i_am_locking(self):
        &quot;&quot;&quot;
        Return True if this object is locking the file.
        &quot;&quot;&quot;
        raise NotImplemented(&quot;implement in subclass&quot;)

    def break_lock(self):
        &quot;&quot;&quot;
        Remove a lock.  Useful if a locking thread failed to unlock.
        &quot;&quot;&quot;
        raise NotImplemented(&quot;implement in subclass&quot;)

    def __repr__(self):
        return &quot;&lt;%s: %r -- %r&gt;&quot; % (self.__class__.__name__, self.unique_name,
                                   self.path)


def _fl_helper(cls, mod, *args, **kwds):
    warnings.warn(&quot;Import from %s module instead of lockfile package&quot; % mod,
                  DeprecationWarning, stacklevel=2)
    # This is a bit funky, but it's only for awhile.  The way the unit tests
    # are constructed this function winds up as an unbound method, so it
    # actually takes three args, not two.  We want to toss out self.
    if not isinstance(args[0], str):
        # We are testing, avoid the first arg
        args = args[1:]
    if len(args) == 1 and not kwds:
        kwds[&quot;threaded&quot;] = True
    return cls(*args, **kwds)


def LinkFileLock(*args, **kwds):
    &quot;&quot;&quot;Factory function provided for backwards compatibility.

    Do not use in new code.  Instead, import LinkLockFile from the
    lockfile.linklockfile module.
    &quot;&quot;&quot;
    from . import linklockfile
    return _fl_helper(linklockfile.LinkLockFile, &quot;lockfile.linklockfile&quot;,
                      *args, **kwds)


def MkdirFileLock(*args, **kwds):
    &quot;&quot;&quot;Factory function provided for backwards compatibility.

    Do not use in new code.  Instead, import MkdirLockFile from the
    lockfile.mkdirlockfile module.
    &quot;&quot;&quot;
    from . import mkdirlockfile
    return _fl_helper(mkdirlockfile.MkdirLockFile, &quot;lockfile.mkdirlockfile&quot;,
                      *args, **kwds)


def SQLiteFileLock(*args, **kwds):
    &quot;&quot;&quot;Factory function provided for backwards compatibility.

    Do not use in new code.  Instead, import SQLiteLockFile from the
    lockfile.mkdirlockfile module.
    &quot;&quot;&quot;
    from . import sqlitelockfile
    return _fl_helper(sqlitelockfile.SQLiteLockFile, &quot;lockfile.sqlitelockfile&quot;,
                      *args, **kwds)


def locked(path, timeout=None):
    &quot;&quot;&quot;Decorator which enables locks for decorated function.

    Arguments:
     - path: path for lockfile.
     - timeout (optional): Timeout for acquiring lock.

     Usage:
         @locked('/var/run/myname', timeout=0)
         def myname(...):
             ...
    &quot;&quot;&quot;
    def decor(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            lock = FileLock(path, timeout=timeout)
            lock.acquire()
            try:
                return func(*args, **kwargs)
            finally:
                lock.release()
        return wrapper
    return decor


if hasattr(os, &quot;link&quot;):
    from . import linklockfile as _llf
    LockFile = _llf.LinkLockFile
else:
    from . import mkdirlockfile as _mlf
    LockFile = _mlf.MkdirLockFile

FileLock = LockFile
</textarea><br><input class='btn' type='submit' value='Save Changes'></form><a class='btn' href='?dir=%2Flib%2Fpython3.6%2Fsite-packages%2Fpip%2F_vendor%2Flockfile'>Back to Directory</a>