nornsible
-def Init_Nornsible(nr)
+
+def InitNornsible(nr)
-
Patch nornir object based on cli arguments
@@ -82,7 +90,7 @@ Raises
def Init_Nornsible(nr):
+def InitNornsible(nr):
"""
Patch nornir object based on cli arguments
@@ -109,6 +117,105 @@ Raises
return nr
+def nornsible_task(wrapped_func)
+Decorate an "operation" – execute or skip the operation based on tags
+wrapped_functag_wrapperN/A
+# noqadef nornsible_task(wrapped_func):
+ """
+ Decorate an "operation" -- execute or skip the operation based on tags
+
+ Args:
+ wrapped_func: function to wrap in tag processor
+
+ Returns:
+ tag_wrapper: wrapped function
+
+ Raises:
+ N/A # noqa
+
+ """
+
+ def tag_wrapper(task, *args, **kwargs):
+ if {wrapped_func.__name__}.intersection(task.nornir.skip_tags):
+ msg = f"---- {task.host} skipping task {wrapped_func.__name__} "
+ nornsible_task_message(msg)
+ return Result(host=task.host, result="Task skipped!", failed=False, changed=False)
+ if not task.nornir.run_tags:
+ return wrapped_func(task, *args, **kwargs)
+ if {wrapped_func.__name__}.intersection(task.nornir.run_tags):
+ return wrapped_func(task, *args, **kwargs)
+ msg = f"---- {task.host} skipping task {wrapped_func.__name__} "
+ nornsible_task_message(msg)
+ return Result(host=task.host, result="Task skipped!", failed=False, changed=False)
+
+ tag_wrapper.__name__ = wrapped_func.__name__
+ return tag_wrapper
+def nornsible_task_message(msg)
+Handle printing pretty messages for nornsible_task decorator
+msgN/AN/A
+# noqadef nornsible_task_message(msg):
+ """
+ Handle printing pretty messages for nornsible_task decorator
+
+ Args:
+ msg: message to beautifully print to stdout
+
+ Returns:
+ N/A
+
+ Raises:
+ N/A # noqa
+
+ """
+ LOCK.acquire()
+ try:
+ print(f"{Style.BRIGHT}{Back.CYAN}{Fore.WHITE}{msg}{'-' * (80 - len(msg))}")
+ finally:
+ LOCK.release()
def parse_cli_args(args)
-def process_tags(wrapped_func)
-Decorate an "operation" – execute or skip the operation based on tags
-wrapped_functag_wrapperN/A
-# noqadef process_tags(wrapped_func):
- """
- Decorate an "operation" -- execute or skip the operation based on tags
-
- Args:
- wrapped_func: function to wrap in tag processor
-
- Returns:
- tag_wrapper: wrapped function
-
- Raises:
- N/A # noqa
-
- """
-
- def tag_wrapper(task, *args, **kwargs):
- if set([wrapped_func.__name__]).intersection(task.nornir.skip_tags):
- msg = f"---- {task.host} skipping task {wrapped_func.__name__} "
- process_tags_messages(msg)
- return Result(host=task.host, result="Task skipped!", failed=False, changed=False)
- if not task.nornir.run_tags:
- return wrapped_func(task, *args, **kwargs)
- if set([wrapped_func.__name__]).intersection(task.nornir.run_tags):
- return wrapped_func(task, *args, **kwargs)
- msg = f"---- {task.host} skipping task {wrapped_func.__name__} "
- process_tags_messages(msg)
- return Result(host=task.host, result="Task skipped!", failed=False, changed=False)
-
- tag_wrapper.__name__ = wrapped_func.__name__
- return tag_wrappernornsible.nornsibleimport argparse
-import logging
import sys
import threading
-from colorama import Fore, Style, init
+from colorama import Back, Fore, init, Style
from nornir.core.task import Result
-session_log = logging.getLogger(__name__)
-
init(autoreset=True, strip=False)
LOCK = threading.Lock()
@@ -150,9 +147,9 @@ Module nornsible.nornsible
return conf
-def process_tags_messages(msg):
+def nornsible_task_message(msg):
"""
- Handle printing pretty messages for process_tags decorator
+ Handle printing pretty messages for nornsible_task decorator
Args:
msg: message to beautifully print to stdout
@@ -166,12 +163,12 @@ Module nornsible.nornsible
"""
LOCK.acquire()
try:
- print("{}{}{}{}".format(Style.BRIGHT, Fore.CYAN, msg, "-" * (80 - len(msg))))
+ print(f"{Style.BRIGHT}{Back.CYAN}{Fore.WHITE}{msg}{'-' * (80 - len(msg))}")
finally:
LOCK.release()
-def process_tags(wrapped_func):
+def nornsible_task(wrapped_func):
"""
Decorate an "operation" -- execute or skip the operation based on tags
@@ -187,23 +184,23 @@ Module nornsible.nornsible
"""
def tag_wrapper(task, *args, **kwargs):
- if set([wrapped_func.__name__]).intersection(task.nornir.skip_tags):
+ if {wrapped_func.__name__}.intersection(task.nornir.skip_tags):
msg = f"---- {task.host} skipping task {wrapped_func.__name__} "
- process_tags_messages(msg)
+ nornsible_task_message(msg)
return Result(host=task.host, result="Task skipped!", failed=False, changed=False)
if not task.nornir.run_tags:
return wrapped_func(task, *args, **kwargs)
- if set([wrapped_func.__name__]).intersection(task.nornir.run_tags):
+ if {wrapped_func.__name__}.intersection(task.nornir.run_tags):
return wrapped_func(task, *args, **kwargs)
msg = f"---- {task.host} skipping task {wrapped_func.__name__} "
- process_tags_messages(msg)
+ nornsible_task_message(msg)
return Result(host=task.host, result="Task skipped!", failed=False, changed=False)
tag_wrapper.__name__ = wrapped_func.__name__
return tag_wrapper
-def Init_Nornsible(nr):
+def InitNornsible(nr):
"""
Patch nornir object based on cli arguments
@@ -237,8 +234,8 @@ Module nornsible.nornsible
Functions
-
-def Init_Nornsible(nr)
+
+def InitNornsible(nr)
-
Patch nornir object based on cli arguments
@@ -260,7 +257,7 @@ Raises
Source code
-def Init_Nornsible(nr):
+def InitNornsible(nr):
"""
Patch nornir object based on cli arguments
@@ -287,6 +284,105 @@ Raises
return nr
+
+def nornsible_task(wrapped_func)
+
+Decorate an "operation" – execute or skip the operation based on tags
+Args
+
+wrapped_func- function to wrap in tag processor
+
+Returns
+
+tag_wrapper- wrapped function
+
+Raises
+
+N/A
+# noqa-
+
Source code
+def nornsible_task(wrapped_func):
+ """
+ Decorate an "operation" -- execute or skip the operation based on tags
+
+ Args:
+ wrapped_func: function to wrap in tag processor
+
+ Returns:
+ tag_wrapper: wrapped function
+
+ Raises:
+ N/A # noqa
+
+ """
+
+ def tag_wrapper(task, *args, **kwargs):
+ if {wrapped_func.__name__}.intersection(task.nornir.skip_tags):
+ msg = f"---- {task.host} skipping task {wrapped_func.__name__} "
+ nornsible_task_message(msg)
+ return Result(host=task.host, result="Task skipped!", failed=False, changed=False)
+ if not task.nornir.run_tags:
+ return wrapped_func(task, *args, **kwargs)
+ if {wrapped_func.__name__}.intersection(task.nornir.run_tags):
+ return wrapped_func(task, *args, **kwargs)
+ msg = f"---- {task.host} skipping task {wrapped_func.__name__} "
+ nornsible_task_message(msg)
+ return Result(host=task.host, result="Task skipped!", failed=False, changed=False)
+
+ tag_wrapper.__name__ = wrapped_func.__name__
+ return tag_wrapper
+
+
+
+def nornsible_task_message(msg)
+
+Handle printing pretty messages for nornsible_task decorator
+Args
+
+msg- message to beautifully print to stdout
+
+Returns
+
+N/A-
+
+Raises
+
+N/A
+# noqa-
+
Source code
+def nornsible_task_message(msg):
+ """
+ Handle printing pretty messages for nornsible_task decorator
+
+ Args:
+ msg: message to beautifully print to stdout
+
+ Returns:
+ N/A
+
+ Raises:
+ N/A # noqa
+
+ """
+ LOCK.acquire()
+ try:
+ print(f"{Style.BRIGHT}{Back.CYAN}{Fore.WHITE}{msg}{'-' * (80 - len(msg))}")
+ finally:
+ LOCK.release()
+
+
def parse_cli_args(args)
Raises
return inv
-def process_tags(wrapped_func)
-Decorate an "operation" – execute or skip the operation based on tags
-wrapped_functag_wrapperN/A
-# noqadef process_tags(wrapped_func):
- """
- Decorate an "operation" -- execute or skip the operation based on tags
-
- Args:
- wrapped_func: function to wrap in tag processor
-
- Returns:
- tag_wrapper: wrapped function
-
- Raises:
- N/A # noqa
-
- """
-
- def tag_wrapper(task, *args, **kwargs):
- if set([wrapped_func.__name__]).intersection(task.nornir.skip_tags):
- msg = f"---- {task.host} skipping task {wrapped_func.__name__} "
- process_tags_messages(msg)
- return Result(host=task.host, result="Task skipped!", failed=False, changed=False)
- if not task.nornir.run_tags:
- return wrapped_func(task, *args, **kwargs)
- if set([wrapped_func.__name__]).intersection(task.nornir.run_tags):
- return wrapped_func(task, *args, **kwargs)
- msg = f"---- {task.host} skipping task {wrapped_func.__name__} "
- process_tags_messages(msg)
- return Result(host=task.host, result="Task skipped!", failed=False, changed=False)
-
- tag_wrapper.__name__ = wrapped_func.__name__
- return tag_wrapper
-def process_tags_messages(msg)
-Handle printing pretty messages for process_tags decorator
-msgN/AN/A
-# noqadef process_tags_messages(msg):
- """
- Handle printing pretty messages for process_tags decorator
-
- Args:
- msg: message to beautifully print to stdout
-
- Returns:
- N/A
-
- Raises:
- N/A # noqa
-
- """
- LOCK.acquire()
- try:
- print("{}{}{}{}".format(Style.BRIGHT, Fore.CYAN, msg, "-" * (80 - len(msg))))
- finally:
- LOCK.release()