p4core.py

import types
import copy
from pprint import pformat
import os

from P4 import P4, P4Exception

__all__ = ['p4', 'p4run', 'get_msg', 'tolist', 'p4sync', 'is_depot_path', 'touch']

# Initialized p4 instance
p4 = P4()

_last_messages = []
_last_warnings = []
_last_errors = []

def get_msg():
    """
    P4 stores messages, warnings and errors of the last executed command in three
    global variables.

    return: A dictionary which contains these information. 
    """
    return {
        'messages': _last_messages,
        'warnings': _last_warnings,
        'errors': _last_errors,
    }

class RunResult(object):
    
    """
    This class provides easy access to results returned from p4.run() command.
    """
    
    def __init__(self, result):
        """
        result: Object (typically a list) returned by p4.run().
        """
        self.__result = result

    @property
    def value(self):
        """
        Raw result object.
        """
        return self.__result

    def __iter__(self):
        """
        Yield each item from result if it is a list, otherwise yield the result.
        """
        if isinstance(self.__result, types.ListType):
            for r in self.__result:
                yield r
        else:
            yield self.__result

    def first(self):
        """
        return: The first item in result if it is a list, otherwise return the result.
        """
        try:
            return self.__result[0]
        except IndexError:
            return self.__result

    def __getitem__(self, key):
        """
        key: Dictionary key for the first item.
        return: [key]'s value from the first item.
        """
        return self.first()[key]

    def __setitem__(self, key, value):
        """
        key: Dictionary key for the first item.
        value: Value to set in first item's key.
        """
        self.first()[key] = value

    def get(self, key, default=None):
        """
        key: Dictionary key for the first item.
        default: Value to substitute if the key does not exist.
        return: Value belong to this key in the first item.
        """
        try:
            return self[key]
        except KeyError:
            return default

    def at(self, index):
        """
        index: Item in position <index> you want to access. 
        return: Item at position <index> if it exists.
        """
        if not isinstance(self.__result, types.ListType):
            raise TypeError('Result is not a list. Type: {0}'.format(type(self.__result)))
        return self.__result[index]

    def __len__(self):
        """
        return: Number of items in result.
        """
        if self.__result:
            if isinstance(self.__result, types.ListType):
                return len(self.__result)
            return 1
        return 0

    def has_key(self, key):
        """
        key: Dictionary key for the first item.
        return: True if the key exists otherwise False.
        """
        try:
            self[key]
        except (TypeError, KeyError):
            # todo: return original exception
            return False
        return True

    def __str__(self):
        """
        A string representation of this class.
        """
        if len(self) == 1:
            return pformat(self.first())
        return pformat(self.__result)

    def dump(self):
        """
        Print this class in string.
        """
        print(self)


def p4run(*args, **kwargs):
    """
    A single interface to communicate with Perforce API
    We can use this as choke point to insert logging and 
    telemetry data points.

    args: Same as p4.run() args
    kwargs: Same as p4.run() kwargs
    return: A RunResult object.
    """
    global p4
    global _last_messages
    global _last_warnings
    global _last_errors

    if not args:
        raise P4Exception('No command specified.')

    messages = []
    warnings = []
    errors = []

    if not p4.connected():
        p4.connect()

    if kwargs:
        results = p4.run(args, kwargs)
    else:
        results = p4.run(args)
    
    # Store information generated by executing this command
    # Todo: Extend this to a full log which records everything including
    # commands
    _last_messages = copy.copy(p4.messages)
    _last_warnings = copy.copy(p4.warnings)
    _last_errors = copy.copy(p4.errors)

    return RunResult(results)

def tolist(item, remove_duplicate=True):
    """
    Converts an item to a list if it is not already wrapped by a list.
    
    item: Anything
    return: A list that contains item
    """
    if isinstance(item, (types.ListType, types.TupleType)):
       ret = item
    elif isinstance(item, set):
        ret = list(item)
    else:
        ret = [item]

    if remove_duplicate and type(item) != set:
        try:
            ret = list(set(ret))
        except TypeError:
            # Some objects may not be hashable used by set.
            # In this case we ignore duplicates and move on.
            pass

    return ret

# Cached p4 information
p4info = p4run('info')

def p4sync(paths, ignore_errors=True):
    """
    Download files from Perforce server to local client.

    paths: A file path or a list of paths. Perforce filespec is supported.
    ignore_errors: If True, no exception is raised when error is encountered;
    if False, exception is raised.
    """
    try:
        paths = tolist(paths)
        p4run('sync', *paths)
    except P4Exception:
        if not ignore_errors:
            raise

def is_depot_path(path):
    """
    path: A file path
    return True if it starts with //; False otherwise
    """
    if not isinstance(path, types.StringTypes):
        raise TypeError('A string file path is expected, not {0}.'.format(type(path)))
    return path.startswith('//')

def touch(path):
    """
    Similar to the unix command: touch

    path: A local file path
    """
    if os.path.exists(path):
        return
    with open(path, 'w') as f:
        f.write('')