Chaffelson / nipyapi / #2

Committed 25 Mar 2025 10:32AM CUT 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

94.44

/nipyapi/parameters.py

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

"""
For Managing NiFi Parameter Contexts
"""

from __future__ import absolute_import
import logging
import six
import nipyapi
from nipyapi.utils import exception_handler, enforce_min_ver
from nipyapi.nifi import ParameterContextEntity, ParameterDTO, \
    ParameterEntity, ParameterContextDTO

log = logging.getLogger(__name__)

__all__ = [
    "list_all_parameter_contexts", "create_parameter_context",
    "delete_parameter_context", "get_parameter_context",
    "update_parameter_context", "prepare_parameter",
    "delete_parameter_from_context", "upsert_parameter_to_context",
    "assign_context_to_process_group",
    "remove_context_from_process_group"
]


def list_all_parameter_contexts():
    """
    Lists all Parameter Contexts available on the Canvas

    Returns:
        list(ParameterContextEntity)
    """
    enforce_min_ver('1.10.0')
    handle = nipyapi.nifi.FlowApi()
    return handle.get_parameter_contexts().parameter_contexts


@exception_handler(404, None)
def get_parameter_context(identifier, identifier_type='name', greedy=True):
    """
    Gets one or more Parameter Contexts matching a given identifier

    Args:
        identifier (str): The Name or ID matching Parameter Context(s)
        identifier_type (str): 'name' or 'id'
        greedy (bool): False for exact match, True for string match

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

    """
    enforce_min_ver('1.10.0')
    assert isinstance(identifier, six.string_types)
    assert identifier_type in ['name', 'id']
    if identifier_type == 'id':
        handle = nipyapi.nifi.ParameterContextsApi()
        out = handle.get_parameter_context(identifier)
    else:
        obj = list_all_parameter_contexts()
        out = nipyapi.utils.filter_obj(
            obj, identifier, identifier_type, greedy=greedy
        )
    return out


def create_parameter_context(name, description=None, parameters=None,
                             inherited_contexts=None):
    """
    Create a new Parameter Context with optional description and
        initial Parameters

    Args:
        name (str): The Name for the new Context
        description (str): An optional description
        parameters (list[ParameterEntity]): A list of prepared Parameters
        inherited_contexts (list[ParameterContextEntity]): A list of
            inherited Parameter Contexts

    Returns:
        (ParameterContextEntity) The New Parameter Context

    """
    enforce_min_ver('1.10.0')
    assert isinstance(name, str)
    assert description is None or isinstance(description, str)
    handle = nipyapi.nifi.ParameterContextsApi()
    inherited = inherited_contexts if inherited_contexts else []
    out = handle.create_parameter_context(
        body=ParameterContextEntity(
            revision=nipyapi.nifi.RevisionDTO(version=0),
            component=ParameterContextDTO(
                name=name,
                description=description,
                parameters=parameters if parameters else [],
                # list() per NiFi Jira 7995
                inherited_parameter_contexts=inherited
                # requires empty list per NiFi Jira 9470
            )
        )
    )
    return out


def update_parameter_context(context):
    """
    Update an already existing Parameter Context

    Args:
        context (ParameterContextEntity): Parameter Context updated
          to be applied
        refresh (bool): Whether to refresh the object before Updating

    Returns:
        (ParameterContextEntity) The updated Parameter Context
    """
    enforce_min_ver('1.10.0')

    def _update_complete(context_id, request_id):
        return nipyapi.nifi.ParameterContextsApi()\
            .get_parameter_context_update(
            context_id, request_id)\
            .request.complete

    if not isinstance(context, ParameterContextEntity):
        raise ValueError("Supplied Parameter Context update should "
                         "be an instance of nipyapi.nifi.ParameterContextDTO")
    handle = nipyapi.nifi.ParameterContextsApi()
    target = get_parameter_context(context.id, identifier_type='id')
    update_request = handle.submit_parameter_context_update(
        context_id=target.id,
        body=ParameterContextEntity(
            id=target.id,
            revision=target.revision,
            component=context.component

        )
    )
    nipyapi.utils.wait_to_complete(
        _update_complete, target.id, update_request.request.request_id,
        nipyapi_delay=1, nipyapi_max_wait=10
    )
    _ = handle.delete_update_request(
        context_id=target.id,
        request_id=update_request.request.request_id
    )
    return get_parameter_context(context.id, identifier_type='id')


def delete_parameter_context(context, refresh=True):
    """
    Removes a Parameter Context

    Args:
        context (ParameterContextEntity): Parameter Context to be deleted
        refresh (bool): Whether to refresh the Context before Deletion

    Returns:
        (ParameterContextEntity) The removed Parameter Context
    """
    enforce_min_ver('1.10.0')
    assert isinstance(context, nipyapi.nifi.ParameterContextEntity)
    handle = nipyapi.nifi.ParameterContextsApi()
    if refresh:
        context = handle.get_parameter_context(context.id)
    return handle.delete_parameter_context(
        id=context.id,
        version=context.revision.version
    )


def prepare_parameter(name, value, description=None, sensitive=False):
    """
    Parses basic inputs into a Parameter object ready for submission

    Args:
        name (str): The Name for the Parameter
        value (str, int, float): The Value for the Parameter
        description (str): Optional Description for the Parameter
        sensitive (bool): Whether to mark the Parameter Value as sensitive

    Returns:
        (ParameterEntity) The ParameterEntity ready for use
    """
    enforce_min_ver('1.10.0')
    assert all(x is None or isinstance(x, str) for x in [name, description])
    out = ParameterEntity(
        parameter=ParameterDTO(
            name=name,
            value=value,
            description=description,
            sensitive=sensitive
        )
    )
    return out


def delete_parameter_from_context(context, parameter_name):
    """
    Delete a specific Parameter from a Parameter Context
    Args:
        context (ParameterContextEntity): The Parameter Context to Update
        parameter_name (str): The Parameter to delete

    Returns:
        (ParameterContextEntity) The updated Parameter Context
    """
    enforce_min_ver('1.10.0')
    context.component.parameters = [
        ParameterEntity(
            parameter=ParameterDTO(
                name=parameter_name
            )
        )
    ]
    return update_parameter_context(
        context=context
    )


def upsert_parameter_to_context(context, parameter):
    """
    Insert or Update Parameter within a Parameter Context

    Args:
        context (ParameterContextEntity): The Parameter Context to Modify
        parameter(ParameterEntity): The ParameterEntity to insert or update

    Returns:
        (ParameterContextEntity) The updated Parameter Context
    """
    enforce_min_ver('1.10.0')
    context.component.parameters = [parameter]
    return update_parameter_context(context=context)


def assign_context_to_process_group(pg, context_id, cascade=False):
    """
    Assigns a given Parameter Context to a specific Process Group
    Optionally cascades down to direct children Process Groups

    Args:
        pg (ProcessGroupEntity): The Process Group to target
        context_id (str): The ID of the Parameter Context
        cascade (bool): Cascade Parameter Context down to child Process Groups?

    Returns:
        (ProcessGroupEntity) The updated Process Group
    """
    assert isinstance(context_id, str)
    if cascade:
        # Update the specified Process Group & all children
        child_pgs = nipyapi.canvas.list_all_process_groups(pg_id=pg.id)
        for child_pg in child_pgs:
            nipyapi.canvas.update_process_group(
                pg=child_pg,
                update={
                    'parameter_context': {
                        'id': context_id
                    }
                }
            )
    return nipyapi.canvas.update_process_group(
        pg=pg,
        update={
            'parameter_context': {
                'id': context_id
            }
        }
    )


def remove_context_from_process_group(pg):
    """
    Clears any Parameter Context from the given Process Group

    Args:
        pg (ProcessGroupEntity): The Process Group to target

    Returns:
        (ProcessGroupEntity) The updated Process Group
    """
    return nipyapi.canvas.update_process_group(
        pg=pg,
        update={
            'parameter_context': {
                'id': None
            }
        }
    )

1	# -- coding: utf-8 --
2
3	"""
4	For Managing NiFi Parameter Contexts
5	"""
6
7	from __future__ import absolute_import	1✔
8	import logging	1✔
9	import six	1✔
10	import nipyapi	1✔
11	from nipyapi.utils import exception_handler, enforce_min_ver	1✔
12	from nipyapi.nifi import ParameterContextEntity, ParameterDTO, \	1✔
13	ParameterEntity, ParameterContextDTO
14
15	log = logging.getLogger(__name__)	1✔
16
17	__all__ = [	1✔
18	"list_all_parameter_contexts", "create_parameter_context",
19	"delete_parameter_context", "get_parameter_context",
20	"update_parameter_context", "prepare_parameter",
21	"delete_parameter_from_context", "upsert_parameter_to_context",
22	"assign_context_to_process_group",
23	"remove_context_from_process_group"
24	]
25
26
27	def list_all_parameter_contexts():	1✔
28	"""
29	Lists all Parameter Contexts available on the Canvas
30
31	Returns:
32	list(ParameterContextEntity)
33	"""
34	enforce_min_ver('1.10.0')	1✔
35	handle = nipyapi.nifi.FlowApi()	1✔
36	return handle.get_parameter_contexts().parameter_contexts	1✔
37
38
39	@exception_handler(404, None)	1✔
40	def get_parameter_context(identifier, identifier_type='name', greedy=True):	1✔
41	"""
42	Gets one or more Parameter Contexts matching a given identifier
43
44	Args:
45	identifier (str): The Name or ID matching Parameter Context(s)
46	identifier_type (str): 'name' or 'id'
47	greedy (bool): False for exact match, True for string match
48
49	Returns:
50	None for no matches, Single Object for unique match,
51	list(Objects) for multiple matches
52
53	"""
54	enforce_min_ver('1.10.0')	1✔
55	assert isinstance(identifier, six.string_types)	1✔
56	assert identifier_type in ['name', 'id']	1✔
57	if identifier_type == 'id':	1✔
58	handle = nipyapi.nifi.ParameterContextsApi()	1✔
59	out = handle.get_parameter_context(identifier)	1✔
60	else:
61	obj = list_all_parameter_contexts()	1✔
62	out = nipyapi.utils.filter_obj(	1✔
63	obj, identifier, identifier_type, greedy=greedy
64	)
65	return out	1✔
66
67
68	def create_parameter_context(name, description=None, parameters=None,	1✔
69	inherited_contexts=None):
70	"""
71	Create a new Parameter Context with optional description and
72	initial Parameters
73
74	Args:
75	name (str): The Name for the new Context
76	description (str): An optional description
77	parameters (list[ParameterEntity]): A list of prepared Parameters
78	inherited_contexts (list[ParameterContextEntity]): A list of
79	inherited Parameter Contexts
80
81	Returns:
82	(ParameterContextEntity) The New Parameter Context
83
84	"""
85	enforce_min_ver('1.10.0')	1✔
86	assert isinstance(name, str)	1✔
87	assert description is None or isinstance(description, str)	1✔
88	handle = nipyapi.nifi.ParameterContextsApi()	1✔
89	inherited = inherited_contexts if inherited_contexts else []	1✔
90	out = handle.create_parameter_context(	1✔
91	body=ParameterContextEntity(
92	revision=nipyapi.nifi.RevisionDTO(version=0),
93	component=ParameterContextDTO(
94	name=name,
95	description=description,
96	parameters=parameters if parameters else [],
97	# list() per NiFi Jira 7995
98	inherited_parameter_contexts=inherited
99	# requires empty list per NiFi Jira 9470
100	)
101	)
102	)
103	return out	1✔
104
105
106	def update_parameter_context(context):	1✔
107	"""
108	Update an already existing Parameter Context
109
110	Args:
111	context (ParameterContextEntity): Parameter Context updated
112	to be applied
113	refresh (bool): Whether to refresh the object before Updating
114
115	Returns:
116	(ParameterContextEntity) The updated Parameter Context
117	"""
118	enforce_min_ver('1.10.0')	1✔
119
120	def _update_complete(context_id, request_id):	1✔
121	return nipyapi.nifi.ParameterContextsApi()\	1✔
122	.get_parameter_context_update(
123	context_id, request_id)\
124	.request.complete
125
126	if not isinstance(context, ParameterContextEntity):	1✔
127	raise ValueError("Supplied Parameter Context update should "	×
128	"be an instance of nipyapi.nifi.ParameterContextDTO")
129	handle = nipyapi.nifi.ParameterContextsApi()	1✔
130	target = get_parameter_context(context.id, identifier_type='id')	1✔
131	update_request = handle.submit_parameter_context_update(	1✔
132	context_id=target.id,
133	body=ParameterContextEntity(
134	id=target.id,
135	revision=target.revision,
136	component=context.component
137
138	)
139	)
140	nipyapi.utils.wait_to_complete(	1✔
141	_update_complete, target.id, update_request.request.request_id,
142	nipyapi_delay=1, nipyapi_max_wait=10
143	)
144	_ = handle.delete_update_request(	1✔
145	context_id=target.id,
146	request_id=update_request.request.request_id
147	)
148	return get_parameter_context(context.id, identifier_type='id')	1✔
149
150
151	def delete_parameter_context(context, refresh=True):	1✔
152	"""
153	Removes a Parameter Context
154
155	Args:
156	context (ParameterContextEntity): Parameter Context to be deleted
157	refresh (bool): Whether to refresh the Context before Deletion
158
159	Returns:
160	(ParameterContextEntity) The removed Parameter Context
161	"""
162	enforce_min_ver('1.10.0')	1✔
163	assert isinstance(context, nipyapi.nifi.ParameterContextEntity)	1✔
164	handle = nipyapi.nifi.ParameterContextsApi()	1✔
165	if refresh:	1✔
166	context = handle.get_parameter_context(context.id)	1✔
167	return handle.delete_parameter_context(	1✔
168	id=context.id,
169	version=context.revision.version
170	)
171
172
173	def prepare_parameter(name, value, description=None, sensitive=False):	1✔
174	"""
175	Parses basic inputs into a Parameter object ready for submission
176
177	Args:
178	name (str): The Name for the Parameter
179	value (str, int, float): The Value for the Parameter
180	description (str): Optional Description for the Parameter
181	sensitive (bool): Whether to mark the Parameter Value as sensitive
182
183	Returns:
184	(ParameterEntity) The ParameterEntity ready for use
185	"""
186	enforce_min_ver('1.10.0')	1✔
187	assert all(x is None or isinstance(x, str) for x in [name, description])	1✔
188	out = ParameterEntity(	1✔
189	parameter=ParameterDTO(
190	name=name,
191	value=value,
192	description=description,
193	sensitive=sensitive
194	)
195	)
196	return out	1✔
197
198
199	def delete_parameter_from_context(context, parameter_name):	1✔
200	"""
201	Delete a specific Parameter from a Parameter Context
202	Args:
203	context (ParameterContextEntity): The Parameter Context to Update
204	parameter_name (str): The Parameter to delete
205
206	Returns:
207	(ParameterContextEntity) The updated Parameter Context
208	"""
209	enforce_min_ver('1.10.0')	1✔
210	context.component.parameters = [	1✔
211	ParameterEntity(
212	parameter=ParameterDTO(
213	name=parameter_name
214	)
215	)
216	]
217	return update_parameter_context(	1✔
218	context=context
219	)
220
221
222	def upsert_parameter_to_context(context, parameter):	1✔
223	"""
224	Insert or Update Parameter within a Parameter Context
225
226	Args:
227	context (ParameterContextEntity): The Parameter Context to Modify
228	parameter(ParameterEntity): The ParameterEntity to insert or update
229
230	Returns:
231	(ParameterContextEntity) The updated Parameter Context
232	"""
233	enforce_min_ver('1.10.0')	1✔
234	context.component.parameters = [parameter]	1✔
235	return update_parameter_context(context=context)	1✔
236
237
238	def assign_context_to_process_group(pg, context_id, cascade=False):	1✔
239	"""
240	Assigns a given Parameter Context to a specific Process Group
241	Optionally cascades down to direct children Process Groups
242
243	Args:
244	pg (ProcessGroupEntity): The Process Group to target
245	context_id (str): The ID of the Parameter Context
246	cascade (bool): Cascade Parameter Context down to child Process Groups?
247
248	Returns:
249	(ProcessGroupEntity) The updated Process Group
250	"""
251	assert isinstance(context_id, str)	1✔
252	if cascade:	1✔
253	# Update the specified Process Group & all children
254	child_pgs = nipyapi.canvas.list_all_process_groups(pg_id=pg.id)	×
255	for child_pg in child_pgs:	×
256	nipyapi.canvas.update_process_group(	×
257	pg=child_pg,
258	update={
259	'parameter_context': {
260	'id': context_id
261	}
262	}
263	)
264	return nipyapi.canvas.update_process_group(	1✔
265	pg=pg,
266	update={
267	'parameter_context': {
268	'id': context_id
269	}
270	}
271	)
272
273
274	def remove_context_from_process_group(pg):	1✔
275	"""
276	Clears any Parameter Context from the given Process Group
277
278	Args:
279	pg (ProcessGroupEntity): The Process Group to target
280
281	Returns:
282	(ProcessGroupEntity) The updated Process Group
283	"""
284	return nipyapi.canvas.update_process_group(	1✔
285	pg=pg,
286	update={
287	'parameter_context': {
288	'id': None
289	}
290	}
291	)

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