Canvas

Canvas operations and flow management

For interactions with the NiFi Canvas.

class nipyapi.canvas.FlowSubgraph(components, connections)[source]

Bases: tuple

Create new instance of FlowSubgraph(components, connections)

components[source]: Alias for field number 0

connections[source]: Alias for field number 1

nipyapi.canvas.clear_controller_state(controller)[source]

Clear all state for a controller service.

This removes all state entries from the controller service. Use with caution as this may affect the controller’s behavior (e.g., clearing a CDC table state service will cause tables to be re-snapshotted).

Note: The controller must be DISABLED before clearing state. Attempting to clear state on an enabled controller will raise an error.

Parameters:

controller – ControllerServiceEntity or controller service ID (str)

Returns:

The cleared state entity (should have 0 entries)

Return type:

ComponentStateEntity

Raises:

ValueError – Controller not found or controller is enabled
ApiException – NiFi API errors

Example:

# Disable controller first
nipyapi.canvas.schedule_controller(my_controller, scheduled=False)

# Clear all state
nipyapi.canvas.clear_controller_state(my_controller)

# Verify cleared
state = nipyapi.canvas.get_controller_state(my_controller)
assert state.component_state.local_state.total_entry_count == 0

nipyapi.canvas.clear_processor_state(processor)[source]

Clear all state for a processor.

This removes all state entries from the processor. Use with caution as this may affect the processor’s behavior (e.g., clearing ListFile state will cause all files to be re-listed).

Parameters:

processor – ProcessorEntity or processor ID (str)

Returns:

The cleared state entity (should have 0 entries)

Return type:

ComponentStateEntity

Raises:

ValueError – Processor not found
ApiException – NiFi API errors

Example:

# Clear all state
nipyapi.canvas.clear_processor_state(my_processor)

# Verify cleared
state = nipyapi.canvas.get_processor_state(my_processor)
assert state.component_state.local_state.total_entry_count == 0

nipyapi.canvas.create_connection(source, target, relationships=None, name=None, bends=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)
bends (list) – List of PositionDTO or (x, y) tuples for connection bends. For self-loop connections (source == target), bends are auto-calculated if not provided, to ensure the loop renders correctly in the UI.

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 (str or ProcessGroupEntity) – The parent Process Group ID (as a string) or ProcessGroupEntity object to create the new process group in. Use “root” for the root canvas.
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 (str or ProcessGroupEntity) – The parent Process Group ID (as a string) or ProcessGroupEntity object
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, refresh=True)[source]

Delete a Controller service, with optional prejudice

Parameters:

controller (ControllerServiceEntity or str) – Target Controller to delete, either as a ControllerServiceEntity object or a controller ID string
force (bool) – True to attempt Disable the Controller before deletion
refresh (bool) – Whether to refresh the controller to get latest revision before deletion. Defaults to True to avoid stale revision errors.

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, force=False)[source]

Deletes a given remote process group.

Parameters:

rpg (RemoteProcessGroupEntity) – Remote Process Group to remove
refresh (bool) – Whether to refresh the object before action
force (bool) – If True, stop transmission before deleting. Use this when the RPG may be transmitting and you want to ensure deletion.

Returns:

(RemoteProcessGroupEntity)

nipyapi.canvas.get_bulletin_board(pg_id=None, source_name=None, message=None, limit=None)[source]

Retrieves bulletins from the bulletin board with optional filtering.

This is an alias for nipyapi.bulletins.get_bulletin_board().

Parameters:

pg_id (str, optional) – Filter to bulletins from this process group ID. If None, returns bulletins from all groups.
source_name (str, optional) – Filter by source component name (regex pattern).
message (str, optional) – Filter by message content (regex pattern).
limit (int, optional) – Maximum number of bulletins to return.

Returns:

List of bulletin objects matching the filters.: Returns empty list if no bulletins match.

Return type:

list[BulletinDTO]

nipyapi.canvas.get_bulletins()[source]

Retrieves current bulletins (alerts) from the Flow Canvas.

This is an alias for nipyapi.bulletins.get_bulletins().

Returns:

The native datatype containing a list: of bulletins

Return type:

(ControllerBulletinsEntity)

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_connection(connection)[source]

Get a connection by ID or refresh a ConnectionEntity.

Parameters:: connection – Either a connection UUID (str) or a ConnectionEntity object. If a ConnectionEntity is provided, fetches a fresh copy (useful for getting the latest revision before updates).
Returns:: The requested connection
Return type:: ConnectionEntity

Example:

# Fetch by ID
conn = nipyapi.canvas.get_connection("abc-123-uuid")

# Refresh an existing entity
fresh_conn = nipyapi.canvas.get_connection(conn)

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_controller_service_docs(controller)[source]

Get detailed documentation for a controller service type.

This function retrieves the full ControllerServiceDefinition from NiFi, which contains comprehensive documentation useful for understanding controller service capabilities and configuration options.

Parameters:

controller (ControllerServiceEntity or DocumentedTypeDTO or str) – An existing controller service, a type from get_controller_type(), or a type name string (e.g., “JsonTreeReader” or full qualified name).

Returns:

Controller: documentation including property_descriptors, tags, and more.

None: If controller type not found.

Return type:

ControllerServiceDefinition

Example:

# From existing controller service
cs = nipyapi.canvas.get_controller("MyJsonReader")
docs = nipyapi.canvas.get_controller_service_docs(cs)
print(docs.tags)
print(docs.property_descriptors.keys())

# From controller type
cs_type = nipyapi.canvas.get_controller_type("JsonTreeReader")
docs = nipyapi.canvas.get_controller_service_docs(cs_type)

# From type name string
docs = nipyapi.canvas.get_controller_service_docs("AvroReader")

nipyapi.canvas.get_controller_state(controller)[source]

Get the state for a controller service.

Controller services can maintain internal state (e.g., cache entries, connection tracking, CDC table status). This function retrieves that state.

Parameters:

controller – ControllerServiceEntity or controller service ID (str)

Returns:

ComponentStateEntity with component_state containing component_id, state_description, local_state (single node), and cluster_state (distributed). Check whichever state map has entries. Each state map has scope, total_entry_count, and state (list of StateEntryDTO).

Raises:

ValueError – Controller not found
ApiException – NiFi API errors

Example:

state = nipyapi.canvas.get_controller_state(my_controller)
state_map = state.component_state.local_state
if state_map and state_map.state:
    for entry in state_map.state:
        print(f"{entry.key}: {entry.value}")

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

Gets the abstract object describing a controller, 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_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_flow_components(start_component, pg_id=None)[source]

Find all components and connections in a connected flow subgraph.

Performs a breadth-first traversal of the connection graph to find the complete connected subgraph (the ‘flow’). Useful for selecting an entire flow to move or analyze as a unit. The algorithm fetches all components in one API call, builds an adjacency map, then performs BFS from the start component to find all reachable nodes.

Parameters:

start_component – Any component entity (processor, funnel, port) to start the traversal from
pg_id – Process group ID containing the flow. If None, inferred from start_component.component.parent_group_id

Returns:

FlowSubgraph named tuple with ‘components’ (list of component entities) and ‘connections’ (list of ConnectionEntity objects within the flow).

Example:

# Get the complete flow subgraph
flow = nipyapi.canvas.get_flow_components(proc1)

# Access components
for c in flow.components:
    print(c.component.name)

# Access connections (useful for transpose_flow)
nipyapi.layout.transpose_flow(
    flow.components, offset=(400, 0), connections=flow.connections
)

nipyapi.canvas.get_flowfile_content(connection, flowfile_uuid, decode='auto', output_file=None, cluster_node_id=None)[source]

Download the content of a specific FlowFile.

This is a non-destructive operation - the FlowFile remains in the queue.

Parameters:

connection (str or ConnectionEntity) – Connection ID or ConnectionEntity.
flowfile_uuid (str) – UUID of the FlowFile.
decode (str) – How to decode content: “auto” (mime-based), “text” (UTF-8), or “bytes” (raw).
output_file (None or bool or str) – None returns content directly, True saves to current dir with FlowFile’s filename, str saves to that path/directory.
cluster_node_id (str or None) – Node ID for clustered NiFi. If not provided, will be auto-resolved from queue listing (adds one API call).

Returns:

bytes or str (depending on decode) If output_file is set: str path where file was saved

Return type:

If output_file is None

Example:

# Get content as auto-detected type
content = nipyapi.canvas.get_flowfile_content(connection_id, flowfile_uuid)

# Force text decoding
text = nipyapi.canvas.get_flowfile_content(conn, uuid, decode='text')

# Save to file using FlowFile's filename
path = nipyapi.canvas.get_flowfile_content(conn, uuid, output_file=True)
print(f"Saved to: {path}")

# Save to specific directory
path = nipyapi.canvas.get_flowfile_content(conn, uuid, output_file='/tmp/')

# In clustered NiFi, pass the cluster_node_id from listing
flowfiles = nipyapi.canvas.list_flowfiles(connection_id)
if flowfiles:
    content = nipyapi.canvas.get_flowfile_content(
        connection_id,
        flowfiles[0].uuid,
        cluster_node_id=flowfiles[0].cluster_node_id
    )

nipyapi.canvas.get_flowfile_details(connection, flowfile_uuid, cluster_node_id=None)[source]

Get full details for a specific FlowFile, including its attributes.

This is a non-destructive operation - the FlowFile remains in the queue.

Parameters:

connection – Connection ID (str) or ConnectionEntity
flowfile_uuid – UUID of the FlowFile to retrieve
cluster_node_id – Node ID for clustered NiFi. If not provided, will be auto-resolved from queue listing (adds one API call).

Returns:

FlowFile details including: attributes dict, filename, size, queued_duration, etc.

Return type:

FlowFileDTO

Example:

# Get FlowFile details
details = nipyapi.canvas.get_flowfile_details(connection_id, flowfile_uuid)
print(f"Filename: {details.filename}")
print(f"Size: {details.size} bytes")
print(f"Attributes: {details.attributes}")

# In clustered NiFi, pass the cluster_node_id from listing
flowfiles = nipyapi.canvas.list_flowfiles(connection_id)
if flowfiles:
    details = nipyapi.canvas.get_flowfile_details(
        connection_id,
        flowfiles[0].uuid,
        cluster_node_id=flowfiles[0].cluster_node_id
    )

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_docs(processor)[source]

Get detailed documentation for a processor, including properties, use cases, and tags.

This function retrieves the full ProcessorDefinition from NiFi, which contains comprehensive documentation useful for understanding processor capabilities and configuration options.

Parameters:

processor (ProcessorEntity or DocumentedTypeDTO or str) – An existing processor, a processor type from get_processor_type(), or a type name string (e.g., “GenerateFlowFile” or full qualified name).

Returns:

Processor documentation: including property_descriptors, tags, supported_relationships, multi_processor_use_cases, dynamic_properties, and more.

None: If processor type not found.

Return type:

ProcessorDefinition

Example:

# From existing processor
proc = nipyapi.canvas.get_processor("MyProcessor")
docs = nipyapi.canvas.get_processor_docs(proc)
print(docs.tags)  # ['record', 'update', 'json', ...]
print(docs.property_descriptors.keys())

# From processor type
proc_type = nipyapi.canvas.get_processor_type("UpdateRecord")
docs = nipyapi.canvas.get_processor_docs(proc_type)

# From type name string
docs = nipyapi.canvas.get_processor_docs("GenerateFlowFile")

nipyapi.canvas.get_processor_state(processor)[source]

Get the state for a processor.

Processors can maintain internal state (e.g., ListFile tracks listed files, TailFile tracks file positions). This function retrieves that state.

Parameters:

processor – ProcessorEntity or processor ID (str)

Returns:

ComponentStateEntity with component_state containing component_id, state_description, local_state (single node), and cluster_state (distributed). Check whichever state map has entries. Each state map has scope, total_entry_count, and state (list of StateEntryDTO).

Raises:

ValueError – Processor not found
ApiException – NiFi API errors

Example:

state = nipyapi.canvas.get_processor_state(my_list_file_processor)
state_map = state.component_state.local_state
if state_map and state_map.state:
    for entry in state_map.state:
        print(f"{entry.key}: {entry.value}")

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_flowfiles(connection, limit=100)[source]

List FlowFiles waiting in a connection’s queue.

This is a non-destructive operation - FlowFiles remain in the queue. Returns basic metadata for each FlowFile; use get_flowfile() to retrieve full details including attributes.

Parameters:

connection – Connection ID (str) or ConnectionEntity
limit – Maximum number of FlowFiles to return (default 100)

Returns:

List of FlowFile: summaries with uuid, filename, size, queued_duration, etc. Returns empty list if queue is empty.

Return type:

list[FlowFileSummaryDTO]

Example:

# List FlowFiles in a connection
conn = nipyapi.canvas.get_connection(connection_id)
flowfiles = nipyapi.canvas.list_flowfiles(conn)
for ff in flowfiles:
    print(f"{ff.uuid}: {ff.filename} ({ff.size} bytes, queued {ff.queued_duration}ms)")

# Get first FlowFile's full details
if flowfiles:
    details = nipyapi.canvas.get_flowfile(conn, flowfiles[0].uuid)
    print(details.attributes)

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.peek_flowfiles(connection, limit=1)[source]

Convenience function to list and get full details for FlowFiles at front of queue.

Combines list_flowfiles() and get_flowfile() to return complete FlowFile details including attributes for the first N FlowFiles in the queue.

Parameters:

connection – Connection ID (str) or ConnectionEntity
limit – Number of FlowFiles to retrieve details for (default 1)

Returns:

List of FlowFile details: with full attributes. Returns empty list if queue is empty.

Return type:

list[FlowFileDTO]

Example:

# Peek at the first FlowFile
flowfiles = nipyapi.canvas.peek_flowfiles(connection_id)
if flowfiles:
    ff = flowfiles[0]
    print(f"Filename: {ff.filename}")
    print(f"Attributes: {ff.attributes}")

# Peek at first 5
flowfiles = nipyapi.canvas.peek_flowfiles(connection_id, limit=5)

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. Waits for all controllers to reach the target state before returning.

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]

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

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

This function only supports RUNNING and STOPPED states. For RUN_ONCE, use schedule_processor() on individual processors.

Parameters:

pg_id (str) – The UUID of the parent Process Group
scheduled (bool) – True to start (RUNNING), False to stop (STOPPED)
components (list[ComponentType]) – The list of Component Entities to schedule, e.g. ProcessorEntity’s. If None, schedules all components in the Process Group.

Returns:

True for success, False for failure

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, Stop, Disable, or Run Once.

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 reaches the target state.

Parameters:

processor (str or ProcessorEntity) – The Processor ID or ProcessorEntity object.
scheduled (bool or str) – True/False for RUNNING/STOPPED, or one of “RUNNING”, “STOPPED”, “DISABLED”, “RUN_ONCE”.
refresh (bool) – Whether to refresh the object before action.

Returns:

True for success, False for failure.

Return type:

bool

Example:

# Start a processor by ID
nipyapi.canvas.schedule_processor("<processor-id>", True)

# Start a processor object (backwards compatible)
nipyapi.canvas.schedule_processor(proc, True)

# Stop a processor
nipyapi.canvas.schedule_processor(proc, False)

# Disable a processor (prevents starting, useful for maintenance)
nipyapi.canvas.schedule_processor(proc, "DISABLED")

# Run once - executes one scheduling cycle then stops
nipyapi.canvas.schedule_processor(proc, "RUN_ONCE")

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

Enable or Disable Transmission for an RPG.

Waits for the transmission state to actually change before returning.

Parameters:

rpg (RemoteProcessGroupEntity) – The remote process group to modify
enable (bool) – True to enable transmission, False to disable
refresh (bool) – Whether to refresh the object before action

Returns:

The updated remote process group

Return type:

(RemoteProcessGroupEntity)

Raises:

ValueError – If the state change times out

nipyapi.canvas.update_connection(connection, name=None, bends=None, refresh=True)[source]

Update a connection’s configuration.

Only parameters explicitly provided will be updated. To clear bends, pass an empty list.

Parameters:

connection (ConnectionEntity or str) – ConnectionEntity or connection ID to update.
name (str or None) – New name for the connection, or None to leave unchanged.
bends (list or None) – Bend points as list of PositionDTO or (x, y) tuples. None keeps existing bends, [] clears all bends, […] sets specific points.
refresh (bool) – Whether to refresh the connection before updating.

Returns:

The updated connection

Return type:

ConnectionEntity

Example:

# Clear all bends from a connection
updated = nipyapi.canvas.update_connection(conn, bends=[])

# Clear bends by connection ID
updated = nipyapi.canvas.update_connection("abc-123-uuid", bends=[])

# Rename a connection
updated = nipyapi.canvas.update_connection(conn, name="Primary Path")

# Set specific bend points
updated = nipyapi.canvas.update_connection(conn, bends=[(500, 300), (500, 400)])

nipyapi.canvas.update_controller(controller, update, refresh=True)[source]

Updates the Configuration of a Controller Service

Parameters:

controller (ControllerServiceEntity or str) – Target Controller to update, either as a ControllerServiceEntity object or a controller ID string
update (ControllerServiceDTO) – Controller Service configuration object containing the new config params and properties
refresh (bool) – Whether to refresh the controller to get latest revision before update. Defaults to True to avoid stale revision errors.

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

nipyapi.canvas.verify_controller(controller, properties=None, attributes=None)[source]

Verify a controller service’s configuration properties are valid.

Validates that all required properties are set and property values meet their defined constraints. Does NOT test actual connectivity or credentials. Handles the async verification workflow: submit request, poll until complete, cleanup.

The controller service must be DISABLED before verification.

Parameters:

controller – ControllerServiceEntity or controller service ID (str)
properties – Optional dict of property overrides to verify
attributes – Optional dict of FlowFile attributes for Expression Language

Returns:

Verification results. Each has verification_step_name, outcome (“SUCCESSFUL”/”FAILED”/”SKIPPED”), and explanation.

Return type:

list[ConfigVerificationResultDTO]

Raises:

ValueError – Controller not found or is currently enabled
ApiException – NiFi API errors

Example:

results = nipyapi.canvas.verify_controller(dbcp_service)
for r in results:
    print(f"{r.verification_step_name}: {r.outcome}")
    if r.outcome == "FAILED":
        print(f"  Reason: {r.explanation}")

nipyapi.canvas.verify_processor(processor, properties=None, attributes=None)[source]

Verify a processor’s configuration properties are valid.

Validates that all required properties are set and property values meet their defined constraints. Does NOT test actual connectivity or external service availability. Handles the async verification workflow: submit request, poll until complete, cleanup.

The processor must be STOPPED before verification.

Parameters:

processor – ProcessorEntity or processor ID (str)
properties – Optional dict of property overrides to verify
attributes – Optional dict of FlowFile attributes for Expression Language

Returns:

Verification results. Each has verification_step_name, outcome (“SUCCESSFUL”/”FAILED”/”SKIPPED”), and explanation.

Return type:

list[ConfigVerificationResultDTO]

Raises:

ValueError – Processor not found or is currently running
ApiException – NiFi API errors

Example:

results = nipyapi.canvas.verify_processor(my_processor)
for r in results:
    print(f"{r.verification_step_name}: {r.outcome}")
    if r.outcome == "FAILED":
        print(f"  Reason: {r.explanation}")