2026-04-17 18:42:08 -03:00
<!doctype html>
< html lang = "en" >
< head >
< meta charset = "utf-8" >
< meta name = "viewport" content = "width=device-width, initial-scale=1, minimum-scale=1" >
2026-05-28 18:22:00 -03:00
< meta name = "generator" content = "pdoc3 0.11.5" >
2026-04-17 18:42:08 -03:00
< title > connpy.services.node_service API documentation</ title >
< meta name = "description" content = "" >
< link rel = "stylesheet" href = "https://cdnjs.cloudflare.com/ajax/libs/10up-sanitize.css/13.0.0/sanitize.min.css" integrity = "sha512-y1dtMcuvtTMJc1yPgEqF0ZjQbhnc/bFhyvIyVNb9Zk5mIGtqVaAB1Ttl28su8AvFMOY0EwRbAe+HCLqj6W7/KA==" crossorigin >
< link rel = "stylesheet" href = "https://cdnjs.cloudflare.com/ajax/libs/10up-sanitize.css/13.0.0/typography.min.css" integrity = "sha512-Y1DYSb995BAfxobCkKepB1BqJJTPrOp3zPL74AWFugHHmmdcvO+C48WLrUOlhGMc0QG7AE3f7gmvvcrmX2fDoA==" crossorigin >
< link rel = "stylesheet" href = "https://cdnjs.cloudflare.com/ajax/libs/highlight.js/11.9.0/styles/default.min.css" crossorigin >
< style >: root { --highlight-color : #fe9 }. flex { display : flex !important } body { line-height : 1.5 em }# content { padding : 20 px }# sidebar { padding : 1.5 em ; overflow : hidden }# sidebar > * : last-child { margin-bottom : 2 cm }. http-server-breadcrumbs { font-size : 130 % ; margin : 0 0 15 px 0 }# footer { font-size : .75 em ; padding : 5 px 30 px ; border-top : 1 px solid #ddd ; text-align : right }# footer p { margin : 0 0 0 1 em ; display : inline-block }# footer p : last-child { margin-right : 30 px } h1 , h2 , h3 , h4 , h5 { font-weight : 300 } h1 { font-size : 2.5 em ; line-height : 1.1 em } h2 { font-size : 1.75 em ; margin : 2 em 0 .50 em 0 } h3 { font-size : 1.4 em ; margin : 1.6 em 0 .7 em 0 } h4 { margin : 0 ; font-size : 105 % } h1 : target , h2 : target , h3 : target , h4 : target , h5 : target , h6 : target { background : var ( -- highlight - color ); padding : .2 em 0 } a { color : #058 ; text-decoration : none ; transition : color .2 s ease-in-out } a : visited { color : #503 } a : hover { color : #b62 }. title code { font-weight : bold } h2 [ id ^= "header-" ] { margin-top : 2 em }. ident { color : #900 ; font-weight : bold } pre code { font-size : .8 em ; line-height : 1.4 em ; padding : 1 em ; display : block } code { background : #f3f3f3 ; font-family : "DejaVu Sans Mono" , monospace ; padding : 1 px 4 px ; overflow-wrap : break-word } h1 code { background : transparent } pre { border-top : 1 px solid #ccc ; border-bottom : 1 px solid #ccc ; margin : 1 em 0 }# http-server-module-list { display : flex ; flex-flow : column }# http-server-module-list div { display : flex }# http-server-module-list dt { min-width : 10 % }# http-server-module-list p { margin-top : 0 }. toc ul , # index { list-style-type : none ; margin : 0 ; padding : 0 }# index code { background : transparent }# index h3 { border-bottom : 1 px solid #ddd }# index ul { padding : 0 }# index h4 { margin-top : .6 em ; font-weight : bold }@ media ( min-width : 200ex ) {# index . two-column { column-count : 2 }}@ media ( min-width : 300ex ) {# index . two-column { column-count : 3 }} dl { margin-bottom : 2 em } dl dl : last-child { margin-bottom : 4 em } dd { margin : 0 0 1 em 3 em }# header-classes + dl > dd { margin-bottom : 3 em } dd dd { margin-left : 2 em } dd p { margin : 10 px 0 }. name { background : #eee ; font-size : .85 em ; padding : 5 px 10 px ; display : inline-block ; min-width : 40 % }. name : hover { background : #e0e0e0 } dt : target . name { background : var ( -- highlight - color )}. name > span : first-child { white-space : nowrap }. name . class > span : nth-child ( 2 ) { margin-left : .4 em }. inherited { color : #999 ; border-left : 5 px solid #eee ; padding-left : 1 em }. inheritance em { font-style : normal ; font-weight : bold }. desc h2 { font-weight : 400 ; font-size : 1.25 em }. desc h3 { font-size : 1 em }. desc dt code { background : inherit }. source > summary , . git-link-div { color : #666 ; text-align : right ; font-weight : 400 ; font-size : .8 em ; text-transform : uppercase }. source summary > * { white-space : nowrap ; cursor : pointer }. git-link { color : inherit ; margin-left : 1 em }. source pre { max-height : 500 px ; overflow : auto ; margin : 0 }. source pre code { font-size : 12 px ; overflow : visible ; min-width : max-content }. hlist { list-style : none }. hlist li { display : inline }. hlist li : after { content : ',\2002' }. hlist li : last-child : after { content : none }. hlist . hlist { display : inline ; padding-left : 1 em } img { max-width : 100 % } td { padding : 0 .5 em }. admonition { padding : .1 em 1 em ; margin : 1 em 0 }. admonition-title { font-weight : bold }. admonition . note , . admonition . info , . admonition . important { background : #aef }. admonition . todo , . admonition . versionadded , . admonition . tip , . admonition . hint { background : #dfd }. admonition . warning , . admonition . versionchanged , . admonition . deprecated { background : #fd4 }. admonition . error , . admonition . danger , . admonition . caution { background : lightpink }</ style >
< style media = "screen and (min-width: 700px)" >@ media screen and ( min-width : 700px ) {# sidebar { width : 30 % ; height : 100 vh ; overflow : auto ; position : sticky ; top : 0 }# content { width : 70 % ; max-width : 100 ch ; padding : 3 em 4 em ; border-left : 1 px solid #ddd } pre code { font-size : 1 em }. name { font-size : 1 em } main { display : flex ; flex-direction : row-reverse ; justify-content : flex-end }. toc ul ul , # index ul ul { padding-left : 1 em }. toc > ul > li { margin-top : .5 em }}</ style >
< style media = "print" >@ media print {# sidebar h1 { page-break-before : always }. source { display : none }}@ media print { * { background : transparent !important ; color : #000 !important ; box-shadow : none !important ; text-shadow : none !important } a [ href ] : after { content : " (" attr ( href ) ")" ; font-size : 90 % } a [ href ][ title ] : after { content : none } abbr [ title ] : after { content : " (" attr ( title ) ")" }. ir a : after , a [ href ^= "javascript:" ] : after , a [ href ^= "#" ] : after { content : "" } pre , blockquote { border : 1 px solid #999 ; page-break-inside : avoid } thead { display : table-header-group } tr , img { page-break-inside : avoid } img { max-width : 100 % !important }@ page { margin : 0 . 5cm } p , h2 , h3 { orphans : 3 ; widows : 3 } h1 , h2 , h3 , h4 , h5 , h6 { page-break-after : avoid }}</ style >
< script defer src = "https://cdnjs.cloudflare.com/ajax/libs/highlight.js/11.9.0/highlight.min.js" integrity = "sha512-D9gUyxqja7hBtkWpPWGt9wfbfaMGVt9gnyCvYa+jojwwPHLCzUm5i8rpk7vD7wNee9bA35eYIjobYPaQuKS1MQ==" crossorigin ></ script >
< script > window . addEventListener ( 'DOMContentLoaded' , () => {
hljs . configure ({ languages : [ 'bash' , 'css' , 'diff' , 'graphql' , 'ini' , 'javascript' , 'json' , 'plaintext' , 'python' , 'python-repl' , 'rust' , 'shell' , 'sql' , 'typescript' , 'xml' , 'yaml' ]});
hljs . highlightAll ();
/* Collapse source docstrings */
setTimeout (() => {
[... document . querySelectorAll ( '.hljs.language-python > .hljs-string' )]
. filter ( el => el . innerHTML . length > 200 && [ '"""' , "'''" ]. includes ( el . innerHTML . substring ( 0 , 3 )))
. forEach ( el => {
let d = document . createElement ( 'details' );
d . classList . add ( 'hljs-string' );
d . innerHTML = '<summary>"""</summary>' + el . innerHTML . substring ( 3 );
el . replaceWith ( d );
});
}, 100 );
})</ script >
</ head >
< body >
< main >
< article id = "content" >
< header >
< h1 class = "title" > Module < code > connpy.services.node_service</ code ></ h1 >
</ header >
< section id = "section-intro" >
</ section >
< section >
</ section >
< section >
</ section >
< section >
</ section >
< section >
< h2 class = "section-title" id = "header-classes" > Classes</ h2 >
< dl >
< dt id = "connpy.services.node_service.NodeService" >< code class = "flex name class" >
< span > class < span class = "ident" > NodeService</ span ></ span >
< span > (</ span >< span > config=None)</ span >
</ code ></ dt >
< dd >
< details class = "source" >
< summary >
< span > Expand source code</ span >
</ summary >
< pre >< code class = "python" > class NodeService(BaseService):
def __init__(self, config=None):
super().__init__(config)
def list_nodes(self, filter_str=None, format_str=None):
""" Return a listed filtered by regex match and formatted if needed."""
nodes = self.config._getallnodes()
case_sensitive = self.config.config.get(" case" , False)
if filter_str:
flags = re.IGNORECASE if not case_sensitive else 0
nodes = [n for n in nodes if re.search(filter_str, n, flags)]
if not format_str:
return nodes
from .profile_service import ProfileService
profile_service = ProfileService(self.config)
formatted_nodes = []
for n_id in nodes:
# Use ProfileService to resolve profiles for dynamic formatting
details = self.config.getitem(n_id, extract=False)
if details:
details = profile_service.resolve_node_data(details)
name = n_id.split(" @" )[0]
location = n_id.partition(" @" )[2] or " root"
# Prepare context for .format() with all details
context = details.copy()
context.update({
" name" : name,
" NAME" : name.upper(),
" location" : location,
" LOCATION" : location.upper(),
})
# Add exploded uniques (id, folder, subfolder)
uniques = self.config._explode_unique(n_id)
if uniques:
context.update(uniques)
# Add uppercase versions of all keys for convenience
for k, v in list(context.items()):
if isinstance(v, str):
context[k.upper()] = v.upper()
try:
formatted_nodes.append(format_str.format(**context))
except (KeyError, IndexError, ValueError):
# Fallback to original string if format fails
formatted_nodes.append(n_id)
return formatted_nodes
def list_folders(self, filter_str=None):
""" Return all unique folders, optionally filtered by regex."""
folders = self.config._getallfolders()
case_sensitive = self.config.config.get(" case" , False)
if filter_str:
2026-05-01 18:55:25 -03:00
if filter_str.startswith(" @" ):
if not case_sensitive:
folders = [f for f in folders if f.lower() == filter_str.lower()]
else:
folders = [f for f in folders if f == filter_str]
else:
flags = re.IGNORECASE if not case_sensitive else 0
folders = [f for f in folders if re.search(filter_str, f, flags)]
2026-04-17 18:42:08 -03:00
return folders
def get_node_details(self, unique_id):
""" Return full configuration dictionary for a specific node."""
2026-04-24 19:23:00 -03:00
try:
details = self.config.getitem(unique_id)
if not details:
raise NodeNotFoundError(f" Node ' {unique_id}' not found." )
return details
except (KeyError, TypeError):
2026-04-17 18:42:08 -03:00
raise NodeNotFoundError(f" Node ' {unique_id}' not found." )
def explode_unique(self, unique_id):
""" Explode a unique ID into a dictionary of its parts."""
return self.config._explode_unique(unique_id)
def generate_cache(self, nodes=None, folders=None, profiles=None):
""" Generate and update the internal nodes cache."""
self.config._generate_nodes_cache(nodes=nodes, folders=folders, profiles=profiles)
2026-05-01 18:55:25 -03:00
def validate_parent_folder(self, unique_id, is_folder=False):
2026-04-24 19:23:00 -03:00
""" Check if parent folder exists for a given node unique ID."""
2026-05-01 18:55:25 -03:00
if is_folder:
uniques = self.config._explode_unique(unique_id)
if uniques and " subfolder" in uniques and " folder" in uniques:
parent_folder = f" @{uniques[' folder' ]}"
if parent_folder not in self.config._getallfolders():
raise NodeNotFoundError(f" Folder ' {parent_folder}' not found." )
else:
node_folder = unique_id.partition(" @" )[2]
if node_folder:
parent_folder = f" @{node_folder}"
if parent_folder not in self.config._getallfolders():
raise NodeNotFoundError(f" Folder ' {parent_folder}' not found." )
2026-04-24 19:23:00 -03:00
2026-04-17 18:42:08 -03:00
def add_node(self, unique_id, data, is_folder=False):
""" Logic for adding a new node or folder to configuration."""
if not is_folder:
self._validate_node_name(unique_id)
all_nodes = self.config._getallnodes()
all_folders = self.config._getallfolders()
if is_folder:
if unique_id in all_folders:
raise NodeAlreadyExistsError(f" Folder ' {unique_id}' already exists." )
uniques = self.config._explode_unique(unique_id)
if not uniques:
raise InvalidConfigurationError(f" Invalid folder name ' {unique_id}' ." )
# Check if parent folder exists when creating a subfolder
if " subfolder" in uniques:
2026-05-01 18:55:25 -03:00
self.validate_parent_folder(unique_id, is_folder=True)
2026-04-17 18:42:08 -03:00
self.config._folder_add(**uniques)
self.config._saveconfig(self.config.file)
else:
if unique_id in all_nodes:
raise NodeAlreadyExistsError(f" Node ' {unique_id}' already exists." )
# Check if parent folder exists when creating a node in a folder
2026-04-24 19:23:00 -03:00
self.validate_parent_folder(unique_id)
2026-04-17 18:42:08 -03:00
# Ensure ' id' is in data for config._connections_add
if " id" not in data:
uniques = self.config._explode_unique(unique_id)
if uniques and " id" in uniques:
data[" id" ] = uniques[" id" ]
self.config._connections_add(**data)
self.config._saveconfig(self.config.file)
2026-05-28 18:22:00 -03:00
def update_node(self, unique_id, data, save=True):
2026-04-17 18:42:08 -03:00
""" Explicitly update an existing node."""
all_nodes = self.config._getallnodes()
if unique_id not in all_nodes:
raise NodeNotFoundError(f" Node ' {unique_id}' not found." )
# Ensure ' id' is in data for config._connections_add
if " id" not in data:
uniques = self.config._explode_unique(unique_id)
if uniques:
data[" id" ] = uniques[" id" ]
# config._connections_add actually handles updates if ID exists correctly
self.config._connections_add(**data)
2026-05-28 18:22:00 -03:00
if save:
self.config._saveconfig(self.config.file)
2026-04-17 18:42:08 -03:00
2026-05-28 18:22:00 -03:00
def delete_node(self, unique_id, is_folder=False, save=True):
2026-04-17 18:42:08 -03:00
""" Logic for deleting a node or folder."""
if is_folder:
uniques = self.config._explode_unique(unique_id)
if not uniques:
raise NodeNotFoundError(f" Folder ' {unique_id}' not found or invalid." )
self.config._folder_del(**uniques)
else:
uniques = self.config._explode_unique(unique_id)
if not uniques:
raise NodeNotFoundError(f" Node ' {unique_id}' not found or invalid." )
self.config._connections_del(**uniques)
2026-05-28 18:22:00 -03:00
if save:
self.config._saveconfig(self.config.file)
2026-04-17 18:42:08 -03:00
def connect_node(self, unique_id, sftp=False, debug=False, logger=None):
""" Interact with a node directly."""
from connpy.core import node
from .profile_service import ProfileService
node_data = self.config.getitem(unique_id, extract=False)
if not node_data:
raise NodeNotFoundError(f" Node ' {unique_id}' not found." )
# Resolve profiles
profile_service = ProfileService(self.config)
resolved_data = profile_service.resolve_node_data(node_data)
n = node(unique_id, **resolved_data, config=self.config)
if sftp:
n.protocol = " sftp"
2026-04-27 15:12:07 -03:00
2026-04-17 18:42:08 -03:00
n.interact(debug=debug, logger=logger)
def move_node(self, src_id, dst_id, copy=False):
""" Move or copy a node."""
self._validate_node_name(dst_id)
node_data = self.config.getitem(src_id)
if not node_data:
raise NodeNotFoundError(f" Source node ' {src_id}' not found." )
if dst_id in self.config._getallnodes():
raise NodeAlreadyExistsError(f" Destination node ' {dst_id}' already exists." )
new_uniques = self.config._explode_unique(dst_id)
if not new_uniques:
raise InvalidConfigurationError(f" Invalid destination format ' {dst_id}' ." )
new_node_data = node_data.copy()
new_node_data.update(new_uniques)
self.config._connections_add(**new_node_data)
if not copy:
src_uniques = self.config._explode_unique(src_id)
self.config._connections_del(**src_uniques)
self.config._saveconfig(self.config.file)
def bulk_add(self, ids, hosts, common_data):
""" Add multiple nodes with shared common configuration."""
count = 0
all_nodes = self.config._getallnodes()
for i, uid in enumerate(ids):
if uid in all_nodes:
continue
try:
self._validate_node_name(uid)
except ReservedNameError:
# For bulk, we might want to just skip or log.
# CLI caller will handle if it wants to be strict.
continue
host = hosts[i] if i < len(hosts) else hosts[0]
uniques = self.config._explode_unique(uid)
if not uniques:
continue
node_data = common_data.copy()
node_data.pop(" ids" , None)
node_data.pop(" location" , None)
node_data.update(uniques)
node_data[" host" ] = host
node_data[" type" ] = " connection"
self.config._connections_add(**node_data)
count += 1
if count > 0:
self.config._saveconfig(self.config.file)
return count
def full_replace(self, connections, profiles):
""" Replace all connections and profiles with new data."""
self.config.connections = connections
self.config.profiles = profiles
self.config._saveconfig(self.config.file)
def get_inventory(self):
""" Return a full snapshot of connections and profiles."""
return {
" connections" : self.config.connections,
" profiles" : self.config.profiles
}</ code ></ pre >
</ details >
< div class = "desc" >< p > Base class for all connpy services, providing common configuration access.</ p >
< p > Initialize the service.</ p >
< h2 id = "args" > Args</ h2 >
< dl >
< dt >< strong >< code > config</ code ></ strong ></ dt >
< dd > An instance of configfile (or None to instantiate a new one/use global context).</ dd >
</ dl ></ div >
< h3 > Ancestors</ h3 >
< ul class = "hlist" >
< li >< a title = "connpy.services.base.BaseService" href = "base.html#connpy.services.base.BaseService" > BaseService</ a ></ li >
</ ul >
< h3 > Methods</ h3 >
< dl >
< dt id = "connpy.services.node_service.NodeService.add_node" >< code class = "name flex" >
< span > def < span class = "ident" > add_node</ span ></ span > (< span > self, unique_id, data, is_folder=False)</ span >
</ code ></ dt >
< dd >
< details class = "source" >
< summary >
< span > Expand source code</ span >
</ summary >
< pre >< code class = "python" > def add_node(self, unique_id, data, is_folder=False):
""" Logic for adding a new node or folder to configuration."""
if not is_folder:
self._validate_node_name(unique_id)
all_nodes = self.config._getallnodes()
all_folders = self.config._getallfolders()
if is_folder:
if unique_id in all_folders:
raise NodeAlreadyExistsError(f" Folder ' {unique_id}' already exists." )
uniques = self.config._explode_unique(unique_id)
if not uniques:
raise InvalidConfigurationError(f" Invalid folder name ' {unique_id}' ." )
# Check if parent folder exists when creating a subfolder
if " subfolder" in uniques:
2026-05-01 18:55:25 -03:00
self.validate_parent_folder(unique_id, is_folder=True)
2026-04-17 18:42:08 -03:00
self.config._folder_add(**uniques)
self.config._saveconfig(self.config.file)
else:
if unique_id in all_nodes:
raise NodeAlreadyExistsError(f" Node ' {unique_id}' already exists." )
# Check if parent folder exists when creating a node in a folder
2026-04-24 19:23:00 -03:00
self.validate_parent_folder(unique_id)
2026-04-17 18:42:08 -03:00
# Ensure ' id' is in data for config._connections_add
if " id" not in data:
uniques = self.config._explode_unique(unique_id)
if uniques and " id" in uniques:
data[" id" ] = uniques[" id" ]
self.config._connections_add(**data)
self.config._saveconfig(self.config.file)</ code ></ pre >
</ details >
< div class = "desc" >< p > Logic for adding a new node or folder to configuration.</ p ></ div >
</ dd >
< dt id = "connpy.services.node_service.NodeService.bulk_add" >< code class = "name flex" >
< span > def < span class = "ident" > bulk_add</ span ></ span > (< span > self, ids, hosts, common_data)</ span >
</ code ></ dt >
< dd >
< details class = "source" >
< summary >
< span > Expand source code</ span >
</ summary >
< pre >< code class = "python" > def bulk_add(self, ids, hosts, common_data):
""" Add multiple nodes with shared common configuration."""
count = 0
all_nodes = self.config._getallnodes()
for i, uid in enumerate(ids):
if uid in all_nodes:
continue
try:
self._validate_node_name(uid)
except ReservedNameError:
# For bulk, we might want to just skip or log.
# CLI caller will handle if it wants to be strict.
continue
host = hosts[i] if i < len(hosts) else hosts[0]
uniques = self.config._explode_unique(uid)
if not uniques:
continue
node_data = common_data.copy()
node_data.pop(" ids" , None)
node_data.pop(" location" , None)
node_data.update(uniques)
node_data[" host" ] = host
node_data[" type" ] = " connection"
self.config._connections_add(**node_data)
count += 1
if count > 0:
self.config._saveconfig(self.config.file)
return count</ code ></ pre >
</ details >
< div class = "desc" >< p > Add multiple nodes with shared common configuration.</ p ></ div >
</ dd >
< dt id = "connpy.services.node_service.NodeService.connect_node" >< code class = "name flex" >
< span > def < span class = "ident" > connect_node</ span ></ span > (< span > self, unique_id, sftp=False, debug=False, logger=None)</ span >
</ code ></ dt >
< dd >
< details class = "source" >
< summary >
< span > Expand source code</ span >
</ summary >
< pre >< code class = "python" > def connect_node(self, unique_id, sftp=False, debug=False, logger=None):
""" Interact with a node directly."""
from connpy.core import node
from .profile_service import ProfileService
node_data = self.config.getitem(unique_id, extract=False)
if not node_data:
raise NodeNotFoundError(f" Node ' {unique_id}' not found." )
# Resolve profiles
profile_service = ProfileService(self.config)
resolved_data = profile_service.resolve_node_data(node_data)
n = node(unique_id, **resolved_data, config=self.config)
if sftp:
n.protocol = " sftp"
2026-04-27 15:12:07 -03:00
2026-04-17 18:42:08 -03:00
n.interact(debug=debug, logger=logger)</ code ></ pre >
</ details >
< div class = "desc" >< p > Interact with a node directly.</ p ></ div >
</ dd >
< dt id = "connpy.services.node_service.NodeService.delete_node" >< code class = "name flex" >
2026-05-28 18:22:00 -03:00
< span > def < span class = "ident" > delete_node</ span ></ span > (< span > self, unique_id, is_folder=False, save=True)</ span >
2026-04-17 18:42:08 -03:00
</ code ></ dt >
< dd >
< details class = "source" >
< summary >
< span > Expand source code</ span >
</ summary >
2026-05-28 18:22:00 -03:00
< pre >< code class = "python" > def delete_node(self, unique_id, is_folder=False, save=True):
2026-04-17 18:42:08 -03:00
""" Logic for deleting a node or folder."""
if is_folder:
uniques = self.config._explode_unique(unique_id)
if not uniques:
raise NodeNotFoundError(f" Folder ' {unique_id}' not found or invalid." )
self.config._folder_del(**uniques)
else:
uniques = self.config._explode_unique(unique_id)
if not uniques:
raise NodeNotFoundError(f" Node ' {unique_id}' not found or invalid." )
self.config._connections_del(**uniques)
2026-05-28 18:22:00 -03:00
if save:
self.config._saveconfig(self.config.file)</ code ></ pre >
2026-04-17 18:42:08 -03:00
</ details >
< div class = "desc" >< p > Logic for deleting a node or folder.</ p ></ div >
</ dd >
< dt id = "connpy.services.node_service.NodeService.explode_unique" >< code class = "name flex" >
< span > def < span class = "ident" > explode_unique</ span ></ span > (< span > self, unique_id)</ span >
</ code ></ dt >
< dd >
< details class = "source" >
< summary >
< span > Expand source code</ span >
</ summary >
< pre >< code class = "python" > def explode_unique(self, unique_id):
""" Explode a unique ID into a dictionary of its parts."""
return self.config._explode_unique(unique_id)</ code ></ pre >
</ details >
< div class = "desc" >< p > Explode a unique ID into a dictionary of its parts.</ p ></ div >
</ dd >
< dt id = "connpy.services.node_service.NodeService.full_replace" >< code class = "name flex" >
< span > def < span class = "ident" > full_replace</ span ></ span > (< span > self, connections, profiles)</ span >
</ code ></ dt >
< dd >
< details class = "source" >
< summary >
< span > Expand source code</ span >
</ summary >
< pre >< code class = "python" > def full_replace(self, connections, profiles):
""" Replace all connections and profiles with new data."""
self.config.connections = connections
self.config.profiles = profiles
self.config._saveconfig(self.config.file)</ code ></ pre >
</ details >
< div class = "desc" >< p > Replace all connections and profiles with new data.</ p ></ div >
</ dd >
< dt id = "connpy.services.node_service.NodeService.generate_cache" >< code class = "name flex" >
< span > def < span class = "ident" > generate_cache</ span ></ span > (< span > self, nodes=None, folders=None, profiles=None)</ span >
</ code ></ dt >
< dd >
< details class = "source" >
< summary >
< span > Expand source code</ span >
</ summary >
< pre >< code class = "python" > def generate_cache(self, nodes=None, folders=None, profiles=None):
""" Generate and update the internal nodes cache."""
self.config._generate_nodes_cache(nodes=nodes, folders=folders, profiles=profiles)</ code ></ pre >
</ details >
< div class = "desc" >< p > Generate and update the internal nodes cache.</ p ></ div >
</ dd >
< dt id = "connpy.services.node_service.NodeService.get_inventory" >< code class = "name flex" >
< span > def < span class = "ident" > get_inventory</ span ></ span > (< span > self)</ span >
</ code ></ dt >
< dd >
< details class = "source" >
< summary >
< span > Expand source code</ span >
</ summary >
< pre >< code class = "python" > def get_inventory(self):
""" Return a full snapshot of connections and profiles."""
return {
" connections" : self.config.connections,
" profiles" : self.config.profiles
}</ code ></ pre >
</ details >
< div class = "desc" >< p > Return a full snapshot of connections and profiles.</ p ></ div >
</ dd >
< dt id = "connpy.services.node_service.NodeService.get_node_details" >< code class = "name flex" >
< span > def < span class = "ident" > get_node_details</ span ></ span > (< span > self, unique_id)</ span >
</ code ></ dt >
< dd >
< details class = "source" >
< summary >
< span > Expand source code</ span >
</ summary >
< pre >< code class = "python" > def get_node_details(self, unique_id):
""" Return full configuration dictionary for a specific node."""
2026-04-24 19:23:00 -03:00
try:
details = self.config.getitem(unique_id)
if not details:
raise NodeNotFoundError(f" Node ' {unique_id}' not found." )
return details
except (KeyError, TypeError):
raise NodeNotFoundError(f" Node ' {unique_id}' not found." )</ code ></ pre >
2026-04-17 18:42:08 -03:00
</ details >
< div class = "desc" >< p > Return full configuration dictionary for a specific node.</ p ></ div >
</ dd >
< dt id = "connpy.services.node_service.NodeService.list_folders" >< code class = "name flex" >
< span > def < span class = "ident" > list_folders</ span ></ span > (< span > self, filter_str=None)</ span >
</ code ></ dt >
< dd >
< details class = "source" >
< summary >
< span > Expand source code</ span >
</ summary >
< pre >< code class = "python" > def list_folders(self, filter_str=None):
""" Return all unique folders, optionally filtered by regex."""
folders = self.config._getallfolders()
case_sensitive = self.config.config.get(" case" , False)
if filter_str:
2026-05-01 18:55:25 -03:00
if filter_str.startswith(" @" ):
if not case_sensitive:
folders = [f for f in folders if f.lower() == filter_str.lower()]
else:
folders = [f for f in folders if f == filter_str]
else:
flags = re.IGNORECASE if not case_sensitive else 0
folders = [f for f in folders if re.search(filter_str, f, flags)]
2026-04-17 18:42:08 -03:00
return folders</ code ></ pre >
</ details >
< div class = "desc" >< p > Return all unique folders, optionally filtered by regex.</ p ></ div >
</ dd >
< dt id = "connpy.services.node_service.NodeService.list_nodes" >< code class = "name flex" >
< span > def < span class = "ident" > list_nodes</ span ></ span > (< span > self, filter_str=None, format_str=None)</ span >
</ code ></ dt >
< dd >
< details class = "source" >
< summary >
< span > Expand source code</ span >
</ summary >
< pre >< code class = "python" > def list_nodes(self, filter_str=None, format_str=None):
""" Return a listed filtered by regex match and formatted if needed."""
nodes = self.config._getallnodes()
case_sensitive = self.config.config.get(" case" , False)
if filter_str:
flags = re.IGNORECASE if not case_sensitive else 0
nodes = [n for n in nodes if re.search(filter_str, n, flags)]
if not format_str:
return nodes
from .profile_service import ProfileService
profile_service = ProfileService(self.config)
formatted_nodes = []
for n_id in nodes:
# Use ProfileService to resolve profiles for dynamic formatting
details = self.config.getitem(n_id, extract=False)
if details:
details = profile_service.resolve_node_data(details)
name = n_id.split(" @" )[0]
location = n_id.partition(" @" )[2] or " root"
# Prepare context for .format() with all details
context = details.copy()
context.update({
" name" : name,
" NAME" : name.upper(),
" location" : location,
" LOCATION" : location.upper(),
})
# Add exploded uniques (id, folder, subfolder)
uniques = self.config._explode_unique(n_id)
if uniques:
context.update(uniques)
# Add uppercase versions of all keys for convenience
for k, v in list(context.items()):
if isinstance(v, str):
context[k.upper()] = v.upper()
try:
formatted_nodes.append(format_str.format(**context))
except (KeyError, IndexError, ValueError):
# Fallback to original string if format fails
formatted_nodes.append(n_id)
return formatted_nodes</ code ></ pre >
</ details >
< div class = "desc" >< p > Return a listed filtered by regex match and formatted if needed.</ p ></ div >
</ dd >
< dt id = "connpy.services.node_service.NodeService.move_node" >< code class = "name flex" >
< span > def < span class = "ident" > move_node</ span ></ span > (< span > self, src_id, dst_id, copy=False)</ span >
</ code ></ dt >
< dd >
< details class = "source" >
< summary >
< span > Expand source code</ span >
</ summary >
< pre >< code class = "python" > def move_node(self, src_id, dst_id, copy=False):
""" Move or copy a node."""
self._validate_node_name(dst_id)
node_data = self.config.getitem(src_id)
if not node_data:
raise NodeNotFoundError(f" Source node ' {src_id}' not found." )
if dst_id in self.config._getallnodes():
raise NodeAlreadyExistsError(f" Destination node ' {dst_id}' already exists." )
new_uniques = self.config._explode_unique(dst_id)
if not new_uniques:
raise InvalidConfigurationError(f" Invalid destination format ' {dst_id}' ." )
new_node_data = node_data.copy()
new_node_data.update(new_uniques)
self.config._connections_add(**new_node_data)
if not copy:
src_uniques = self.config._explode_unique(src_id)
self.config._connections_del(**src_uniques)
self.config._saveconfig(self.config.file)</ code ></ pre >
</ details >
< div class = "desc" >< p > Move or copy a node.</ p ></ div >
</ dd >
< dt id = "connpy.services.node_service.NodeService.update_node" >< code class = "name flex" >
2026-05-28 18:22:00 -03:00
< span > def < span class = "ident" > update_node</ span ></ span > (< span > self, unique_id, data, save=True)</ span >
2026-04-17 18:42:08 -03:00
</ code ></ dt >
< dd >
< details class = "source" >
< summary >
< span > Expand source code</ span >
</ summary >
2026-05-28 18:22:00 -03:00
< pre >< code class = "python" > def update_node(self, unique_id, data, save=True):
2026-04-17 18:42:08 -03:00
""" Explicitly update an existing node."""
all_nodes = self.config._getallnodes()
if unique_id not in all_nodes:
raise NodeNotFoundError(f" Node ' {unique_id}' not found." )
# Ensure ' id' is in data for config._connections_add
if " id" not in data:
uniques = self.config._explode_unique(unique_id)
if uniques:
data[" id" ] = uniques[" id" ]
# config._connections_add actually handles updates if ID exists correctly
self.config._connections_add(**data)
2026-05-28 18:22:00 -03:00
if save:
self.config._saveconfig(self.config.file)</ code ></ pre >
2026-04-17 18:42:08 -03:00
</ details >
< div class = "desc" >< p > Explicitly update an existing node.</ p ></ div >
</ dd >
2026-04-24 19:23:00 -03:00
< dt id = "connpy.services.node_service.NodeService.validate_parent_folder" >< code class = "name flex" >
2026-05-01 18:55:25 -03:00
< span > def < span class = "ident" > validate_parent_folder</ span ></ span > (< span > self, unique_id, is_folder=False)</ span >
2026-04-24 19:23:00 -03:00
</ code ></ dt >
< dd >
< details class = "source" >
< summary >
< span > Expand source code</ span >
</ summary >
2026-05-01 18:55:25 -03:00
< pre >< code class = "python" > def validate_parent_folder(self, unique_id, is_folder=False):
2026-04-24 19:23:00 -03:00
""" Check if parent folder exists for a given node unique ID."""
2026-05-01 18:55:25 -03:00
if is_folder:
uniques = self.config._explode_unique(unique_id)
if uniques and " subfolder" in uniques and " folder" in uniques:
parent_folder = f" @{uniques[' folder' ]}"
if parent_folder not in self.config._getallfolders():
raise NodeNotFoundError(f" Folder ' {parent_folder}' not found." )
else:
node_folder = unique_id.partition(" @" )[2]
if node_folder:
parent_folder = f" @{node_folder}"
if parent_folder not in self.config._getallfolders():
raise NodeNotFoundError(f" Folder ' {parent_folder}' not found." )</ code ></ pre >
2026-04-24 19:23:00 -03:00
</ details >
< div class = "desc" >< p > Check if parent folder exists for a given node unique ID.</ p ></ div >
</ dd >
2026-04-17 18:42:08 -03:00
</ dl >
< h3 > Inherited members</ h3 >
< ul class = "hlist" >
< li >< code >< b >< a title = "connpy.services.base.BaseService" href = "base.html#connpy.services.base.BaseService" > BaseService</ a ></ b ></ code > :
< ul class = "hlist" >
< li >< code >< a title = "connpy.services.base.BaseService.set_reserved_names" href = "base.html#connpy.services.base.BaseService.set_reserved_names" > set_reserved_names</ a ></ code ></ li >
</ ul >
</ li >
</ ul >
</ dd >
</ dl >
</ section >
</ article >
< nav id = "sidebar" >
< div class = "toc" >
< ul ></ ul >
</ div >
< ul id = "index" >
< li >< h3 > Super-module</ h3 >
< ul >
< li >< code >< a title = "connpy.services" href = "index.html" > connpy.services</ a ></ code ></ li >
</ ul >
</ li >
< li >< h3 >< a href = "#header-classes" > Classes</ a ></ h3 >
< ul >
< li >
< h4 >< code >< a title = "connpy.services.node_service.NodeService" href = "#connpy.services.node_service.NodeService" > NodeService</ a ></ code ></ h4 >
2026-04-24 19:23:00 -03:00
< ul class = "" >
2026-04-17 18:42:08 -03:00
< li >< code >< a title = "connpy.services.node_service.NodeService.add_node" href = "#connpy.services.node_service.NodeService.add_node" > add_node</ a ></ code ></ li >
< li >< code >< a title = "connpy.services.node_service.NodeService.bulk_add" href = "#connpy.services.node_service.NodeService.bulk_add" > bulk_add</ a ></ code ></ li >
< li >< code >< a title = "connpy.services.node_service.NodeService.connect_node" href = "#connpy.services.node_service.NodeService.connect_node" > connect_node</ a ></ code ></ li >
< li >< code >< a title = "connpy.services.node_service.NodeService.delete_node" href = "#connpy.services.node_service.NodeService.delete_node" > delete_node</ a ></ code ></ li >
< li >< code >< a title = "connpy.services.node_service.NodeService.explode_unique" href = "#connpy.services.node_service.NodeService.explode_unique" > explode_unique</ a ></ code ></ li >
< li >< code >< a title = "connpy.services.node_service.NodeService.full_replace" href = "#connpy.services.node_service.NodeService.full_replace" > full_replace</ a ></ code ></ li >
< li >< code >< a title = "connpy.services.node_service.NodeService.generate_cache" href = "#connpy.services.node_service.NodeService.generate_cache" > generate_cache</ a ></ code ></ li >
< li >< code >< a title = "connpy.services.node_service.NodeService.get_inventory" href = "#connpy.services.node_service.NodeService.get_inventory" > get_inventory</ a ></ code ></ li >
< li >< code >< a title = "connpy.services.node_service.NodeService.get_node_details" href = "#connpy.services.node_service.NodeService.get_node_details" > get_node_details</ a ></ code ></ li >
< li >< code >< a title = "connpy.services.node_service.NodeService.list_folders" href = "#connpy.services.node_service.NodeService.list_folders" > list_folders</ a ></ code ></ li >
< li >< code >< a title = "connpy.services.node_service.NodeService.list_nodes" href = "#connpy.services.node_service.NodeService.list_nodes" > list_nodes</ a ></ code ></ li >
< li >< code >< a title = "connpy.services.node_service.NodeService.move_node" href = "#connpy.services.node_service.NodeService.move_node" > move_node</ a ></ code ></ li >
< li >< code >< a title = "connpy.services.node_service.NodeService.update_node" href = "#connpy.services.node_service.NodeService.update_node" > update_node</ a ></ code ></ li >
2026-04-24 19:23:00 -03:00
< li >< code >< a title = "connpy.services.node_service.NodeService.validate_parent_folder" href = "#connpy.services.node_service.NodeService.validate_parent_folder" > validate_parent_folder</ a ></ code ></ li >
2026-04-17 18:42:08 -03:00
</ ul >
</ li >
</ ul >
</ li >
</ ul >
</ nav >
</ main >
< footer id = "footer" >
2026-05-28 18:22:00 -03:00
< p > Generated by < a href = "https://pdoc3.github.io/pdoc" title = "pdoc: Python API documentation generator" >< cite > pdoc</ cite > 0.11.5</ a > .</ p >
2026-04-17 18:42:08 -03:00
</ footer >
</ body >
</ html >
