Installing the Extension

After the extension files are created and edited to fit the use case, they need to be installed. This process will create a record of this extension in the database with the extension name as the lookup value. The install process will also install any database configs that are defined in the extension code.

Install the extension with the q2 install command, and complete the prompts shown. After, the install can be confirmed with the q2 get_installed command.

Calling the Extension

Now that the extension is installed, it should be callable from the API. If calling the Caliper API is new, there are tutorials on how to call various endpoints in the Caliper API documentation.

The get method can be tested locally without the API. After starting the SDK server with the q2 run command, the curl program can be used to send a request to the local host URL:

$ curl http://localhost:1980/TestCustomEndpoint\?getProductTypes\=true

A web browser can also be used to verify the functionality of the endpoint.

Before continuing, double check that the code looks similar to the below example.

extension.py

"""
TestCustomEndpoint Extension
"""

from q2_sdk.core.http_handlers.caliper_api_custom_handler import (
    Q2CaliperAPICustomHandler,
)
from q2_sdk.hq.db.product import Product
from q2_sdk.hq.db.product_type import ProductType
from q2_sdk.hq.hq_api.q2_api import ChangeEndUserAccountNickname
from q2_sdk.hq.models.db_config.db_config import DbConfig
from q2_sdk.hq.models.db_config.db_config_list import DbConfigList
from q2_sdk.tools.utils import to_bool

from .install.db_plan import DbPlan


class TestCustomEndpointHandler(Q2CaliperAPICustomHandler):
    ## REQUIRED_CONFIGURATIONS is a dictionary of key value pairs that are necessary
    ## for the extension to run. If set, ensures the entries are set in the
    ## extension's settings file or the web server will not start.
    ## Keys are names and values are defaults written into the settings file.
    ## To override the defaults, generate the config (`q2 generate_config`) and
    ## then alter the resulting file

    # REQUIRED_CONFIGURATIONS = {
    #    # 'CONFIG1': 'Default',
    #    # 'CONFIG2': 'Default',
    # }

    # # Behaves the same way as REQUIRED_CONFIGURATIONS, but will not stop the web server
    # # from starting if omitted from the extension's settings file
    # OPTIONAL_CONFIGURATIONS = {}

    # # Behaves in a similar manner to REQUIRED_CONFIGURATIONS,
    # # but stores the data in the database instead of the settings file. Will be
    # # written into the database on `q2 install`
    WEDGE_ADDRESS_CONFIGS = DbConfigList(
        [
            DbConfig("NickNamePrefix", "API Set: "),
        ]
    )

    DESCRIPTION = "Caliper API Custom Endpoint TestCustomEndpoint"

    DB_PLAN = DbPlan()

    CONFIG_FILE_NAME = "TestCustomEndpoint"  # configuration/TestCustomEndpoint.py file must exist if REQUIRED_CONFIGURATIONS exist

    REQUIRED_PARAMETERS = ["LoginName", "HostAccountID", "RequestedNickName"]

    def __init__(self, application, request, **kwargs):
        """
        If you need variables visible through the lifetime of this request,
        feel free to add them in this function
        """
        super().__init__(application, request, **kwargs)
        # self.variable_example = 12345

    # # Uncomment this to allow the IDE to give you better hinting on a specific core (Symitar in this example)
    # from q2_cores.Symitar.core import Core as SymitarCore
    # @property
    # def core(self) -> SymitarCore:

    #     # noinspection PyTypeChecker
    #     return super().core
    async def get(self, *args, **kwargs):

        self.set_header("Content-Type", "application/json")

        object_class = Product(self.logger, hq_credentials=self.hq_credentials)
        product_types = to_bool(self.request_parameters.get("getProductTypes"))
        if product_types:
            object_class = ProductType(self.logger, hq_credentials=self.hq_credentials)

        data = await object_class.get()
        product_data = []
        for row in data:
            product_data.append({x.tag: x.text for x in row.getchildren()})

        response = self.return_data(product_data)

        self.logger.info(response)
        self.write(response)

    async def post(self):

        self.set_header("Content-Type", "application/json")

        if not self.request_parameters:
            errors = [{"code": 400, "message": "No request body"}]
            to_return = self.return_data({}, success=False, errors=errors)
            self.write(to_return)
            return

        missing_parameters = [x for x in self.REQUIRED_PARAMETERS
                              if x not in self.request_parameters.keys()]
        if missing_parameters:
            errors = [{ "code": 400, "message": f"missing required parameters: {missing_parameters}"}]
            to_return = self.return_data({}, success=False, errors=errors)
            self.write(to_return)
            return

        constructed_nickname = f'{self.wedge_address_configs["NickNamePrefix"]}{self.request_parameters["RequestedNickName"]}'
        nickname_parameters = ChangeEndUserAccountNickname.ParamsObj(
            self.logger,
            self.request_parameters["LoginName"],
            "OnlineBanking",
            self.request_parameters["HostAccountID"],
            self.request_parameters["RequestedNickName"],
            hq_credentials=self.hq_credentials,
        )
        hq_response = await ChangeEndUserAccountNickname.execute(nickname_parameters)
        if not hq_response.success:
            self.logger.debug(f"Failed to update account nickname: {hq_response.error_message}")
            errors = [{"code": 500, "message": f"Internal platform error: failed to update account nickname"}]
            to_return = self.return_data({}, success=False, errors=errors)
            self.write(to_return)
            return

        data = {}
        response = self.return_data(data)
        self.write(response)

To continue the tutorial, click here.