Source code for pdiffcopy.operations

"""Utility functions used by the client as well as the server."""

# Standard library modules.
import errno
import logging
import os

# External dependencies.
from humanfriendly import format_size
from humanfriendly.testing import make_dirs

# Public identifiers that require documentation.
__all__ = ("get_file_info", "get_file_size", "logger", "read_block", "resize_file", "write_block")

# Initialize a logger for this module.
logger = logging.getLogger(__name__)


[docs]def get_file_info(filename): """ Get information about a local file. :param filename: An absolute filename (a string). :returns: A dictionary with file metadata, currently only the file size is included. If the file doesn't exist an empty dictionary is returned. """ size = get_file_size(filename) if size is not None: return {"size": size} else: return {}
[docs]def get_file_size(filename): """ Get the size of a local file. :param filename: An absolute filename (a string). :returns: The size of the file (an integer) or :data:`None` when the file doesn't exist. """ try: return os.path.getsize(filename) except OSError as e: if e.errno == errno.ENOENT: return None raise
[docs]def read_block(filename, offset, size): """ Read a block of data from a local file. :param filename: An absolute filename (a string). :param offset: The byte offset were reading starts (an integer). :param size: The number of bytes to read (an integer). :returns: The read data (a byte string). """ logger.debug("Reading %s block %s (%i bytes) ..", filename, offset, size) with open(filename, "rb") as handle: handle.seek(offset) return handle.read(size)
[docs]def resize_file(filename, size): """ Create or resize a local file, in preparation for synchronizing its contents. :param filename: An absolute filename (a string). :param size: The new size in bytes (an integer). """ try: handle = open(filename, "r+b") logger.info("Resizing %s to %s (%s bytes) ..", filename, format_size(size), size) except IOError as e: if e.errno != errno.ENOENT: raise logger.info("Creating %s with size %s (%s bytes) ..", filename, format_size(size), size) make_dirs(os.path.dirname(filename)) handle = open(filename, "wb") try: handle.truncate(size) finally: handle.close()
[docs]def write_block(filename, offset, data): """ Write a block of data to a local file. :param filename: An absolute filename (a string). :param offset: The byte offset were writing starts (an integer). :param data: The data to write (a byte string). """ logger.debug("Writing %s block %s (size: %s) ..", filename, offset, len(data)) with open(filename, "r+b") as handle: handle.seek(offset) handle.write(data) handle.flush()