Canvas

Canvas operations and flow management

For interactions with the NiFi Canvas.

nipyapi.canvas.create_connection(source, target, relationships=None, name=None)[source]

Creates a connection between two objects for the given relationships

Parameters:

  • source – Object to initiate the connection, e.g. ProcessorEntity

  • target – Object to terminate the connection, e.g. FunnelEntity

  • relationships (list) – list of strings of relationships to connect, may be collected from the object ‘relationships’ property (optional)

  • name (str) – Defaults to None, String of Name for Connection (optional)

Returns:

for the created connection

Return type:

ConnectionEntity

nipyapi.canvas.create_controller(parent_pg, controller, name=None)[source]

Creates a new Controller Service in a given Process Group of the given Controller type, with the given Name.

Parameters:

  • parent_pg (ProcessGroupEntity) – Target Parent PG

  • controller (DocumentedTypeDTO) – Type of Controller to create, found via the list_all_controller_types method

  • name (str, optional) – Name for the new Controller as a String

Returns:

The created controller service

Return type:

ControllerServiceEntity

nipyapi.canvas.create_funnel(pg_id, position=None)[source]

Creates a Funnel Object

Parameters:

  • pg_id (str) – ID of the parent Process Group

  • position (tuple[int, int]) – Position on canvas

Returns:

Created Funnel

Return type:

FunnelEntity

nipyapi.canvas.create_port(pg_id, port_type, name, state, position=None)[source]

Creates a new input or output port of given characteristics

Parameters:

  • pg_id (str) – ID of the parent Process Group

  • port_type (str) – Either of INPUT_PORT or OUTPUT_PORT

  • name (str) – optional, Name to assign to the port

  • state (str) – One of RUNNING, STOPPED, DISABLED

  • position (tuple) – optional, tuple of ints like (400, 400)

Returns:

(PortEntity) of the created port

nipyapi.canvas.create_process_group(parent_pg, new_pg_name, location, comment='')[source]

Creates a new Process Group with the given name under the provided parent Process Group at the given Location

Parameters:

  • parent_pg (ProcessGroupEntity) – The parent Process Group to create the new process group in

  • new_pg_name (str) – The name of the new Process Group

  • location (tuple[x, y]) – the x,y coordinates to place the new Process Group under the parent

  • comment (str) – Entry for the Comments field

Returns:

The new Process Group

Return type:

ProcessGroupEntity

nipyapi.canvas.create_processor(parent_pg, processor, location, name=None, config=None)[source]

Instantiates a given processor on the canvas

Parameters:

  • parent_pg (ProcessGroupEntity) – The parent Process Group

  • processor (DocumentedTypeDTO) – The abstract processor type object to be instantiated

  • location (tuple[x, y]) – The location coordinates

  • name (Optional [str]) – The name for the new Processor

  • config (Optional [ProcessorConfigDTO]) – A configuration object for the new processor

Returns:

The new Processor

Return type:

ProcessorEntity

nipyapi.canvas.create_remote_process_group(target_uris, transport='RAW', pg_id='root', position=None)[source]

Creates a new Remote Process Group with given parameters

Parameters:

  • target_uris (str) – Comma separated list of target URIs

  • transport (str) – optional, RAW or HTTP

  • pg_id (str) – optional, UUID of parent Process Group for remote process group

  • position (tuple) – optional, tuple of location ints

Returns:

(RemoteProcessGroupEntity)

nipyapi.canvas.delete_connection(connection, purge=False)[source]

Deletes a connection, optionally purges it first

Parameters:

  • connection (ConnectionEntity) – Connection to delete

  • purge (bool) – True to Purge, Defaults to False

Returns:

the modified Connection

Return type:

ConnectionEntity

nipyapi.canvas.delete_controller(controller, force=False)[source]

Delete a Controller service, with optional prejudice

Parameters:

  • controller (ControllerServiceEntity) – Target Controller to delete

  • force (bool) – True to attempt Disable the Controller before deletion

Returns:

(ControllerServiceEntity)

nipyapi.canvas.delete_funnel(funnel, refresh=True)[source]

Deletes a Funnel Object

Parameters:

  • funnel (FunnelEntity) – The Funnel to delete

  • refresh (bool) – Whether to refresh the object state before execution

Returns:

Deleted FunnelEntity reference

Return type:

FunnelEntity

nipyapi.canvas.delete_port(port)[source]

Deletes a given port from the canvas if possible

nipyapi.canvas.delete_process_group(process_group, force=False, refresh=True)[source]

Deletes a given Process Group, with optional prejudice.

Parameters:

  • process_group (ProcessGroupEntity) – The target Process Group

  • force (bool) – Stop, purge and clean the target Process Group before deletion. Experimental.

  • refresh (bool) – Whether to refresh the state first

Returns:

The updated object state

Return type:

(ProcessGroupEntity

nipyapi.canvas.delete_processor(processor, refresh=True, force=False)[source]

Deletes a Processor from the canvas, with optional prejudice.

Parameters:

  • processor (ProcessorEntity) – The processor to delete

  • refresh (bool) – Whether to refresh the Processor state before action

  • force (bool) – Whether to stop, purge and remove connections to the Processor before deletion. Behavior may change in future releases.

Returns:

The updated ProcessorEntity

Return type:

ProcessorEntity

nipyapi.canvas.delete_remote_process_group(rpg, refresh=True)[source]

Deletes a given remote process group

Parameters:
Returns:

(RemoteProcessGroupEntity)

nipyapi.canvas.get_bulletin_board()[source]

Retrieves the bulletin board object

Returns:

The native datatype BulletinBoard object

Return type:

(BulletinBoardEntity)

nipyapi.canvas.get_bulletins()[source]

Retrieves current bulletins (alerts) from the Flow Canvas

Returns:

The native datatype containing a list

Return type:

(ControllerBulletinsEntity)

of bulletins

nipyapi.canvas.get_component_connections(component)[source]

Returns list of Connections related to a given Component, e.g. Processor

Parameters:

component – Component Object to filter by, e.g. a ProcessorEntity

Returns:

List of ConnectionEntity Objects

Return type:

(list)

nipyapi.canvas.get_controller(identifier, identifier_type='name', bool_response=False, include_reporting_tasks=True)[source]

Retrieve a given Controller

Parameters:

  • identifier (str) – ID or Name of a Controller to find

  • identifier_type (str) – ‘id’ or ‘name’, defaults to name

  • bool_response (bool) – If True, will return False if the Controller is not found - useful when testing for deletion completion

  • include_reporting_tasks (bool) – If True, will include Reporting Tasks in the search

Returns:

nipyapi.canvas.get_flow(pg_id='root')[source]

Returns information about a Process Group and flow.

This surfaces the native implementation, for the recursed implementation see ‘recurse_flow’

Parameters:

pg_id (str) – id of the Process Group to retrieve, defaults to the root process group if not set

Returns:

The Process Group object

Return type:

ProcessGroupFlowEntity

nipyapi.canvas.get_funnel(funnel_id)[source]

Gets a given Funnel by ID

nipyapi.canvas.get_pg_parents_ids(pg_id)[source]

Retrieve the ids of the parent Process Groups.

Parameters:

pg_id (str) – Process group id

Returns:

(list) List of ids of the input PG parents

nipyapi.canvas.get_process_group(identifier, identifier_type='name', greedy=True)[source]

Filters the list of all process groups against a given identifier string occurring in a given identifier_type field.

Parameters:

  • identifier (str) – the string to filter the list for

  • identifier_type (str) – the field to filter on, set in config.py

  • greedy (bool) – True for partial match, False for exact match

Returns:

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

nipyapi.canvas.get_process_group_status(pg_id='root', detail='names')[source]

Returns an entity containing the status of the Process Group. Optionally may be configured to return a simple dict of name:id pairings

Note that there is also a ‘process group status’ command available, but it returns a subset of this data anyway, and this call is more useful

Parameters:

  • pg_id (str) – The UUID of the Process Group

  • detail (str) – ‘names’ or ‘all’; whether to return a simple dict of name:id pairings, or the full details. Defaults to ‘names’

Returns:

The Process Group Entity including the status

Return type:

ProcessGroupEntity

nipyapi.canvas.get_processor(identifier, identifier_type='name', greedy=True)[source]

Filters the list of all Processors against the given identifier string in the given identifier_type field

Parameters:

  • identifier (str) – The String to filter against

  • identifier_type (str) – The field to apply the filter to. Set in config.py

  • greedy (bool) – Whether to exact match (False) or partial match (True)

Returns:

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

nipyapi.canvas.get_processor_type(identifier, identifier_type='name', greedy=True)[source]

Gets the abstract object describing a Processor, or list thereof

Parameters:

  • identifier (str) – the string to filter the list for

  • identifier_type (str) – the field to filter on, set in config.py

  • greedy (bool) – False for exact match, True for greedy match

Returns:

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

nipyapi.canvas.get_remote_process_group(rpg_id, summary=False)[source]

Fetch a remote process group object, with optional summary of just ports

nipyapi.canvas.get_root_pg_id()[source]

Convenience function to return the UUID of the Root Process Group

Returns (str): The UUID of the root PG

nipyapi.canvas.get_variable_registry(process_group, ancestors=True)[source]

Gets the contents of the variable registry attached to a Process Group

Parameters:

  • process_group (ProcessGroupEntity) – The Process Group to retrieve the Variable Registry from

  • ancestors (bool) – Whether to include the Variable Registries from child Process Groups

Returns:

The Variable Registry

Return type:

VariableRegistryEntity

nipyapi.canvas.list_all_by_kind(kind, pg_id='root', descendants=True)[source]

Retrieves a list of all instances of a supported object type

Parameters:

  • kind (str) – one of input_ports, output_ports, funnels, controllers, connections, remote_process_groups

  • pg_id (str) – optional, ID of the Process Group to use as search base

  • descendants (bool) – optional, whether to collect child group info

Returns:

list of the Entity type of the kind, or single instance, or None

nipyapi.canvas.list_all_connections(pg_id='root', descendants=True)[source]

Lists all connections for a given Process Group ID

Parameters:

  • pg_id (str) – ID of the Process Group to retrieve Connections from

  • descendants (bool) – True to recurse child PGs, False to not

Returns:

List of ConnectionEntity objects

Return type:

(list)

nipyapi.canvas.list_all_controller_types()[source]

Lists all Controller Service types available on the environment

Returns:

list(DocumentedTypeDTO)

nipyapi.canvas.list_all_controllers(pg_id='root', descendants=True, include_reporting_tasks=False)[source]

Lists all controllers under a given Process Group, defaults to Root. Optionally recurses all child Process Groups as well.

Parameters:

  • pg_id (str) – String of the ID of the Process Group to list from

  • descendants (bool) – True to recurse child PGs, False to not

  • include_reporting_tasks (bool) – True to include Reporting Tasks, False to not

Returns:

None, ControllerServiceEntity, or list(ControllerServiceEntity)

nipyapi.canvas.list_all_funnels(pg_id='root', descendants=True)[source]

Convenience wrapper for list_all_by_kind for funnels

nipyapi.canvas.list_all_input_ports(pg_id='root', descendants=True)[source]

Convenience wrapper for list_all_by_kind for input ports

nipyapi.canvas.list_all_output_ports(pg_id='root', descendants=True)[source]

Convenience wrapper for list_all_by_kind for output ports

nipyapi.canvas.list_all_process_groups(pg_id='root')[source]

Returns a flattened list of all Process Groups on the canvas. Potentially slow if you have a large canvas.

Note that the ProcessGroupsApi().get_process_groups(pg_id) command only provides the first layer of pgs, whereas this trawls the entire canvas

Parameters:

pg_id (str) – The UUID of the Process Group to start from, defaults to the Canvas root

Returns:

list[ProcessGroupEntity]

nipyapi.canvas.list_all_processor_types()[source]

Produces the list of all available processor types in the NiFi instance

Returns:

A native datatype containing the processors list

Return type:

list(ProcessorTypesEntity)

nipyapi.canvas.list_all_processors(pg_id='root')[source]

Returns a flat list of all Processors under the provided Process Group

Parameters:

pg_id (str) – The UUID of the Process Group to start from, defaults to the Canvas root

Returns:

list[ProcessorEntity]

nipyapi.canvas.list_all_remote_process_groups(pg_id='root', descendants=True)[source]

Convenience wrapper for list_all_by_kind for remote process groups

nipyapi.canvas.list_invalid_processors(pg_id='root', summary=False)[source]

Returns a flattened list of all Processors with Invalid Statuses

Parameters:

  • pg_id (str) – The UUID of the Process Group to start from, defaults to the Canvas root

  • summary (bool) – True to return just the list of relevant properties per Processor, False for the full listing

Returns:

list[ProcessorEntity]

nipyapi.canvas.list_sensitive_processors(pg_id='root', summary=False)[source]

Returns a flattened list of all Processors on the canvas which have sensitive properties that would need to be managed during deployment

Parameters:

  • pg_id (str) – The UUID of the Process Group to start from, defaults to the Canvas root

  • summary (bool) – True to return just the list of relevant properties per Processor, False for the full listing

Returns:

list[ProcessorEntity] or list(dict)

nipyapi.canvas.purge_connection(con_id)[source]

EXPERIMENTAL Drops all FlowFiles in a given connection. Waits until the action is complete before returning.

Note that if upstream component isn’t stopped, more data may flow into the connection after this action.

Parameters:

con_id (str) – The UUID of the Connection to be purged

Returns:

The status reporting object for the drop request.

Return type:

DropRequestEntity

nipyapi.canvas.purge_process_group(process_group, stop=False)[source]

EXPERIMENTAL Purges the connections in a given Process Group of FlowFiles, and optionally stops it first

Parameters:

  • process_group (ProcessGroupEntity) – Target Process Group

  • stop (Optional [bool]) – Whether to stop the Process Group before action

Returns:

True|False}]): Result set. A list of Dicts of

Return type:

(list[dict{ID

Connection IDs mapped to True or False for success of each connection

nipyapi.canvas.recurse_flow(pg_id='root')[source]

Returns information about a Process Group and all its Child Flows.

Recurses the child flows by appending each process group with a ‘nipyapi_extended’ parameter which contains the child process groups, etc.

Note: This previously used actual recursion which broke on large NiFi environments, we now use a task/list update approach.

Parameters:

pg_id (str) – The Process Group UUID

Returns:

enriched NiFi Flow object

Return type:

ProcessGroupFlowEntity

nipyapi.canvas.schedule_all_controllers(pg_id, scheduled)[source]

Enable or Disable all Controller Services in a Process Group.

Uses NiFi’s native bulk controller service activation API which handles all descendant controller services automatically.

Parameters:

  • pg_id (str) – The UUID of the Process Group

  • scheduled (bool) – True to enable, False to disable

Returns:

The result of the operation

Return type:

ActivateControllerServicesEntity

nipyapi.canvas.schedule_components(pg_id, scheduled, components=None)[source]

Changes the scheduled target state of a list of components within a given Process Group.

Note that this does not guarantee that components will be Started or Stopped afterwards, merely that they will have their scheduling updated.

Parameters:

  • pg_id (str) – The UUID of the parent Process Group

  • scheduled (bool) – True to start, False to stop

  • components (list[ComponentType]) – The list of Component Entities to schdule, e.g. ProcessorEntity’s

Returns:

True for success, False for not

Return type:

(bool)

nipyapi.canvas.schedule_controller(controller, scheduled, refresh=False)[source]

Start/Enable or Stop/Disable a Controller Service

Parameters:

  • controller (ControllerServiceEntity) – Target Controller to schedule

  • scheduled (bool) – True to start, False to stop

  • refresh (bool) – Whether to refresh the component revision before execution

Returns:

(ControllerServiceEntity)

nipyapi.canvas.schedule_process_group(process_group_id, scheduled)[source]

Start or Stop a Process Group and all components.

Note that this doesn’t guarantee that all components have started, as some may be in Invalid states.

Parameters:

  • process_group_id (str) – The UUID of the target Process Group

  • scheduled (bool) – True to start, False to stop

Returns:

True of successfully scheduled, False if not

Return type:

(bool)

nipyapi.canvas.schedule_processor(processor, scheduled, refresh=True)[source]

Set a Processor to Start or Stop.

Note that this doesn’t guarantee that it will change state, merely that it will be instructed to try. Some effort is made to wait and see if the processor starts

Parameters:

  • processor (ProcessorEntity) – The Processor to target

  • scheduled (bool) – True to start, False to stop

  • refresh (bool) – Whether to refresh the object before action

Returns:

True for success, False for failure

Return type:

(bool)

nipyapi.canvas.set_remote_process_group_transmission(rpg, enable=True, refresh=True)[source]

Enable or Disable Transmission for an RPG

Parameters:

  • rpg (RemoteProcessGroupEntity) – The ID of the remote process group to modify

  • enable (bool) – True to enable, False to disable

  • refresh (bool) – Whether to refresh the object before action

Returns:

nipyapi.canvas.update_controller(controller, update)[source]

Updates the Configuration of a Controller Service

Parameters:
Returns:

(ControllerServiceEntity)

nipyapi.canvas.update_process_group(pg, update, refresh=True)[source]

Updates a given Process Group.

Parameters:

  • pg (ProcessGroupEntity) – The Process Group to target for update

  • update (dict) – key:value pairs to update

  • refresh (bool) – Whether to refresh the Process Group before applying the update

Returns:

The updated ProcessorEntity

Return type:

ProcessGroupEntity

nipyapi.canvas.update_processor(processor, update=None, name=None, refresh=True, auto_stop=False)[source]

Updates a Processor’s configuration and/or name.

For configuration changes, pass a ProcessorConfigDTO:

nifi.ProcessorConfigDTO(scheduling_period=’3s’)

For renaming, pass the new name. Both can be provided together.

Processors must be stopped for certain updates (including renaming). If auto_stop is True (default), the processor will be stopped before updating and restarted afterward if it was originally running.

Parameters:

  • processor (ProcessorEntity) – The Processor to target for update

  • update (ProcessorConfigDTO, optional) – Configuration parameters to update

  • name (str, optional) – New name for the processor

  • refresh (bool) – Whether to refresh the Processor object state before applying the update. Default True.

  • auto_stop (bool) – If True, automatically stop the processor before updating and restart afterward if it was running. Default False.

Returns:

The updated ProcessorEntity

Return type:

ProcessorEntity

Raises:

ValueError – If neither update nor name is provided, or if update is not a ProcessorConfigDTO, or if processor is running and auto_stop=False.

nipyapi.canvas.update_variable_registry(process_group, update, refresh=True)[source]

Updates one or more key:value pairs in the variable registry

Parameters:

  • process_group (ProcessGroupEntity) – The Process Group which has the Variable Registry to be updated

  • update (list[tuple]) – The variables to write to the registry

  • refresh (bool) – Whether to refresh the object revision before updating

Returns:

The created or updated Variable

Registry Entries

Return type:

VariableRegistryEntity