Custom ansible Plugin to update nodered flow

test_upload_flow.yml

---
- hosts: localhost
  gather_facts: false
  tasks:
    - name: test upload flow to nodered
      nodered_flow:
        nodered_url: http://localhost/nodered
        nodered_user: admin
        nodered_password: testnodered
        path: "{{ playbook_dir}}/flows_test.json"
        influxdb_user: admin
        influxdb_password: test
        influxdb_database: test
        # # influxdb_token: testtoken
        postgres_user: admin
        postgres_password: quest
        postgres_database: qdb
        postgres_db_type: questdb

update_flow.py

from __future__ import absolute_import, division, print_function

DOCUMENTATION='''
---
module: nodered_flow
author:
  - Shantanoo 'Shan' Desai
version_added: "0.2.1"
short_description: Manage Node-RED active flow
description:
  - upload Node-RED flows vai API.
options:
  state:
    description:
      - The state of the flow.
    choices: [started, stopped]
    default: started
    type: str
  path:
    description:
      - The path to the json file containing the Node-RED flows to upload
    version_added: "0.2.1"
    type: str
'''

EXAMPLES='''
- hosts: localhost
  gather_facts: false
  tasks:
    - name: test upload flow to nodered
      nodered_flow:
        nodered_url: http://localhost/nodered
        nodered_user: admin
        nodered_password: testnodered
        path: "{{ playbook_dir}}/path/to/flows.json"
        influxdb_user: admin
        influxdb_password: test
        influxdb_database: test
        influxdb_token: testtoken
        postgres_user: admin
        postgres_password: quest
        postgres_database: qdb
        postgres_db_type: questdb
'''

import json
from ansible.module_utils.basic import AnsibleModule
from ansible.module_utils.urls import fetch_url, url_argument_spec
from ansible.module_utils._text import to_native
from ansible.module_utils._text import to_text

__metaclass__ = type

## URL Argument Spec for Node-RED Module
def nodered_argument_spec():
    argument_spec = url_argument_spec()

    del argument_spec['force']
    del argument_spec['force_basic_auth']
    del argument_spec['http_agent']
    if "use_gssapi" in argument_spec:
        del argument_spec['use_gssapi']
    
    argument_spec.update(
        url=dict(aliases=['nodered_url'], type='str', required=True),
        url_username=dict(aliases=['nodered_user'], required=True),
        url_password=dict(aliases=['nodered_password'], no_log=True, required=True)
    )

    return argument_spec


class NodeRedAPIException(Exception):
    pass

class NodeRedInterface(object):

    def __init__(self, module):
        self._module = module
        self.nodered_url = module.params.get('url')
        self.headers = {"Content-Type": "application/json"}
        self._token = ''
    
    def _send_request(self, url, data=None, headers=None, method="POST"):
        if data is not None:
            data = json.dumps(data, sort_keys=True)
        if not headers:
            headers = []
        
        full_url ="{nodered_url}{path}".format(nodered_url=self.nodered_url, path=url)
        resp, info = fetch_url(self._module, full_url, data=data, headers=headers, method=method)

        status_code = info["status"]
        
        if status_code == 404:
            return None
        elif status_code == 400:
            self._module.fail_json(failed=True, msg="Invalid API Version '%s' on '%s': '%s'" %(method, full_url, resp.read()))
        elif status_code == 401:
            self._module.fail_json(failed=True, msg="Not Authorized '%s' on '%s'" %(method, full_url))
        elif status_code == 409:
            self._module.fail_json(failed=True, msg="Version Mismatch '%s' on '%s'" %(method, full_url))
        elif status_code == 200:
            if url == '/auth/revoke':
                return None
            # For Node-RED-API-Version: v2
            return self._module.from_json(resp.read())
        elif status_code == 204:
            # For Node-RED-API-Version: v1
            return None
        self._module.fail_json(failed=True, msg="NodeRED API answered with HTTP %d for url %s and data %s" %(status_code, url, data))
            
    
    def _get_auth_token(self):
        url = '/auth/token'
        data = {'client_id': 'node-red-admin', 'grant_type': 'password', 'scope': '*'}
        data['username'] = self._module.params.get('url_username')
        data['password'] = self._module.params.get('url_password')
        response = self._send_request(url, data=data, headers=self.headers)

        # Expected Response from `/auth/token` API:
        # {"access_token": "TOKEN", "expires_in": 604800, "token_type": "Bearer"}
        self.headers['Authorization'] = "%s %s" % (response['token_type'], response['access_token'])
        self._token = response['access_token']
    
    def _revoke_auth_token(self):
        url = '/auth/revoke'
        if self._token != '':
            data = {'token': self._token}
            self._token = ''
            self._send_request(url, data=data, headers=self.headers)

    def upload_flow(self, data):
        url = '/flows'
        # Get Access Token
        self._get_auth_token()
        self.headers['Node-RED-deployment-Type'] = 'full'
        # Upload Flow
        response = self._send_request(url=url,data=data, headers=self.headers)
        # Revoke Token
        self._revoke_auth_token()

        return None


def insert_additional_creds(data):
    payload = {}
    # print(data)
    if data.get('path'):
        try:
            with open(data['path'], 'r', encoding='utf-8') as flow_json_file:
                payload = json.load(flow_json_file)
        except Exception as e:
            raise NodeRedAPIException("Cant Load JSON File %s" % to_native(e))

    if data.get('influxdb_user'):
        for each_node in payload:
            if each_node['type'] == 'influxdb':
                if 'credentials' not in each_node:
                    if data.get('influxdb_password'):
                        
                        each_node['credentials'] = {'username': '', 'password': ''}
                        
                        each_node['credentials']['username'] = data['influxdb_user']
                        each_node['credentials']['password'] = data['influxdb_password']
                    else:
                        each_node['credentials']= {'username': '', 'password': '', 'token': ''}
                        each_node['credentials']['username'] = data['influxdb_user']
                        each_node['credentials']['token'] = data['influxdb_token']
        
    if data.get('postgres_user'):
        for each_node in payload:
            if each_node['type'] == 'postgreSQLConfig':
                each_node['user'] = data['postgres_user']
                each_node['password'] = data['postgres_password']
                each_node['database'] = data['postgres_database']
                if data['postgres_db_type'] == 'timescaledb':
                    each_node['host'] = 'komponist_timescaledb'
                else:
                    each_node['host'] = 'komponist_questdb'
                    each_node['port'] = '8812'
    return payload

def setup_module_object():
    argument_spec = nodered_argument_spec()

    argument_spec.update(
        path=dict(type='str', required=True),
        influxdb_user=dict(type='str', required=False),
        influxdb_password=dict(type='str', required=False, no_log=True),
        influxdb_token=dict(type='str', required=False, no_log=True),
        influxdb_database=dict(type='str', required=False),
        postgres_user=dict(type='str', required=False),
        postgres_database=dict(type='str', required=False, no_log=True),
        postgres_password=dict(type='str', required=False, no_log=True),
        postgres_db_type=dict(type='str', choices=['questdb', 'timescaledb'], required=False)
    )

    module = AnsibleModule(
        argument_spec=argument_spec,
        supports_check_mode=False,
        mutually_exclusive=[('influxdb_password', 'influxdb_token')],
    )

    return module

def main():
    module = setup_module_object()
    
    nodered_iface = NodeRedInterface(module=module)

    flow_with_creds = insert_additional_creds(module.params)
    print(flow_with_creds)


    result = nodered_iface.upload_flow(flow_with_creds)
    
    if result is None:
        module.exit_json(failed=False)

if __name__ == "__main__":
    main()