iPOPO in 10 minutes

Authors

Shadi Abras, Thomas Calmant

This tutorial presents how to use the iPOPO framework and its associated service-oriented component model. The concepts of the service-oriented component model are introduced, followed by a simple example that demonstrates the features of iPOPO. This framework uses decorators to describe components.

Note

This tutorial has been updated to use types instead of named specifications.

Introduction

iPOPO aims to simplify service-oriented programming on OSGi frameworks in Python language; the name iPOPO is an abbreviation for injected POPO, where POPO would stand for Plain Old Python Object. The name is in fact a simple modification of the Apache iPOJO project, which stands for injected Plain Old Java Object

iPOPO provides a new way to develop OSGi/iPOJO-like service components in Python, simplifying service component implementation by transparently managing the dynamics of the environment as well as other non-functional requirements. The iPOPO framework allows developers to more clearly separate functional code (i.e. POPOs) from the non-functional code (i.e. dependency management, service provision, configuration, etc.). At run time, iPOPO combines the functional and non-functional aspects. To achieve this, iPOPO provides a simple and extensible service component model based on POPOs.

Since iPOPO v3, we recommend using types as much as possible to avoid issues when developing large software with the framework.

Basic concepts

iPOPO is separated into two parts:

• Pelix, the underlying bundle and service registry

• iPOPO, the service-oriented component framework

It also defines three major concepts:

• A bundle is a single Python module, i.e. a .py file, that is loaded using the Pelix API.

• A service is a Python object that is registered to the service registry using the Pelix API, associated to a set of specifications and to a dictionary of properties.

• A component is an instance of component factory, i.e. a class manipulated by iPOPO decorators. Those decorators injects information into the class that are later used by iPOPO to manage the components. Components are defined inside bundles.

Simple example

In this tutorial we will present how to:

• Publish a service

• Require a service

• Use lifecycle callbacks to activate and deactivate components

Presentation of the Spell application

To illustrate some of iPOPO features, we will implement a very simple application. Three bundles compose this application:

• A bundle that defines a component implementing a dictionary service (an English and a French dictionaries).

• One with a component requiring the dictionary service and providing a spell checker service.

• One that defines a component requiring the spell checker and providing a user line interface.

The spell dictionary components provide the spell_dictionary_service specification. The spell checker provides a spell_checker_service specification.

Preparing the tutorial

The example contains several bundles:

• spell_checker_api.py defines the Python protocols that describe the different services in use.

• spell_dictionary_EN.py defines a component that implements the Dictionary service, containing some English words.

• spell_dictionary_FR.py defines a component that implements the Dictionary service, containing some French words.

• spell_checker.py contains an implementation of a Spell Checker. The spell checker requires a dictionary service and checks if an input passage is correct, according to the words contained in the wished dictionary.

• spell_client.py provides commands for the Pelix shell service. This component uses a spell checker service. The user can interact with the spell checker with this command line interface.

Finally, a main_pelix_launcher.py script starts the Pelix framework. It is not considered as a bundle as it is not loaded by the framework, but it can control the latter.

Defining specifications

Note

This section is new in iPOPO v3

Instead of relying exclusively on specification names to link components together, it is now recommended to declare protocols or classes and to use those to declare injected fields.

For example, the spell_checker_api bundle contains only the definition of the specifications we will use in this project. It is recommended to use a specific file to define specifications and constants in order to share it between the provider and consumer bundles.

#!/usr/bin/python
# -- Content-Encoding: UTF-8 --
"""
Module defining the types used in the Spell Checker
"""

from typing import Protocol

from pelix.constants import Specification


# Set the name of the specification
@Specification("spell_dictionary_service")
class SpellDictionary(Protocol):
    """
    Definition of the spell dictionary service
    """

    def check_word(self, word: str) -> bool:
        """
        Determines if the given word is contained in the dictionary.

        @param word the word to be checked.
        @return True if the word is in the dictionary, False otherwise.
        """
        ...


@Specification("spell_checker_service")
class SpellChecker(Protocol):
    """
    Definition of the spell checker service
    """

    def check(self, passage: str, language: str = "EN") -> list[str] | None:
        """
        Checks the given passage for misspelled words.

        :param passage: the passage to spell check.
        :param language: language of the spell dictionary to use
        :return: An array of misspelled words or null if no words are misspelled
        :raise KeyError: No dictionary for this language
        """
        ...

• The @Specification decorator will store in the protocol/class the name of the specification. This is highly recommended as it will be the name used in the Pelix service registry and when communicating with remote framework if you want to use remote services. See Specifications for the details of the decorator, and why the name should be given explicitly.

• We recommend using the Python Protocol as parent of each specification class as it is meant to declare a type.

Once the specifications are defined, we can continue by implementing them with different components.

Note

Depending on your own code style, you can either repeat the explicit types on the methods implementing a specification, or let them be inherited from the protocol.

Also note that the specification module must be importable by the bundles, but doesn’t need to be installed as a bundle itself.

The English dictionary bundle: Providing a service

The spell_dictionary_EN bundle is a simple implementation of the Dictionary service. It contains few English words.

#!/usr/bin/python
# -- Content-Encoding: UTF-8 --
"""
This bundle provides a component that is a simple implementation of the
Dictionary service. It contains some English words.
"""

from spell_checker_api import SpellDictionary

from pelix.framework import BundleContext
from pelix.ipopo.decorators import ComponentFactory, Instantiate, Invalidate, Property, Provides, Validate


# Name the iPOPO component factory
@ComponentFactory("spell_dictionary_en_factory")
# This component provides a dictionary service
@Provides(SpellDictionary)
# It is the English dictionary
@Property("_language", "language", "EN")
# Automatically instantiate a component when this factory is loaded
@Instantiate("spell_dictionary_en_instance")
class EnglishSpellDictionary(SpellDictionary):
    """
    Implementation of a spell dictionary, for English language.
    """

    def __init__(self) -> None:
        """
        Declares members, to respect PEP-8.
        """
        self.dictionary: set[str] = set()

    @Validate
    def validate(self, context: BundleContext) -> None:
        """
        The component is validated. This method is called right before the
        provided service is registered to the framework.
        """
        # All setup should be done here
        self.dictionary = {"hello", "world", "welcome", "to", "the", "ipopo", "tutorial"}
        print("An English dictionary has been added")

    @Invalidate
    def invalidate(self, context: BundleContext) -> None:
        """
        The component has been invalidated. This method is called right after
        the provided service has been removed from the framework.
        """
        self.dictionary = set()

    # No need to have explicit types: annotations are inherited
    def check_word(self, word):
        word = word.lower().strip()
        return not word or word in self.dictionary

• The @Component decorator is used to declare an iPOPO component. It must always be on top of other decorators.

• The @Provides decorator indicates that the component provides a service. We also indicate the type of service we provide, either using the type directly (recommended) or its specification name.

• The @Instantiate decorator instructs iPOPO to automatically create an instance of our component. The relation between components and instances is the same than between classes and objects in the object-oriented programming.

• The @Property decorator indicates the properties associated to this component and to its services (e.g. French or English language).

• The method decorated with @Validate will be called when the instance becomes valid.

• The method decorated with @Invalidate will be called when the instance becomes invalid (e.g. when one its dependencies goes away) or is stopped.

For more information about decorators, see iPOPO Decorators.

In order for IDEs and type checking tools like MyPy to help you developing components, you should indicate that the component class inherits from the specification protocols it provides.

The French dictionary bundle: Providing a service

The spell_dictionary_FR bundle is a similar to the spell_dictionary_EN one. It only differs in the language component property, as it checks some French words declared during component validation.

#!/usr/bin/python
# -- Content-Encoding: UTF-8 --
"""
This bundle provides a component that is a simple implementation of the
Dictionary service. It contains some French words.
"""

from spell_checker_api import SpellDictionary

from pelix.framework import BundleContext
from pelix.ipopo.decorators import ComponentFactory, Instantiate, Invalidate, Property, Provides, Validate


# Name the iPOPO component factory
@ComponentFactory("spell_dictionary_fr_factory")
# This component provides a dictionary service
@Provides(SpellDictionary)
# It is the French dictionary
@Property("_language", "language", "FR")
# Automatically instantiate a component when this factory is loaded
@Instantiate("spell_dictionary_fr_instance")
class FrenchSpellDictionary(SpellDictionary):
    """
    Implementation of a spell dictionary, for French language.
    """

    def __init__(self) -> None:
        """
        Declares members, to respect PEP-8.
        """
        self.dictionary: set[str] = set()

    @Validate
    def validate(self, context: BundleContext) -> None:
        """
        The component is validated. This method is called right before the
        provided service is registered to the framework.
        """
        # All setup should be done here
        self.dictionary = {"bonjour", "le", "monde", "au", "a", "ipopo", "tutoriel"}
        print("A French dictionary has been added")

    @Invalidate
    def invalidate(self, context: BundleContext) -> None:
        """
        The component has been invalidated. This method is called right after
        the provided service has been removed from the framework.
        """
        self.dictionary = set()

    # No need to have explicit types: annotations are inherited
    def check_word(self, word):
        word = word.lower().strip()
        return not word or word in self.dictionary

It is important to note that the iPOPO factory name must be unique in a framework: only the first one to be registered with a given name will be taken into account. The name of component instances follows the same rule.

The spell checker bundle: Requiring a service

The spell_checker bundle aims to provide a spell checker service. However, to serve this service, this implementation requires a dictionary service. During this step, we will create an iPOPO component requiring a Dictionary service and providing the Spell Checker service.

#!/usr/bin/python
# -- Content-Encoding: UTF-8 --
"""
The spell_checker component uses the dictionary services to check the spell of
a given text.
"""

import re

from spell_checker_api import SpellChecker, SpellDictionary

from pelix.framework import BundleContext
from pelix.internals.registry import ServiceReference
from pelix.ipopo.decorators import (
    BindField,
    ComponentFactory,
    Instantiate,
    Invalidate,
    Provides,
    Requires,
    UnbindField,
    Validate,
)


# Name the component factory
@ComponentFactory("spell_checker_factory")
# Provide a Spell Checker service
@Provides(SpellChecker)
# Consume all Spell Dictionary services available (aggregate them)
@Requires("_spell_dictionaries", SpellDictionary, aggregate=True)
# Automatic instantiation
@Instantiate("spell_checker_instance")
class SpellCheckerImpl:
    """
    A component that uses spell dictionary services to check the spelling of
    given texts.
    """

    # We can declare the type of injected fields
    _spell_dictionaries: list[SpellDictionary]

    def __init__(self) -> None:
        """
        Define class members
        """
        # the list of available dictionaries, constructed
        self.languages: dict[str, SpellDictionary] = {}

        # list of some punctuation marks could be found in the given passage,
        # internal
        self.punctuation_marks: set[str] = set()

    @BindField("_spell_dictionaries")
    def bind_dict(
        self, field: str, service: SpellDictionary, svc_ref: ServiceReference[SpellDictionary]
    ) -> None:
        """
        Called by iPOPO when a spell dictionary service is bound to this
        component
        """
        # Extract the dictionary language from its properties
        language = svc_ref.get_property("language")

        # Store the service according to its language
        self.languages[language] = service

    @UnbindField("_spell_dictionaries")
    def unbind_dict(
        self, field: str, service: SpellDictionary, svc_ref: ServiceReference[SpellDictionary]
    ) -> None:
        """
        Called by iPOPO when a dictionary service has gone away
        """
        # Extract the dictionary language from its properties
        language = svc_ref.get_property("language")

        # Remove it from the computed storage
        # The injected list of services is updated by iPOPO
        del self.languages[language]

    @Validate
    def validate(self, context: BundleContext) -> None:
        """
        This spell checker has been validated, i.e. at least one dictionary
        service has been bound.
        """
        # Set up internal members
        self.punctuation_marks = {",", ";", ".", "?", "!", ":", " "}
        print("A spell checker has been started")

    @Invalidate
    def invalidate(self, context: BundleContext) -> None:
        """
        The component has been invalidated
        """
        self.punctuation_marks = set()
        print("A spell checker has been stopped")

    def check(self, passage, language="EN"):
        # list of words to be checked in the given passage
        # without the punctuation marks
        checked_list = re.split("([!,?.:; ])", passage)
        try:
            # Get the dictionary corresponding to the requested language
            dictionary = self.languages[language]
        except KeyError:
            # Not found
            raise KeyError(f"Unknown language: {language}")

        # Do the job, calling the found service
        return [
            word
            for word in checked_list
            if word not in self.punctuation_marks and not dictionary.check_word(word)
        ]

• The @Requires decorator specifies a service dependency. This required service is injected in a local variable in this bundle. Its aggregate attribute tells iPOPO to collect the list of services. Again, the specification can be given by type or by name. providing the required specification, instead of the first one.

• The @BindField decorator indicates that a new required service bounded to the platform.

• The @UnbindField decorator indicates that one of required service has gone away.

As you can see, the injected field is declared twice:

• in the @Requires decorator, so that iPOPO knows what fields must be injected and how, which is mandatory for iPOPO to work

• at the class level, to give the injected field a type hint

This is due to a limitation of Python that doesn’t support annotating class members. That being said, it is highly recommended to manually declare the field class level at class with its type in order to benefit fully from type checking tools and code completion in your IDE.

Also note that type hint you indicate must match the parameters you give to @Requires:

• If optional is set True, the injected value can be None, else it can only be of the specification type

• If aggregate is set to True, the injected value is a list

• Adapt the type for more complex decorators like @RequiresMap, …

The spell client bundle

The spell_client bundle contains a very simple user interface allowing a user to interact with a spell checker service.

#!/usr/bin/python
# -- Content-Encoding: UTF-8 --
"""
This bundle defines a component that consumes a spell checker.
It provides a shell command service, registering a "spell" command that can be
used in the shell of Pelix.

It uses a dictionary service to check for the proper spelling of a word by check
for its existence in the dictionary.
"""

from spell_checker_api import SpellChecker

from pelix.framework import BundleContext
from pelix.ipopo.decorators import ComponentFactory, Instantiate, Invalidate, Provides, Requires, Validate
from pelix.shell import ShellCommandsProvider
from pelix.shell.beans import ShellSession


# Name the component factory
@ComponentFactory("spell_client_factory")
# Consume a single Spell Checker service
@Requires("_spell_checker", SpellChecker)
# Provide a shell command service
@Provides(ShellCommandsProvider)
# Automatic instantiation
@Instantiate("spell_client_instance")
class SpellClient(ShellCommandsProvider):
    """
    A component that provides a shell command (spell.spell), using a
    Spell Checker service.
    """

    # Declare the injected field with its type
    _spell_checker: SpellChecker

    @Validate
    def validate(self, context: BundleContext) -> None:
        """
        Component validated, just print a trace to visualize the event.
        Between this call and the call to invalidate, the _spell_checker member
        will point to a valid spell checker service.
        """
        print("A client for spell checker has been started")

    @Invalidate
    def invalidate(self, context: BundleContext) -> None:
        """
        Component invalidated, just print a trace to visualize the event
        """
        print("A spell client has been stopped")

    def get_namespace(self):
        """
        Retrieves the name space of this shell command provider.
        Look at the shell tutorial for more information.
        """
        return "spell"

    def get_methods(self):
        """
        Retrieves the list of (command, method) tuples for all shell commands
        provided by this component.
        Look at the shell tutorial for more information.
        """
        return [("spell", self.spell)]

    def spell(self, session: ShellSession):
        """
        Reads words from the standard input and checks for their existence
        from the selected dictionary.

        :param session: The shell session, a bean to interact with the user
        """
        # Request the language of the text to the user
        passage = None
        language = session.prompt("Please enter your language, EN or FR: ")
        language = language.upper()

        while passage != "quit":
            # Request the text to check
            passage = session.prompt("Please enter your paragraph, or 'quit' to exit:\n")

            if passage and passage != "quit":
                # A text has been given: call the spell checker, which have been
                # injected by iPOPO.
                misspelled_words = self._spell_checker.check(passage, language)
                if not misspelled_words:
                    session.write_line("All words are well spelled!")
                else:
                    session.write_line(f"The misspelled words are: {misspelled_words}")

The component defined here implements and provides a shell command service, which will be consumed by the Pelix Shell Core Service. It registers a spell shell command.

Main script: Launching the framework

We have all the bundles required to start playing with the application. To run the example, we have to start Pelix, then all the required bundles.

#!/usr/bin/python
# -- Content-Encoding: UTF-8 --
"""
Starts a Pelix framework and installs the Spell Checker bundles
"""

# Standard library
import logging

# Our spell checker API
from spell_checker_api import SpellChecker

# Pelix framework module and utility methods
import pelix.framework
from pelix.utilities import use_service


def main():
    """
    Starts a Pelix framework and waits for it to stop
    """
    # Prepare the framework, with iPOPO and the shell console
    # Warning: we only use the first argument of this method, a list of bundles
    framework = pelix.framework.create_framework(
        (
            # iPOPO
            "pelix.ipopo.core",
            # Shell core (engine)
            "pelix.shell.core",
            # Text console
            "pelix.shell.console",
        )
    )

    # Start the framework, and the pre-installed bundles
    framework.start()

    # Get the bundle context of the framework, i.e. the link between the
    # framework starter and its content.
    context = framework.get_bundle_context()

    # Start the spell dictionary bundles, which provide the dictionary services
    context.install_bundle("spell_dictionary_EN").start()
    context.install_bundle("spell_dictionary_FR").start()

    # Start the spell checker bundle, which provides the spell checker service.
    context.install_bundle("spell_checker").start()

    # Sample usage of the spell checker service
    # 1. get its service reference, that describes the service itself
    ref_config = context.get_service_reference(SpellChecker)
    if ref_config is None:
        print("Error: spell service not found")
        return

    # 2. the use_service method allows to grab a service and to use it inside a
    # with block. It automatically releases the service when exiting the block,
    # even if an exception was raised
    with use_service(context, ref_config) as svc_config:
        # Here, svc_config points to the spell checker service
        passage = "Welcome to our framwork iPOPO"
        print("1. Testing Spell Checker:", passage)
        misspelled_words = svc_config.check(passage)
        print(">  Misspelled_words are:", misspelled_words)

    # Start the spell client bundle, which provides a shell command
    context.install_bundle("spell_client").start()

    # Wait for the framework to stop
    framework.wait_for_stop()


# Classic entry point...
if __name__ == "__main__":
    logging.basicConfig(level=logging.DEBUG)
    main()

Running the application

Launch the main_pelix_launcher.py script. When the framework is running, type in the console: spell to enter your language choice and then your passage.

Here is a sample run, calling python main_pelix_launcher.py:

INFO:pelix.shell.core:Shell services registered
An English dictionary has been added
** Pelix Shell prompt **
A French dictionary has been added
A dictionary checker has been started
1. Testing Spell Checker: Welcome to our framwork iPOPO
>  Misspelled_words are: ['our', 'framwork']
A client for spell checker has been started

$ spell
Please enter your language, EN or FR: FR
Please enter your paragraph, or 'quit' to exit:
Bonjour le monde !
All words are well spelled !
Please enter your paragraph, or 'quit' to exit:
quit
$ spell
Please enter your language, EN or FR: EN
Please enter your paragraph, or 'quit' to exit:
Hello, world !
All words are well spelled !
Please enter your paragraph, or 'quit' to exit:
Bonjour le monde !
The misspelled words are: ['Bonjour', 'le', 'monde']
Please enter your paragraph, or 'quit' to exit:
quit
$ quit
Bye !
A spell client has been stopped
INFO:pelix.shell.core:Shell services unregistered

You can now go back to see other tutorials or take a look at the Reference Cards.