API Files

Requiring External Requests

Some plugins are required to make external API calls in order to function. To assist in this matter, creating an api.py file can be used to make calls to an external API that are utilised by the plugin and also be used in parsing the responses from the API.

Typically, it is best to organize utility files such that the api.py file is inside a new directory called utils within the plugin structure to maintain a standardized and organized framework. This helps the code within this file to be easily accessible by any action, trigger, or connection.

example-plugin/example-plugin/utils/api.py


import requests
from insightconnect_plugin_runtime.exceptions import PluginException
import json
 
class ExamplePluginAPI:
    def __init__(self, logger):
        self.logger = logger
        self.url = "https://www.example-url.com/api/v1"
 
    def _call_api(self, method, url, params=None, json_data=None):
        try:
            # Make the API call
            response = requests.request(method, url, json=json_data, params=params)
            
            # Usually, 204 returns None if successful
            if response.status_code == 204:
                return None
            # All other 200 codes, return response.json()
            if response.status_code in range(200, 300):
                return response.json()
            # Handle errors and raise PluginException
            if response.status_code == 404:
                raise PluginException(preset=PluginException.Preset.NOT_FOUND)
            if response.status_code == 500:
                raise PluginException(preset=PluginException.Preset.SERVER_ERROR)
            if response.status_code == 503:
                raise PluginException(preset=PluginException.Preset.SERVICE_UNAVAILABLE)
            if response.status_code >= 400:
                response_data = response.json()
                raise PluginException(preset=PluginException.Preset.UNKNOWN, data=response_data.message)
        # Handle JSON errors
        except json.decoder.JSONDecodeError as exception:
            self.logger.info(f"Invalid JSON: {exception}")
            raise PluginException(preset=PluginException.Preset.INVALID_JSON, data=response.text)
        # Handle HTTP errors
        except requests.exceptions.HTTPError as exception:
            self.logger.info(f"Call to Example API failed: {exception}")
            raise PluginException(preset=PluginException.Preset.UNKNOWN, data=response.text)
 
 
    def list_users(self) -> str:
        return self._call_api("GET", "users/list")
    
    def add_user(self, params):
        return self._call_api("POST", "users/add", params=params)
 
    # etc

This example uses the Requests module to make an external API call, attempts to handle any errors arising from unexpected responses, and returns the result of the HTTP request.

In this file we define a class to contain the functions and logic necessary to interact with our chosen API.

By creating individual functions for each request such as the list_users method, which calls upon the _call_api function, we can simplify our API access logic.

Responses are also handled using PluginException, InsightConnects custom exception type which allows for a standardized response for plugin users and debugging.

Using the API File to Send Requests

To utilize these requests we can use the created API class as a property of the Connection class within the connection.py file created when the insight-plugin create command was first run.


import insightconnect_plugin_runtime
from .schema import ConnectionSchema, Input
 
# Custom imports below
from example_plugin.util.api import ExampleAPI
 
 
class Connection(insightconnect_plugin_runtime.Connection):
    def __init__(self):
        super(self.__class__, self).__init__(input=ConnectionSchema())
        self.client = None
 
    def connect(self, params):
        self.logger.info("Connect: Connecting...")
        
        # Retrieve all inputs
        account_id = params.get(Input.ACCOUNT_ID)
        client_id = params.get(Input.CLIENT_ID)
        client_secret = params.get(Input.CLIENT_SECRET, {}).get("secretKey")  
        
        # Connect to API
        self.example_api = ExampleAPI(
            logger=self.logger,
            account_id=account_id,
            client_id=client_id,
            client_secret=client_secret,
        )
 
    def test(self):
        return {"success": True}

In this example, we’ve assigned our ExampleAPI class to the client property of our connection.

We’re now able to access these API contact methods from our action files run method and return the response to complete the plugin action.


import insightconnect_plugin_runtime
from .schema import GetUsersInput, GetUsersOutput, Input, Output, Component
 
 
class GetUsers(insightconnect_plugin_runtime.Action):
 
    def __init__(self):
        super(self.__class__, self).__init__(
                name="get_user",
                description=Component.DESCRIPTION,
                input=GetUserInput(),
                output=GetUserOutput())
 
    def run(self, params={}):
        # Get input from schema
        user_id = params.get(Input.USER_ID)
        # Call function from api.py file
        users = self.connection.client.list_user_by_id(user_id)
        # Map schema to output
        return {Output.USERS: users}

Individual plugin authors are responsible for writing the api.py code.