Priority Sorted Registry

.gitignore

*.pyc

registry.py

from collections import namedtuple


# Used internally by `Registry` for each item in its sorted list.
# Provides an easier to read API when editing the code later.
# For example, `item.name` is more clear than `item[0]`.
PriorityItem = namedtuple('PriorityItem', ['name', 'priority'])


class Registry(object):
    """
    A priority sorted registry.

    A `Registry` instance provides two public methods to alter the data of the
    registry: `register` and `deregister`. Use `register` to add items and
    `deregister` to remove items. See each method for specifics.

    When registering an item, a "name" and a "priority" must be provided. All
    items are automatically sorted by "priority" from highest to lowest. The
    "name" is used to remove ("deregister") and get items.

    A `Registry` instance it like a list (which maintains order) when reading
    data. You may iterate over the items, get an item and get a count (length)
    of all items. You may also check that the registry contains an item.

    When getting an item you may use either the index of the item or the
    string-based "name". For example:

        registry = Registry()
        registry.register(SomeItem(), 'itemname', 20)
        # Get the item by index
        item = registry[0]
        # Get the item by name
        item = registry['itemname']

    When checking that the registry contains an item, you may use either the
    string-based "name", or a reference to the actual item. For example:

        someitem = SomeItem()
        registry.register(someitem, 'itemname', 20)
        # Contains the name
        assert 'itemname' in registry
        # Contains the item instance
        assert someitem in registry

    The method `get_index_for_name` is also available to obtain the index of
    an item using that item's assigned "name".
    """

    def __init__(self):
        self._data = {}
        self._priority = []
        self._is_sorted = False

    def __contains__(self, item):
        if isinstance(item, basestring):
            # Check if an item exists by this name.
            return item in self._data.keys()
        # Check if this instance exists.
        return item in self._data.values()

    def __iter__(self):
        self._sort()
        return iter([self._data[k] for k, p in self._priority])

    def __getitem__(self, key):
        self._sort()
        if isinstance(key, slice):
            data = Registry()
            for k, p in self._priority[key]:
                data.register(self._data[k], k, p)
            return data
        if isinstance(key, int):
            return self._data[self._priority[key].name]
        return self._data[key]

    def __len__(self):
        return len(self._priority)

    def __repr__(self):
        return '<{0}({1})>'.format(self.__class__.__name__, list(self))

    def get_index_for_name(self, name):
        """
        Return the index of the given name.
        """
        self._sort()
        if name in self:
            return self._priority.index(
                [x for x in self._priority if x.name == name][0]
            )
        raise ValueError('No item named "{0}" exists.'.format(name))

    def register(self, item, name, priority):
        """
        Add an item to the registry with the given name and priority.

        Parameters:

        * `item`: The item being registered.
        * `name`: A string used to reference the item.
        * `priority`: An integer or float used to sort all items.

        If an item is registered with a "name" which already exists, the
        existing item is replaced with the new item. Tread carefully as the
        old item is lost with no way to recover it. The new item will be
        sorted according to its priority and will **not** retain the position
        of the old item.
        """
        if name in self:
            # Remove existing item of same name first
            self.deregister(name)
        self._is_sorted = False
        self._data[name] = item
        self._priority.append(PriorityItem(name, priority))

    def deregister(self, name, strict=True):
        """
        Remove an item from the registry.

        Set `strict=False` to fail silently.
        """
        try:
            index = self.get_index_for_name(name)
            del self._priority[index]
            del self._data[name]
        except ValueError:
            if strict:
                raise

    def _sort(self):
        """
        Sort the registry by priority from highest to lowest.

        This method is called internally and should never be explicitly called.
        """
        if not self._is_sorted:
            self._priority.sort(key=lambda item: item.priority, reverse=True)
            self._is_sorted = True

test_registry.py

import unittest
from registry import Registry


class Item(object):
    """ A dummy Registry item object for testing. """
    def __init__(self, data):
        self.data = data

    def __repr__(self):
        return repr(self.data)

    def __eq__(self, other):
        return self.data == other


class RegistryTests(unittest.TestCase):

    def testCreateRegistry(self):
        r = Registry()
        r.register(Item('a'), 'a', 20)
        self.assertEqual(len(r), 1)
        self.assertEqual(repr(r), "<Registry(['a'])>")

    def testRegisterWithoutPriority(self):
        r = Registry()
        with self.assertRaises(TypeError):
            r.register(Item('a'))

    def testSortRegistry(self):
        r = Registry()
        r.register(Item('a'), 'a', 20)
        r.register(Item('b'), 'b', 21)
        r.register(Item('c'), 'c', 20.5)
        self.assertEqual(len(r), 3)
        self.assertEqual(list(r), ['b', 'c', 'a'])

    def testIsSorted(self):
        r = Registry()
        self.assertFalse(r._is_sorted)
        r.register(Item('a'), 'a', 20)
        list(r)
        self.assertTrue(r._is_sorted)
        r.register(Item('b'), 'b', 21)
        self.assertFalse(r._is_sorted)
        r['a']
        self.assertTrue(r._is_sorted)
        r._is_sorted = False
        r.get_index_for_name('a')
        self.assertTrue(r._is_sorted)
        r._is_sorted = False
        repr(r)
        self.assertTrue(r._is_sorted)

    def testDeregister(self):
        r = Registry()
        r.register(Item('a'), 'a',  20)
        r.register(Item('b'), 'b', 30)
        r.register(Item('c'), 'c', 40)
        self.assertEqual(len(r), 3)
        r.deregister('b')
        self.assertEqual(len(r), 2)
        r.deregister('c', strict=False)
        self.assertEqual(len(r), 1)
        # deregister non-existant item with strict=False
        r.deregister('d', strict=False)
        self.assertEqual(len(r), 1)
        with self.assertRaises(ValueError):
            # deregister non-existant item with strict=True
            r.deregister('e')
        self.assertEqual(list(r), ['a'])

    def testRegistryContains(self):
        r = Registry()
        item = Item('a')
        r.register(item, 'a', 20)
        self.assertTrue('a' in r)
        self.assertTrue(item in r)
        self.assertFalse('b' in r)

    def testRegistryIter(self):
        r = Registry()
        r.register(Item('a'), 'a', 20)
        r.register(Item('b'), 'b', 30)
        self.assertEqual(list(r), ['b', 'a'])

    def testRegistryGetItemByIndex(self):
        r = Registry()
        r.register(Item('a'), 'a', 20)
        r.register(Item('b'), 'b', 30)
        self.assertEqual(r[0], 'b')
        self.assertEqual(r[1], 'a')
        with self.assertRaises(IndexError):
            r[3]

    def testRegistryGetItemByItem(self):
        r = Registry()
        r.register(Item('a'), 'a', 20)
        r.register(Item('b'), 'b', 30)
        self.assertEqual(r['a'], 'a')
        self.assertEqual(r['b'], 'b')
        with self.assertRaises(KeyError):
            r['c']

    def testRegistrySetItem(self):
        r = Registry()
        with self.assertRaises(TypeError):
            r[0] = 'a'
        with self.assertRaises(TypeError):
            r['a'] = 'a'

    def testRegistryDelItem(self):
        r = Registry()
        r.register(Item('a'), 'a', 20)
        with self.assertRaises(TypeError):
            del r[0]
        with self.assertRaises(TypeError):
            del r['a']

    def testRegistrySlice(self):
        r = Registry()
        r.register(Item('a'), 'a', 20)
        r.register(Item('b'), 'b', 30)
        r.register(Item('c'), 'c', 40)
        slc = r[1:]
        self.assertEqual(len(slc), 2)
        self.assertTrue(isinstance(slc, Registry))
        self.assertEqual(list(slc), ['b', 'a'])

    def testGetIndexForName(self):
        r = Registry()
        r.register(Item('a'), 'a', 20)
        r.register(Item('b'), 'b', 30)
        self.assertEqual(r.get_index_for_name('a'), 1)
        self.assertEqual(r.get_index_for_name('b'), 0)
        with self.assertRaises(ValueError):
            r.get_index_for_name('c')

    def testRegisterDupplicate(self):
        r = Registry()
        r.register(Item('a'), 'a', 20)
        r.register(Item('b1'), 'b', 10)
        self.assertEqual(list(r), ['a', 'b1'])
        self.assertEqual(len(r), 2)
        r.register(Item('b2'), 'b', 30)
        self.assertEqual(len(r), 2)
        self.assertEqual(list(r), ['b2', 'a'])

if __name__ == '__main__':
    unittest.main()