Terminal todo list

todo_keeper.py

#!/usr/bin/env python
import os
import shelve
import functools
import time
import collections
import argparse
from abc import ABCMeta, abstractmethod

SHELVE_DB_FILE = os.path.join(os.path.dirname(__file__), 'storage.shelve')


class AbstractStorage(object):
    """
    Interface specification for storage classes
    """
    # do it in Python2 way to be compatible with it
    __metaclass__ = ABCMeta

    @abstractmethod
    def create(self, key, value):
        """
        Create action
        """

    @abstractmethod
    def read_all(self):
        """
        SELECT * action
        """

    @abstractmethod
    def update(self, key, value, status):
        """
        Update item action
        """

    @abstractmethod
    def delete(self, key):
        """
        Delete action
        """


def sync_shelve(method):
    """
    Decorator for ShelveStorage class to sync data with shelve file after
    specified method

    :param method: method should be decorated
    :return: method execution results
    """
    @functools.wraps(method)
    def wrapper(self, *args, **kwargs):
        result = method(self, *args, **kwargs)

        connection = getattr(self, 'connection', None)
        if connection:
            connection.sync()
        return result
    return wrapper


class ShelveStorage(AbstractStorage):
    """
    Storage implementation using `shelve`
    See https://docs.python.org/2.7/library/shelve.html
    """
    connection = None

    def __init__(self):
        """
        Class constructor
        """
        if not self.connection:
            self.connection = self.connect()

    @staticmethod
    def connect():
        """
        Wrapper for connection operation

        :return: Shelf object
        :rtype: Shelf
        """
        return shelve.open(SHELVE_DB_FILE, writeback=True)

    @sync_shelve
    def create(self, key, value):
        """
        Create implementation with ShelveStorage

        :param str key:
        :param str value:
        :return: created item
        :rtype: list
        """
        if self.key_exist(key):
            raise KeyError("Item already exist")

        done = False
        self.connection[key] = [value, done, time.time()]
        return self.connection[key]

    def read_all(self):
        """
        Select all implementation with ShelveStorage
        :rtype: collections.OrderedDict
        """
        items = self.connection.items()
        items.sort(key=lambda x: x[1][2])
        return collections.OrderedDict(items)

    @sync_shelve
    def update(self, key, value=None, status=None):
        """
        Update implementation with ShelveStorage

        :param str key:
        :param str value:
        :param bool status:
        :return: Updated item
        :rtype: list
        """
        if not self.key_exist(key):
            raise KeyError("Item does not exist")
        if value:
            self.connection[key][0] = value
        if status is not None:
            self.connection[key][1] = status
        return self.connection[key]

    @sync_shelve
    def delete(self, key):
        """
        Delete implementation with ShelveStorage
        """
        if not self.key_exist(key):
            raise KeyError("Item does not exist")
        del self.connection[key]

    def key_exist(self, key):
        return key in self.connection


def get_storage():
    """
    Later it will be Storage fabric, but I need to implement config at first.
    I will do it... some day.
    """
    return ShelveStorage()


class ToDoKeeper(object):
    """
    Main class which holds keeper logic
    """
    def __init__(self):
        self.storage = get_storage()

    def set(self, key, value):
        """
        Add new item

        :param str key:
        :param list value:
        :raises: ValueError
        :return: output string
        :rtype: str
        """
        if not key:
            raise ValueError("Key should be set")
        if not value:
            raise ValueError("Value should be set")
        value = ' '.join(value)
        self.storage.create(key, value)
        output = 'Item {} has been created'.format(key)
        return output

    def all(self):
        items = self.storage.read_all()
        if not items:
            return 'ToDo list is empty'
        return '\n'.join(['{key}: {value}, done: {done}'.format(
            key=k, value=v[0], done=v[1]) for (k, v) in items.items()])

    def mark_as_done(self, key):
        """
        Set item status to done
        :param str key: item name
        """
        if not key:
            raise ValueError("Key should be set")
        self.storage.update(key, status=True)
        return 'Yay! {} marked as done!'.format(key)

    def mark_as_undone(self, key):
        """
        Set item status to undone
        :param str key: item name
        """
        if not key:
            raise ValueError("Key should be set")
        self.storage.update(key, status=False)
        return '{} marked as undone!'.format(key)

    def update(self, key, value):
        """
        Update item

        :param str key: item name
        :param str value: item value
        """
        if not value:
            raise ValueError("Value should be set")
        value = ' '.join(value)
        self.storage.update(key, value=value)
        return 'Item {} updated'.format(key)

    def rm(self, key):
        """
        Remove item
        :param str key: item name
        """
        self.storage.delete(key)
        return 'Item {} has been deleted'.format(key)

    def run(self):
        """
        Entry point of ToDoKeeper
        """
        args = self.process_args()
        try:
            output = self._command_router(args)
        except (KeyError, ValueError) as exc:
            output = exc.message
        except Exception as exc:
            output = 'Something goes wrong! ' \
                     'We got unexpected error: {}'.format(exc)
        # let's use just `print` for now instead of sys.stdout(
        # which is more powerful, actually)
        print(output)

    def _command_router(self, args):
        """
        Holds logic for choosing method regarding to command
        """
        if args.command == 'set':
            return self.set(args.key, args.value)
        if args.command == 'all':
            return self.all()
        if args.command == 'rm':
            return self.rm(args.key)
        if args.command == 'upd':
            return self.update(args.key, args.value)
        if args.command == 'done':
            return self.mark_as_done(args.key)
        if args.command == 'undone':
            return self.mark_as_undone(args.key)
        # otherwise argparse will raise an exception

    @staticmethod
    def process_args():
        """
        Define and process available args with argparser
        """
        parser = argparse.ArgumentParser()
        parser.add_argument(
            'command', choices=('set', 'all', 'rm', 'upd', 'done', 'undone'))

        parser.add_argument('key', nargs='?')
        parser.add_argument('value', nargs='*')
        args = parser.parse_args()
        return args


if __name__ == '__main__':
    todo = ToDoKeeper()
    todo.run()