Q2 Cores

To interact with a core, the Caliper SDK uses the q2-cores extension. Each core inside this extension uses coreflows to make requests and handle responses. A coreflow includes the code to generate a certain type of request as well as the code to handle and objectify the response for use by an extension. Think of them as endpoints for an API that communicates with a specific core.

As new coreflows are created, they are added to our q2-cores extension. We expect this collection to grow over time, but if your requirement is not yet implemented and a new coreflow is needed to handle it, the create_coreflow CLI entrypoint can be used to extend an existing core and create a new coreflow.

Fetching a user’s demographic information is a common core request. Metavante, a commonly-used core, requires two requests for a complete demographic model: a request for a user’s basic information, and a second request for a user’s email addresses. Both of these are included with q2-cores, but let’s demonstrate how the email request was created using create_coreflow.


Perhaps the trickiest part of making a coreflow is finding an example request and response for your use case, you will need this shape to make your coreflow. We hope to provide some tooling around this task in the future, for now you will need to use your internal resources and documentations to collect this information. For this example, we used the Metavante documentation to get these shapes.

A coreflow is a just a special kind of extension. Instead of create_extension, run the create_coreflow entrypoint:

$ q2 create_coreflow
Which core will your coreflow extend?
Please choose a number above and press Return

    1) base
    2) PCS
    3) UltraData
    4) Metavante
    5) EPL
    6) Summit
    7) Users
    8) JXchange
    9) OSI
    10) XP2
    11) SOA
    12) Symitar

    Page 1 of 2. Select 'n' for next page.

Please make a selection and press Return:

Choose one of the cores to extend. These are cores that have some functionality within the q2-sdk and the one you choose will serve as the parent core that your new core will inherit from. Let’s choose Metavante.


You may have to upgrade your q2-cores module to get the latest cores. If you don’t see a full list like the one above, try running q2 upgrade -p cores

Next, choose a name for your new core. The default is fine here (ExtendedMetavante). Leaving it empty will choose the default:

New core name [ExtendedMetavante] <return>

Choose a name for your new coreflow operation:

Please select a name for your coreflow operation: GetEmail

Finally choose whether you want to set ExtendedMetavante as your default core. We won’t be making actual core calls in this tutorial so it’s not particularly important but you will need to set it to make core calls in the future:

You do not have a core set in your settings file.
Would you like to set ExtendedMetavante as your default core? [Y/n]


You can also use command line options to generate coreflows more quickly. Declare your operation name with -o, parent core with -p, core name with -c. $ q2 create_coreflow -o GetEmail -p Metavante -c ExtendedMetavante

A new directory “ExtendedMetavante” was generated for us, containing 4 subdirectories mappers, models, queries, and tests. It also contains a core.py file at the top level. Each of these contains some automatically built classes and functions to make building coreflows easier. Let’s take a look at the contents of our new custom core.

Coreflow Components

Queries: Query receives all the parameters required to make our desired core request and injects them into a string template, creating a valid core request. It also contains a mock core response for testing.

Mappers: Mapper receives a stringified core response and analyzes it, returning a Python object Model representing that data that can be easily used by other extensions.

Models: A plain Python object that contains the objectified core response. This is created by the Mapper class and passed along to other extensions by the Core class.

Core: Core inherits and extends the parent core you specified when generating this extension. The method contained within will be called by other extensions using this extension to make a core request.

Tests: Includes test functions for both the Mapper and Query; these test the output of those classes’ methods against a desired result that you provide.


These tests are not required, but are the best way to ensure your methods are generating what you think they are! Testing against a real core is tricky, so it is highly recommended you test your code against test data first.

Your extension is generated with example data. You can run the tests with this data if you wish, using q2 test:

$ q2 test


To create our GetEmail coreflow, we will be replacing this example code with code that will work with a Metavante core.

The first step is to know what you are making: finding out the shape of the requests and responses you will be working with. For Metavante’s email API, our response will need to look like this:

    <cfg appID="CI" svcID="CICustEmailInq" svcVer="4.0"></cfg>

We know from our core documentation that “E10033” corresponds to a user’s customer ID. This is a straightforward request: provide the customer ID, and receive the customer’s email address. We will worry about fetching this parameter later, for now we just need our Query class to handle it.

First, we need a way for our Query class to receive and store this id when created. Navigate to queries/get_email_query.py and enter this to receive customer_id:

class GetEmailQuery(BaseQuery):
    def __init__(self,
                 logger: logging.Logger,
                 customer_id: str,
        self.customer_id = customer_id

Then, we need to include the correct XML structure inside the build method:

def build(self) -> str:
    Constructs and returns the core request from a template and provided parameters.
    root = lxml.etree.Element('ExampleRequest')
    root.text = self.customer_id

    xml_as_bytes = lxml.etree.tostring(root)
    return xml_as_bytes.decode('utf8')

This will create a basic xml structure, format the customer_id stored in the query class into its proper position inside the xml element, and then return a request string suitable for sending to our Metavante core.


Caliper SDK uses the “lxml” library to work with XML. Full documentation for “lxml” can be found here: http://lxml.de/.

Let’s test this class.

We know what our end result should look like, so let’s use the test_get_email_query function inside tests/test_get_email.py to compare the result of our build method with the desired result:

def test_get_email_query():
    query = GetEmailQuery(LOGGER, "123456789")
    actual = query.raw_query
    expected = TEST_REQUEST_STRING
    assert actual == expected

Now declare your desired result as the constant variable TEST_REQUEST_STRING:

TEST_REQUEST_STRING = "<ExampleRequest>123456789</ExampleRequest>"

If everything’s working well, your test_get_email_query should pass just fine. If not, it will tell you where the mismatch is.


Now we can build valid queries, provided required parameters. You don’t need to worry about actually sending the request to the core: q2-cores handles that step. Our next step is actually to handle the core response, packaging it into a model for later use.

Our Model class for this task is simple, as we only care about one attribute. Let’s set up our model to accept it. You’ll find it in models/get_email_model.py:

class GetEmailModel:
    A python object representing the data contained in the core response
    def __init__(self, email=None):
        self.email = email

Simple enough, but this will allow consumers of this coreflow to access this data easily.


Now we need to set up our Mapper class to convert the stringified response we will get from Metavante into the easy-to-use model we defined above.

Before we start, let’s take a look at what we expect to get from Metavante when we make our request:


Looks like the email we are looking for will be contained inside that “E10528” tag, several levels deep. Our Mapper needs to account for this structure. We’ll set up our mapper to parse this xml using the included lxml library and extract the email. You’ll find it within mappers/GetEmail.py:

class GetEmailMapper(BaseMapper):
    def parse_returned_queries(list_of_queries) -> GetEmailModel:

        # Sanity checks, change if more than one or more than one type of query is needed
        assert len(list_of_queries) == 1
        assert isinstance(list_of_queries[0],
                          GetEmailQuery), 'Query must be an instance of GetEmailQuery'

        response = list_of_queries[0].raw_core_response

        # Use lxml to create a python object representing the XML response
        MtvnSvcRes = lxml.objectify.fromstring(response)

        #Assign desired attributes from the response and pass them to the model
        email_tag = MtvnSvcRes.Svc.MsgData.CICustEmailInqResData.CICustEmailInqRptDataLst.CICustEmailInqRptData.E10528
        email = email_tag.text

        return GetEmailModel(email=email)

That’s a lot of nesting, but lxml can handle it. lxml represents child elements as attributes of its parent, allowing access to deeply nested children with a simple chain of attributes. Once we have the element we need, we can access its contents with the text attribute– this is the email address we came for.

Our mapper will return an instance of our Model class, with the data we extracted from the response contained inside for later use.

Testing our mapper again requires to know our desired result ahead of time. In this case, we want a GetEmailModel containing the data we extracted assigned to the constant variable TEST_RESULT_MODEL:

TEST_RESULT_MODEL = GetEmailModel(email="test@email.com")

But that’s not enough: our mapper test needs input. It’s time to go back to our Query class and fill out the mock_response method:

def mock_response(self):
    return """<MtvnSvcRes>

Now our test class can generate a Query, run our mock response through our Mapper, and test that the model it generates matches what we expect. Remember that our Query needs a customer_id to be instantiated:

def test_get_email_mapper():
    query = GetEmailQuery(LOGGER, "123456789")
    query.raw_core_response = query.mock_response()

    queries = [query]
    actual = GetEmailMapper.parse_returned_queries(queries)
    expected = TEST_RESULT_MODEL
    assert vars(actual) == vars(expected)

Finishing the Coreflow

Almost done now. Our last step is to open core.py and set up our Core class– this is what another extension would use to make a real core request:

class Core(ParentCore):
    CONFIG_FILE_NAME = "ExtendedMetavante"


    async def build_get_email(self) -> GetEmailMapper:
        Other extensions will call this function to make this extension's core call.
        :return: A model representing the contents of the core response
        query = GetEmailQuery(self.logger, self.configured_user.customer_id)

        return GetEmailMapper([query])

Now its time to figure out where our required customer_id is going to come from. Usually, parameters will come from configuration or from the online_user object used by the SDK to keep track of user information. In this case, we can simply look up self.configured_user, an attribute defined in the base Core class. Its customer_id attribute has what we need.

In this case, we don’t have any required configurations. But what if our query also needed a special key constant, unique to the FI making the request? This key would be added to REQUIRED_CONFIGURATIONS. Our parent core class will set these keys as attributes on an object called self.config; you can then pass them on to the Query just like our customer_id:

class Core(ParentCore):
    CONFIG_FILE_NAME = "ExtendedMetavante"
        "fi_unique_key": "1234567890abcdefghijkl"

    async def build_get_email(self) -> GetEmailMapper:
        Other extensions will call this function to make this extension's core call.
        :return: A model representing the contents of the core response
        query = GetEmailQuery(self.logger, self.configured_user.customer_id, self.config.fi_unique_key)

        return GetEmailMapper([query], hq_session_token=self.hq_auth_token)

And we’re done. You can now use your completed coreflow extension with any other extension to make a core request you had no access to before.

In the consuming extension, set your CORE configuration setting to your new core instead of the base core:

CORE = 'ExtendedMetavante.core'

When your consuming request handler is instantiated, your coreflow Core class will be assigned to self.core. The consuming extension will be able to access the methods you just created to make core calls:

class GetEmailConsumingRequestHandler(q2_sdk.core.web.Q2RequestHandler):
    async def get(self):
        mapper = await self.core.build_get_email()
        core_response = await mapper.execute()
        self.write("The email for the logged in user is {}".format(core_response.email)