Skip to content

Instantly share code, notes, and snippets.

@shantanoo-desai

shantanoo-desai/test_upload_flow.yml

Last active August 2, 2023 14:43
Show Gist options
  • Save shantanoo-desai/54d58db56b509d3632105f6d05d96f16 to your computer and use it in GitHub Desktop.
Save shantanoo-desai/54d58db56b509d3632105f6d05d96f16 to your computer and use it in GitHub Desktop.
Download ZIP
Custom ansible Plugin to update nodered flow
Raw
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
Raw
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()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment