Running a SharedExtension without Entrypoint

1

Hi!
I am developing a ban system based on iptables.
So I need to run a service that will accept incoming rpc for settings & querying iptables rules.
On nameko run some setup procedures should be called like creating new iptables chains etc.
But this must be run only once on startup.
So I am trying to use a SharedExtension. Like this:

class SecurityExtension(DependencyProvider):
    def get_dependency(self, worker_ctx):
        for ext in self.container.extensions:
            if isinstance(ext, SecurityManager):
                return ext
        logger.error('Cannot find SecurityManager!')


class SecurityManager(SharedExtension, ProviderCollector):
    filter_ports = []

    def setup(self):
        self.filter_ports = ['5060'] # TODO: Get from server or directly from asterisk AMI
        logger.info('Security filter ports: %s', self.filter_ports)


class SecurityBroker:
    name = 'security'
    manager = SecurityExtension()

    @rpc
    def test(self):
        return self.manager.filter_ports

But test() always returns [] and setup never is called.
So if I add a new Entrypoint class it works:

class SecurityEvent(Entrypoint):
    manager = SecurityManager()

security = SecurityEvent.decorator

and add the following method to SecurityBroker class:

    @security
    def security_stub():
        pass

And all works.

But I have a question if there is another way to setup some code on startup of nameko run.
Thanks.

2

I don’t think you want a SharedExtension here. SharedExtension is used where you want a single extension instance to be shared by multiple other extension instances (e.g. all Rpc entrypoint instances share one RpcConsumer, because there is one queue from which they all consume).

The pattern you want, where something runs once at service startup, can be achieved with a simple DependencyProvider:

class SecurityManager(DependencyProvider):
    filter_ports = []

    def setup(self):
        # do one-time setup here
        self.filter_ports = ['5060']

    def get_dependency(self, worker_ctx):
        # return anything you want the service methods to access here
        return self.filter_ports

class SecurityBroker:
    name = 'security'
    manager = SecurityManager()

    @rpc
    def test(self):
        return self.manager.filter_ports

Your later example works by accident, and is not the intended way.

It’s normally nice to return a proper class instance from get_dependency so that you can expose a rich API to the service method (while still hiding the complexity of the DependencyProvider internals) but if a simple dict or list is sufficient that’s fine too.

3

Thank you Matt!
It worked.