decoupler.py

'''
Given two modules that want to reference each other in a way that would
otherwise create an import loop, one mechanism for handling this is
a deferred import.

def foo():
    import bar

    bar.bar()

However, this doesn't work for this that need to happen at import time.

@bar.bar
def foo():  # cannot be done if bar imports foo

Although not a general solution to this problem, this module
provides a handy mechanism for "meet in the middle" imports
that allow data structures to be reigstered from one module
to another without either needing to import.
'''
import sys
import traceback
import collections


class DeferredRegistrationError(Exception):
    '''combines stacks when a deferred registration happens'''


class _Registration(object):
    def __init__(self, stack, data):
        self.stack, self.data = stack, data


class Decoupler(object):
    def __init__(self):
        self.registrations = collections.defaultdict(list)
        self.pending_registrations = collections.defaultdict(list)
        self.registries = {(self.__module__, 'registry'): self.register_registry}
        self.finalized = False

    def _flush(self, name):
        if name not in self.registries:
            return
        register = self.registries[name]
        pending_registrations = self.pending_registrations[name]
        while pending_registrations:
            registration = pending_registrations[-1]
            try:
                register(registration.data)
            except Exception as e:
                msg = '\n' + ''.join(registration.stack) + traceback.format_exc()
                msg = msg.replace('\n', '\n  ')
                raise DeferredRegistrationError(msg)
            else:
                pending_registrations.pop()

    def register(self, name, data):
        if self.finalized:
            raise DeferredRegistrationError("cannot register (already finalized)")
        registration = _Registration(traceback.format_stack(), data)
        self.registrations[name].append(registration)
        self.pending_registrations[name].append(registration)
        self._flush(name)
        
    def register_registry(self, name, registry):
        if self.finalized:
            raise DeferredRegistrationError("cannot register (already finalized)")
        if name in self.registries:
            raise ValueError('double registration for {!r}'.format(name))
        self.registries[name] = registry
        self._flush(name)

    def finalize(self):
        '''
        finalize the registry -- no further registrations allowed;
        any outstanding are bugs
        '''
        err_lines = []
        for name in self.pending_registrations:
            for reg in self.pending_registrations[name]:
                err_lines.append('unclaimed registration for ' + name + ' from')
                err_lines += reg.stack
        if err_lines:
            raise DeferredRegistrationError('\n  '.join(err_lines))



_DECOUPLER = Decoupler()


def register(name, data):
    '''register data with the named registry'''
    _DECOUPLER.register(name, data)


def register_registry(name, registry):
    '''register a new registry function under name'''
    _DECOUPLER.register_registry(name, registry)


def _test():
    registered_okay = []
    def reg(data):
        if data is 'error':
            1/0
        registered_okay.append(data)

    register('test.1', 1)
    assert _DECOUPLER.pending_registrations['test.1']
    register_registry('test.1', reg)
    assert 'test.1' in _DECOUPLER.registries
    register('test.1', 2)
    assert registered_okay == [1, 2], registered_okay

    register('test.2', 'error')
    assert _DECOUPLER.pending_registrations['test.2']
    try:
        register_registry('test.2', reg)
    except:
        traceback.print_exc()

    decoupler = Decoupler()
    decoupler.register('test.1', 1)
    try:
        decoupler.finalize()
        assert False, "finalize didn't raise with incomplete registrations"
    except:
        pass
    decoupler.register_registry('test.1', reg)
    decoupler.finalize()

    print("good")


if __name__ == "__main__":
    _test()