What is the modules folder used for?

The modules folder is where you put all of your Python code. You can use it to manipulate the data stream of any of the sources defined in the config.yaml. You can also manipulate the data that will be transmitted to Veza using the SDK.

• each python file in the module must be referenced in the config.yaml file

• @hooks are defined in the order they are found

• fetch function can be defined in any of the python files, though conventionally we use modules/<appname>.py for this.

  • more about the fetch function at How To Write a Fetch Function?

• make sure you install and import modules required to function

# modules/layer7.py

from oaa.modules.lazy_dicts import lazy_dict_table

def fetch(connection=None, source=None, **kwargs):
    '''
    Fetch the data. This is the function that is called by oaa with the custom
    source defined.

    '''
    layer7 = Layer7API(connection, source)

    if source.params['resource'] == '/api-management/1.0/apis':
        res = layer7.fetch_apis(**source.params)
    else:
        res = layer7.fetch_paginated(**source.params)

    # Use lazy_dict_table for memory-efficient processing
    return lazy_dict_table(res)

The entire modules/layer7.py file can be found below.

# modules/layer7.py
'''
This module pull user, roles, group membership from the Layer7API API
'''

import time
import logging
import petl
from dotenv import load_dotenv
from oaa.modules.params_or_env import params_or_env
from oaa.modules import BaseURLSession
# from unittest.mock import MagicMock
logger = logging.getLogger(__name__)

load_dotenv()

class Layer7API:

    """Docstring for FLoqast. """

    def __init__(self, connection, source):
        """TODO: to be defined1.

        :connection: TODO
        :source: TODO

        """
        self._connection = connection
        self._source = source
        self.base_url = connection.params['layer7_base_uri']
        self.api_key = params_or_env(self._connection, self._source, 'LAYER7_API_TOKEN')
        self._session = None

    @property
    def session(self):
        if not self._session:
            self._session = BaseURLSession(self.base_url)
            self._session.headers.update(self.headers())

        return self._session

    def mkpath(self, path):
        return f'{self.base_url}{path}'

    def headers(self):
        header_dict = {
            'Authorization': f'Bearer {{self.api_key}}',
        }

        return header_dict

    def get(self, url, params=None, **kwargs):
        # self.session

        # Status: 429 means raite limit exhausted. Try again after x seconds
        res = self.session.get(url, params=params)

        if res.status_code == 429:
            # make this more rubust, exponential backoff
            time.sleep(5)

            return self.get(url, params=params, **kwargs)

        return res.json()

    def fetch_apis(self, *args, **kwargs):

        for api in self.fetch_paginated(*args, **kwargs):
            apiUuid = api['uuid']
            api['managers'] = [u['uuid'] for u in self.get(f'/api-management/0.1/permissions/apis/{apiUuid}').json()]
            yield api

    def fetch_paginated(self, resource, query={},
                        resource_key='results',
                        pagination_strategy='link-next',
                        _next=None,
                        **kwargs):

        if resource[0] != '/':
            resource = '/' + resource

        if _next:
            # TODO do we need to get just the path part of the URL?
            resource = _next
        res = self.get(resource,
                       params=query)

        if res is None:
            logger.debug('url: %s', resource)
            logger.error('res: %s', res)

            raise Exception(f'{resource} - {res}')
        res_json = res.json()

        yield from res_json.get(resource_key, [])

        if pagination_strategy == 'link-next':
            _next = res_json.get('_links', {}).get('next', {}).get('href')

            if _next:
                logger.info('fetching more results %s', _next)

                yield from self.fetch_paginated(resource,
                                                resource_key=resource_key,
                                                pagination_strategy=pagination_strategy,
                                                query=query,
                                                _next=_next, **kwargs)
        elif pagination_strategy == 'total-current':
            total_pages = res_json['totalPages']
            current_page = res_json['currentPage']

            if current_page < total_pages:
                query.setdefault('page', 1)
                query['page'] += 1
                yield from self.fetch_paginated(resource,
                                                query=query,
                                                resource_key=resource_key,
                                                pagination_strategy=pagination_strategy,
                                                **kwargs)


# ------------- the custom module fetch function

def fetch(connection=None, source=None, **kwargs):
    '''
    Fetch the data. This is the function that is called by oaa with the custom
    source defined.

    '''
    from oaa.modules.lazy_dicts import lazy_dict_table
    
    layer7 = Layer7API(connection, source)

    if source.params['resource'] == '/api-management/1.0/apis':
        res = layer7.fetch_apis(**source.params)
    else:
        res = layer7.fetch_paginated(**source.params)

    # Use lazy_dict_table for memory-efficient processing
    return lazy_dict_table(res)