Changeset - 9343ba3fb2c8
[Not reviewed]
0 1 0
Lance Edgar - 9 years ago 2015-04-14 13:24:42
ledgar@sacfoodcoop.com
Tweak `locking_copy_test()` to assume destination is always a folder.

Also add constant for "file exists" error.
1 file changed with 12 insertions and 8 deletions:
0 comments (0 inline, 0 general)
rattail/files.py
Show inline comments
 
@@ -25,24 +25,25 @@ File Utilities
 

	
 
This module contains various utility functions for use with the filesystem.
 
"""
 

	
 
from __future__ import unicode_literals
 

	
 
import io
 
import os
 
import os.path
 
import shutil
 
import lockfile
 
import tempfile
 
import errno
 
from datetime import datetime
 

	
 
import pkg_resources
 

	
 

	
 
def count_lines(path, encoding=None):
 
    """
 
    Counts the number of lines in a text file.  Some attempt is made to ensure
 
    cross-platform compatibility.
 

	
 
    :param path: Path to the file.
 
    :type path: string
 
@@ -104,51 +105,54 @@ def locking_copy(src, dst, timeout=None):
 
    # being called as a filemon action.
 
    if timeout is not None:
 
        timeout = float(timeout)
 

	
 
    if os.path.isdir(dst):
 
        fn = os.path.basename(src)
 
        dst = os.path.join(dst, fn)
 

	
 
    with lockfile.LockFile(dst, timeout=timeout):
 
        shutil.copy(src, dst)
 

	
 

	
 
def locking_copy_test(src, dst):
 
def locking_copy_test(src, destdir):
 
    """
 
    Implements a "locking" version of the standard library's
 
    :func:`python:shutil.copy()` function.
 

	
 
    This exists to provide a more atomic method for copying a file into a
 
    folder which is being watched by a file monitor.  The assumption is that
 
    the monitor is configured to watch for file "locks" and therefore only
 
    process files once they have had their locks removed.  See also
 
    :ref:`filemon-profile-watchlocks`.
 

	
 
    :param src: Path to the source file.
 
    :type src: string
 

	
 
    :param dst: Path to the destination file (or directory).
 
    :type dst: string
 
    :param destdir: Path to the destination directory.
 
    :type destdir: string
 
    """
 
    if os.path.isdir(dst):
 
        fn = os.path.basename(src)
 
        dst = os.path.join(dst, fn)
 

	
 
    fn = os.path.basename(src)
 
    dst = os.path.join(destdir, fn)
 
    lockdir = '{0}.lock'.format(dst)
 

	
 
    # Attempt to create the lock dir.  We ignore "file exists" error here since
 
    # it's possible this function may be called mutliple times as part of a
 
    # filemon retry strategy.
 
    try:
 
        os.mkdir(lockdir)
 
    except OSError as error:
 
        if error.errno != 17:   # file exists
 
        if error.errno != errno.EEXIST:
 
            raise
 

	
 
    shutil.copy(src, dst)
 
    os.rmdir(lockdir)
 

	
 

	
 
def overwriting_move(src, dst):
 
    """
 
    Convenience function which is equivalent to ``shutil.move()``, except it
 
    will cause the destination file to be overwritten if it exists.
 
    """
 

	
 
    if os.path.isdir(dst):
 
        dst = os.path.join(dst, os.path.basename(src))
0 comments (0 inline, 0 general)