Chaffelson / nipyapi / #2

Committed 25 Mar 2025 10:32AM UTC coverage: 69.867% (+69.9%) from 0.0%

Build # #2

Build Type

push

coveralls-python

Committed by

web-flow

Commit Message

V022release (#384)

* Update history for release

* Bump version: 0.21.0 → 0.22.0

Run Details

1 of 1 new or added line in 1 file covered. (100.0%)

1048 of 1500 relevant lines covered (69.87%)

0.7 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

93.2

/nipyapi/templates.py

# -*- coding: utf-8 -*-

"""
For Managing NiFi Templates in NiFi 1.x
"""

from __future__ import absolute_import

import json
import io
from os import access, R_OK, W_OK
from os.path import isfile, dirname
import logging
import six
import xmltodict
from lxml import etree
import nipyapi

log = logging.getLogger(__name__)

__all__ = [
    "list_all_templates", "get_template_by_name", "deploy_template",
    "upload_template", "create_pg_snippet", "create_template",
    "delete_template", "export_template", 'get_template',
    "load_template_from_xml_file_path", "load_template_from_xml_file_stream",
    "load_template_from_xml_string"
]


def get_template_by_name(name):
    """
    DEPRECATED
    Returns a specific template by name, if it exists.

    Note: This function is replaced by get_template

    Args:
        name (str): The Name of the template, exact match required

    Returns:
        (TemplateEntity)

    """
    out = [
        i for i in
        list_all_templates(native=False)
        if
        name == i.template.name
    ]
    if len(out) == 1:
        return out[0]
    return None


def get_template(identifier, identifier_type='name', greedy=False):
    """
    Filters the list of all Templates for a given string in a given field.
    Note that filters are configured in config.py

    Args:
        identifier (str): The string to filter on
        identifier_type (str): The identifier of the field to filter on
        greedy (bool): True for greedy match, False for exact match

    Returns:
        None for no matches, Single Object for unique match,
        list(Objects) for multiple matches

    """
    assert isinstance(identifier, six.string_types)
    assert identifier_type in ['name', 'id']
    with nipyapi.utils.rest_exceptions():
        obj = nipyapi.templates.list_all_templates(native=False)
    if obj:
        return nipyapi.utils.filter_obj(obj, identifier, identifier_type,
                                        greedy)
    return None


def deploy_template(pg_id, template_id, loc_x=0.0, loc_y=0.0):
    """
    Instantiates a given template request in a given process group

    Args:
        pg_id (str): The UUID of the Process Group to deploy into
        template_id (str): The UUID of the Template to deploy. Note that the
            Template must already be uploaded and available to the target
            Process Group
        loc_x (float): The X coordinate to deploy the Template at. Default(0.0)
        loc_y (float): The X coordinate to deploy the Template at. Default(0.0)

    Returns:
        (FlowEntity): The FlowEntity of the Process Group with the deployed
            template

    """
    nipyapi.utils.validate_templates_version_support()
    with nipyapi.utils.rest_exceptions():
        return nipyapi.nifi.ProcessGroupsApi().instantiate_template(
            id=pg_id,
            body=nipyapi.nifi.InstantiateTemplateRequestEntity(
                origin_x=loc_x,
                origin_y=loc_y,
                template_id=template_id
            )
        )


def create_pg_snippet(pg_id):
    """
    Creates a snippet of the targeted process group, and returns the object
    ready to be turned into a Template

    Args:
        pg_id: UUID of the process Group to snippet

    Returns:
        (SnippetEntity): The Snippet Object
    """
    target_pg = nipyapi.canvas.get_process_group(pg_id, 'id')
    new_snippet_req = nipyapi.nifi.SnippetEntity(
        snippet={
            'processGroups': {
                target_pg.id: target_pg.revision
            },
            'parentGroupId':
                target_pg.component.parent_group_id
        }
    )
    with nipyapi.utils.rest_exceptions():
        snippet_resp = nipyapi.nifi.SnippetsApi().create_snippet(
            new_snippet_req
        )
    return snippet_resp


def create_template(pg_id, name, desc=''):
    """
    Creates a Template from a Process Group

    Args:
        pg_id (str): The UUID of the target Process Group
        name (str): The name for the new Template. Must be unique
        desc (optional[str]): The description for the new Template

    Returns:
        (TemplateEntity): The newly created Template

    """
    nipyapi.utils.validate_templates_version_support()
    snippet = create_pg_snippet(pg_id)
    with nipyapi.utils.rest_exceptions():
        new_template = nipyapi.nifi.CreateTemplateRequestEntity(
            name=str(name),
            description=str(desc),
            snippet_id=snippet.snippet.id
        )
        return nipyapi.nifi.ProcessGroupsApi().create_template(
            id=snippet.snippet.parent_group_id,
            body=new_template
        )


def delete_template(t_id):
    """
    Deletes a Template

    Args:
        t_id (str): UUID of the Template to be deleted

    Returns:
        The updated Template object
    """
    nipyapi.utils.validate_templates_version_support()
    with nipyapi.utils.rest_exceptions():
        return nipyapi.nifi.TemplatesApi().remove_template(id=t_id)


def upload_template(pg_id, template_file):
    """
    Uploads a given template xml from from the file system to the given
    Process Group

    Args:
        pg_id (str): The UUID of the Process Group to upload to
        template_file (str): The path including filename to the template file

    Returns:
        (TemplateEntity): The new Template object

    """
    nipyapi.utils.validate_templates_version_support()
    with nipyapi.utils.rest_exceptions():
        this_pg = nipyapi.canvas.get_process_group(pg_id, 'id')
        assert isinstance(this_pg, nipyapi.nifi.ProcessGroupEntity)
    log.info("Called upload_template against endpoint %s with args %s",
             nipyapi.config.nifi_config.api_client.host, locals())
    # Ensure we are receiving a valid file
    assert isfile(template_file) and access(template_file, R_OK), \
        SystemError("File {0} invalid or unreadable".format(template_file))
    # Test for expected Template XML elements
    tree = etree.parse(template_file)
    root_tag = tree.getroot().tag
    if root_tag != 'template':
        raise TypeError(
            "Expected 'template' as xml root element, got {0} instead."
            "Are you sure this is a Template?"
            .format(root_tag)
        )
    t_name = tree.find('name').text
    with nipyapi.utils.rest_exceptions():
        # For some reason identical code that produces the duplicate error
        # in later versions is going through OK for NiFi-1.1.2
        # The error occurs as normal in Postman, so not sure what's going on
        # Will force this error for consistency until it can be investigated
        if nipyapi.templates.get_template(t_name):
            raise ValueError('A template named {} already exists.'
                             .format(t_name))
        nipyapi.nifi.ProcessGroupsApi().upload_template(
            id=this_pg.id,
            template=template_file
        )
        return nipyapi.templates.get_template(
            tree.find('name').text
        )


def export_template(t_id, output='string', file_path=None):
    """
    Exports a given Template as either a string or a file.

    Note that to reimport the Template it must be a file

    Args:
        t_id (str): The UUID of the Template to export
        output (str): 'string' or 'file' to set the export action
        file_path (Optional [str]): The full path including filename to write
            the Template export to

    Returns:
        (str): A String representation of the exported Template XML. Note
            that this may not be utf-8 encoded.

    """
    nipyapi.utils.validate_templates_version_support()
    assert output in ['string', 'file']
    assert file_path is None or isinstance(file_path, six.string_types)
    template = nipyapi.templates.get_template(t_id, 'id')
    assert isinstance(template, nipyapi.nifi.TemplateEntity)
    obj = nipyapi.nifi.TemplatesApi().export_template(t_id)
    assert isinstance(obj, six.string_types)
    assert obj[0] == '<'
    if output == 'string':
        return obj
    if output == 'file':
        assert access(dirname(file_path), W_OK), \
            "File_path {0} is inaccessible or not writable".format(file_path)
        nipyapi.utils.fs_write(obj, file_path)
        return obj


def list_all_templates(native=True):
    """
    Gets a list of all templates on the canvas

    Returns:
        (list[TemplateEntity]): A list of TemplateEntity's
    """
    nipyapi.utils.validate_templates_version_support()
    with nipyapi.utils.rest_exceptions():
        templates = nipyapi.nifi.FlowApi().get_templates()
    if not native:
        if templates:
            return templates.templates
        return None
    return templates


def load_template_from_xml_file_path(file_path):
    """
    Loads a TemplateEntity from an xml file for a
    given path


    Args:
        file_path (str): path to the xml file

    Returns:
        TemplateEntity
    """
    assert isfile(file_path) and access(file_path, R_OK), \
        SystemError("File {0} invalid or unreadable".format(file_path))
    with io.open(file_path, "r", encoding='utf8') as template_file:
        return load_template_from_xml_file_stream(template_file)


def load_template_from_xml_file_stream(file_stream):
    """
    Loads a TemplateEntity from a template xml file

    Args:
        file_stream (io stream): the xml file stream as returned by open

    Returns:
        TemplateEntity
    """
    return load_template_from_xml_string(file_stream.read())


def load_template_from_xml_string(xml_string):
    """
    Loads a TemplateEntity from xml string, as if
    you had read in the xml file to string

    Args:
        xml_string (str): string of xml

    Returns:
        TemplateEntity
    """
    assert isinstance(xml_string, six.string_types)

    json_string = json.dumps(xmltodict.parse(xml_string))
    unset = False
    if nipyapi.config.nifi_config.api_client is None:
        unset = True
        nipyapi.config.nifi_config.api_client = nipyapi.nifi.ApiClient()

    template_entity = nipyapi.utils.load(json_string,
                                         ('nifi', 'TemplateEntity'))
    if unset:
        nipyapi.config.nifi_config.api_client = None
    return template_entity

1	# -- coding: utf-8 --
2
3	"""
4	For Managing NiFi Templates in NiFi 1.x
5	"""
6
7	from __future__ import absolute_import	1✔
8
9	import json	1✔
10	import io	1✔
11	from os import access, R_OK, W_OK	1✔
12	from os.path import isfile, dirname	1✔
13	import logging	1✔
14	import six	1✔
15	import xmltodict	1✔
16	from lxml import etree	1✔
17	import nipyapi	1✔
18
19	log = logging.getLogger(__name__)	1✔
20
21	__all__ = [	1✔
22	"list_all_templates", "get_template_by_name", "deploy_template",
23	"upload_template", "create_pg_snippet", "create_template",
24	"delete_template", "export_template", 'get_template',
25	"load_template_from_xml_file_path", "load_template_from_xml_file_stream",
26	"load_template_from_xml_string"
27	]
28
29
30	def get_template_by_name(name):	1✔
31	"""
32	DEPRECATED
33	Returns a specific template by name, if it exists.
34
35	Note: This function is replaced by get_template
36
37	Args:
38	name (str): The Name of the template, exact match required
39
40	Returns:
41	(TemplateEntity)
42
43	"""
44	out = [	1✔
45	i for i in
46	list_all_templates(native=False)
47	if
48	name == i.template.name
49	]
50	if len(out) == 1:	1✔
51	return out[0]	1✔
52	return None	×
53
54
55	def get_template(identifier, identifier_type='name', greedy=False):	1✔
56	"""
57	Filters the list of all Templates for a given string in a given field.
58	Note that filters are configured in config.py
59
60	Args:
61	identifier (str): The string to filter on
62	identifier_type (str): The identifier of the field to filter on
63	greedy (bool): True for greedy match, False for exact match
64
65	Returns:
66	None for no matches, Single Object for unique match,
67	list(Objects) for multiple matches
68
69	"""
70	assert isinstance(identifier, six.string_types)	1✔
71	assert identifier_type in ['name', 'id']	1✔
72	with nipyapi.utils.rest_exceptions():	1✔
73	obj = nipyapi.templates.list_all_templates(native=False)	1✔
74	if obj:	1✔
75	return nipyapi.utils.filter_obj(obj, identifier, identifier_type,	1✔
76	greedy)
77	return None	1✔
78
79
80	def deploy_template(pg_id, template_id, loc_x=0.0, loc_y=0.0):	1✔
81	"""
82	Instantiates a given template request in a given process group
83
84	Args:
85	pg_id (str): The UUID of the Process Group to deploy into
86	template_id (str): The UUID of the Template to deploy. Note that the
87	Template must already be uploaded and available to the target
88	Process Group
89	loc_x (float): The X coordinate to deploy the Template at. Default(0.0)
90	loc_y (float): The X coordinate to deploy the Template at. Default(0.0)
91
92	Returns:
93	(FlowEntity): The FlowEntity of the Process Group with the deployed
94	template
95
96	"""
97	nipyapi.utils.validate_templates_version_support()	1✔
98	with nipyapi.utils.rest_exceptions():	1✔
99	return nipyapi.nifi.ProcessGroupsApi().instantiate_template(	1✔
100	id=pg_id,
101	body=nipyapi.nifi.InstantiateTemplateRequestEntity(
102	origin_x=loc_x,
103	origin_y=loc_y,
104	template_id=template_id
105	)
106	)
107
108
109	def create_pg_snippet(pg_id):	1✔
110	"""
111	Creates a snippet of the targeted process group, and returns the object
112	ready to be turned into a Template
113
114	Args:
115	pg_id: UUID of the process Group to snippet
116
117	Returns:
118	(SnippetEntity): The Snippet Object
119	"""
120	target_pg = nipyapi.canvas.get_process_group(pg_id, 'id')	1✔
121	new_snippet_req = nipyapi.nifi.SnippetEntity(	1✔
122	snippet={
123	'processGroups': {
124	target_pg.id: target_pg.revision
125	},
126	'parentGroupId':
127	target_pg.component.parent_group_id
128	}
129	)
130	with nipyapi.utils.rest_exceptions():	1✔
131	snippet_resp = nipyapi.nifi.SnippetsApi().create_snippet(	1✔
132	new_snippet_req
133	)
134	return snippet_resp	1✔
135
136
137	def create_template(pg_id, name, desc=''):	1✔
138	"""
139	Creates a Template from a Process Group
140
141	Args:
142	pg_id (str): The UUID of the target Process Group
143	name (str): The name for the new Template. Must be unique
144	desc (optional[str]): The description for the new Template
145
146	Returns:
147	(TemplateEntity): The newly created Template
148
149	"""
150	nipyapi.utils.validate_templates_version_support()	1✔
151	snippet = create_pg_snippet(pg_id)	1✔
152	with nipyapi.utils.rest_exceptions():	1✔
153	new_template = nipyapi.nifi.CreateTemplateRequestEntity(	1✔
154	name=str(name),
155	description=str(desc),
156	snippet_id=snippet.snippet.id
157	)
158	return nipyapi.nifi.ProcessGroupsApi().create_template(	1✔
159	id=snippet.snippet.parent_group_id,
160	body=new_template
161	)
162
163
164	def delete_template(t_id):	1✔
165	"""
166	Deletes a Template
167
168	Args:
169	t_id (str): UUID of the Template to be deleted
170
171	Returns:
172	The updated Template object
173	"""
174	nipyapi.utils.validate_templates_version_support()	1✔
175	with nipyapi.utils.rest_exceptions():	1✔
176	return nipyapi.nifi.TemplatesApi().remove_template(id=t_id)	1✔
177
178
179	def upload_template(pg_id, template_file):	1✔
180	"""
181	Uploads a given template xml from from the file system to the given
182	Process Group
183
184	Args:
185	pg_id (str): The UUID of the Process Group to upload to
186	template_file (str): The path including filename to the template file
187
188	Returns:
189	(TemplateEntity): The new Template object
190
191	"""
192	nipyapi.utils.validate_templates_version_support()	1✔
193	with nipyapi.utils.rest_exceptions():	1✔
194	this_pg = nipyapi.canvas.get_process_group(pg_id, 'id')	1✔
195	assert isinstance(this_pg, nipyapi.nifi.ProcessGroupEntity)	1✔
196	log.info("Called upload_template against endpoint %s with args %s",	1✔
197	nipyapi.config.nifi_config.api_client.host, locals())
198	# Ensure we are receiving a valid file
199	assert isfile(template_file) and access(template_file, R_OK), \	1✔
200	SystemError("File {0} invalid or unreadable".format(template_file))
201	# Test for expected Template XML elements
202	tree = etree.parse(template_file)	1✔
203	root_tag = tree.getroot().tag	1✔
204	if root_tag != 'template':	1✔
205	raise TypeError(	×
206	"Expected 'template' as xml root element, got {0} instead."
207	"Are you sure this is a Template?"
208	.format(root_tag)
209	)
210	t_name = tree.find('name').text	1✔
211	with nipyapi.utils.rest_exceptions():	1✔
212	# For some reason identical code that produces the duplicate error
213	# in later versions is going through OK for NiFi-1.1.2
214	# The error occurs as normal in Postman, so not sure what's going on
215	# Will force this error for consistency until it can be investigated
216	if nipyapi.templates.get_template(t_name):	1✔
217	raise ValueError('A template named {} already exists.'	1✔
218	.format(t_name))
219	nipyapi.nifi.ProcessGroupsApi().upload_template(	1✔
220	id=this_pg.id,
221	template=template_file
222	)
223	return nipyapi.templates.get_template(	1✔
224	tree.find('name').text
225	)
226
227
228	def export_template(t_id, output='string', file_path=None):	1✔
229	"""
230	Exports a given Template as either a string or a file.
231
232	Note that to reimport the Template it must be a file
233
234	Args:
235	t_id (str): The UUID of the Template to export
236	output (str): 'string' or 'file' to set the export action
237	file_path (Optional [str]): The full path including filename to write
238	the Template export to
239
240	Returns:
241	(str): A String representation of the exported Template XML. Note
242	that this may not be utf-8 encoded.
243
244	"""
245	nipyapi.utils.validate_templates_version_support()	1✔
246	assert output in ['string', 'file']	1✔
247	assert file_path is None or isinstance(file_path, six.string_types)	1✔
248	template = nipyapi.templates.get_template(t_id, 'id')	1✔
249	assert isinstance(template, nipyapi.nifi.TemplateEntity)	1✔
250	obj = nipyapi.nifi.TemplatesApi().export_template(t_id)	1✔
251	assert isinstance(obj, six.string_types)	1✔
252	assert obj[0] == '<'	1✔
253	if output == 'string':	1✔
254	return obj	×
255	if output == 'file':	1✔
256	assert access(dirname(file_path), W_OK), \	1✔
257	"File_path {0} is inaccessible or not writable".format(file_path)
258	nipyapi.utils.fs_write(obj, file_path)	1✔
259	return obj	1✔
260
261
262	def list_all_templates(native=True):	1✔
263	"""
264	Gets a list of all templates on the canvas
265
266	Returns:
267	(list[TemplateEntity]): A list of TemplateEntity's
268	"""
269	nipyapi.utils.validate_templates_version_support()	1✔
270	with nipyapi.utils.rest_exceptions():	1✔
271	templates = nipyapi.nifi.FlowApi().get_templates()	1✔
272	if not native:	1✔
273	if templates:	1✔
274	return templates.templates	1✔
275	return None	×
276	return templates	1✔
277
278
279	def load_template_from_xml_file_path(file_path):	1✔
280	"""
281	Loads a TemplateEntity from an xml file for a
282	given path
283
284
285	Args:
286	file_path (str): path to the xml file
287
288	Returns:
289	TemplateEntity
290	"""
291	assert isfile(file_path) and access(file_path, R_OK), \	1✔
292	SystemError("File {0} invalid or unreadable".format(file_path))
293	with io.open(file_path, "r", encoding='utf8') as template_file:	1✔
294	return load_template_from_xml_file_stream(template_file)	1✔
295
296
297	def load_template_from_xml_file_stream(file_stream):	1✔
298	"""
299	Loads a TemplateEntity from a template xml file
300
301	Args:
302	file_stream (io stream): the xml file stream as returned by open
303
304	Returns:
305	TemplateEntity
306	"""
307	return load_template_from_xml_string(file_stream.read())	1✔
308
309
310	def load_template_from_xml_string(xml_string):	1✔
311	"""
312	Loads a TemplateEntity from xml string, as if
313	you had read in the xml file to string
314
315	Args:
316	xml_string (str): string of xml
317
318	Returns:
319	TemplateEntity
320	"""
321	assert isinstance(xml_string, six.string_types)	1✔
322
323	json_string = json.dumps(xmltodict.parse(xml_string))	1✔
324	unset = False	1✔
325	if nipyapi.config.nifi_config.api_client is None:	1✔
326	unset = True	×
327	nipyapi.config.nifi_config.api_client = nipyapi.nifi.ApiClient()	×
328
329	template_entity = nipyapi.utils.load(json_string,	1✔
330	('nifi', 'TemplateEntity'))
331	if unset:	1✔
332	nipyapi.config.nifi_config.api_client = None	×
333	return template_entity	1✔

Chaffelson / nipyapi / #2

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous