more updates

Author: alexandraBara

nodescraper/base/inbandcollectortask.py

@@ -24,7 +24,7 @@
 #
 ###############################################################################
 import logging
-from typing import Generic, Optional
+from typing import Generic, Optional, Union
 
 from nodescraper.connection.inband import InBandConnection
 from nodescraper.connection.inband.inband import BaseFileArtifact, CommandArtifact
@@ -49,7 +49,7 @@ def __init__(
         connection: InBandConnection,
         logger: Optional[logging.Logger] = None,
         system_interaction_level: SystemInteractionLevel = SystemInteractionLevel.INTERACTIVE,
-        max_event_priority_level: EventPriority | str = EventPriority.CRITICAL,
+        max_event_priority_level: Union[EventPriority, str] = EventPriority.CRITICAL,
         parent: Optional[str] = None,
         task_result_hooks: Optional[list[TaskResultHook]] = None,
         **kwargs,

nodescraper/base/regexanalyzer.py

@@ -24,6 +24,7 @@
 #
 ###############################################################################
 import re
+from typing import Union
 
 from pydantic import BaseModel
 
@@ -36,7 +37,7 @@
 class ErrorRegex(BaseModel):
     regex: re.Pattern
     message: str
-    event_category: str | EventCategory = EventCategory.UNKNOWN
+    event_category: Union[str, EventCategory] = EventCategory.UNKNOWN
     event_priority: EventPriority = EventPriority.ERROR
 
 
@@ -54,17 +55,18 @@ class RegexAnalyzer(DataAnalyzer[TDataModel, TAnalyzeArg]):
     """Parent class for all regex based data analyzers."""
 
     def _build_regex_event(
-        self, regex_obj: ErrorRegex, match: str | list[str], source: str
+        self, regex_obj: ErrorRegex, match: Union[str, list[str]], source: str
     ) -> RegexEvent:
         """Build a RegexEvent object from a regex match and source.
 
-        Args:
-            regex_obj (ErrorRegex): regex object containing the regex pattern, message, category, and priorit
-            match (str | list[str]): matched content from the regex
-            source (str): descriptor for the content where the match was found
+                Args:
+                    regex_obj (ErrorRegex): regex object containing the regex pattern, message, category, and priorit
+                    match (
+        Union[str, list[str]]): matched content from the regex
+                    source (str): descriptor for the content where the match was found
 
-        Returns:
-            RegexEvent: an instance of RegexEvent containing the match details
+                Returns:
+                    RegexEvent: an instance of RegexEvent containing the match details
         """
         return RegexEvent(
             description=regex_obj.message,

nodescraper/cli/dynamicparserbuilder.py

@@ -24,7 +24,6 @@
 #
 ###############################################################################
 import argparse
-import types
 from typing import Optional, Type
 
 from pydantic import BaseModel
@@ -59,7 +58,7 @@ def build_plugin_parser(self) -> dict:
             }
 
             # skip args where generic type has been set to None
-            if types.NoneType in type_class_map:
+            if type(None) in type_class_map:
                 continue
 
             model_arg = self.get_model_arg(type_class_map)
@@ -164,7 +163,7 @@ def build_model_arg_parser(self, model: type[BaseModel], required: bool) -> list
                 type_class.type_class: type_class for type_class in attr_data.type_classes
             }
 
-            if types.NoneType in type_class_map and len(attr_data.type_classes) == 1:
+            if type(None) in type_class_map and len(attr_data.type_classes) == 1:
                 continue
 
             self.add_argument(type_class_map, attr.replace("_", "-"), required)

nodescraper/interfaces/connectionmanager.py

@@ -29,7 +29,7 @@
 import logging
 import types
 from functools import wraps
-from typing import Callable, Generic, Optional, TypeVar
+from typing import Callable, Generic, Optional, TypeVar, Union
 
 from pydantic import BaseModel
 
@@ -89,9 +89,9 @@ def __init__(
         self,
         system_info: SystemInfo,
         logger: Optional[logging.Logger] = None,
-        max_event_priority_level: EventPriority | str = EventPriority.CRITICAL,
+        max_event_priority_level: Union[EventPriority, str] = EventPriority.CRITICAL,
         parent: Optional[str] = None,
-        task_result_hooks: list[TaskResultHook] | types.NoneType = None,
+        task_result_hooks: Optional[list[TaskResultHook], None] = None,
         connection_args: Optional[TConnectArg | dict] = None,
         **kwargs,
     ):

nodescraper/interfaces/datacollectortask.py

@@ -27,7 +27,7 @@
 import inspect
 import logging
 from functools import wraps
-from typing import Callable, ClassVar, Generic, Optional, Type
+from typing import Callable, ClassVar, Generic, Optional, Type, Union
 
 from pydantic import BaseModel, ValidationError
 
@@ -123,7 +123,7 @@ def __init__(
         connection: TConnection,
         logger: Optional[logging.Logger] = None,
         system_interaction_level: SystemInteractionLevel | str = SystemInteractionLevel.INTERACTIVE,
-        max_event_priority_level: EventPriority | str = EventPriority.CRITICAL,
+        max_event_priority_level: Union[EventPriority, str] = EventPriority.CRITICAL,
         parent: Optional[str] = None,
         task_result_hooks: Optional[list[TaskResultHook]] = None,
         **kwargs,

nodescraper/interfaces/dataplugin.py

@@ -24,7 +24,7 @@
 #
 ###############################################################################
 import logging
-from typing import Generic, Optional, Type
+from typing import Generic, Optional, Type, Union
 
 from nodescraper.enums import EventPriority, ExecutionStatus, SystemInteractionLevel
 from nodescraper.generictypes import TAnalyzeArg, TCollectArg, TDataModel
@@ -64,7 +64,7 @@ def __init__(
         system_info: SystemInfo,
         logger: Optional[logging.Logger] = None,
         connection_manager: Optional[TConnectionManager] = None,
-        connection_args: Optional[TConnectArg | dict] = None,
+        connection_args: Optional[Union[TConnectArg, dict]] = None,
         task_result_hooks: Optional[list[TaskResultHook]] = None,
         log_path: Optional[str] = None,
         **kwargs,
@@ -127,7 +127,7 @@ def data(self) -> Optional[TDataModel]:
         return self._data
 
     @data.setter
-    def data(self, data: str | dict | TDataModel):
+    def data(self, data: Optional[str, dict, TDataModel]):
         if isinstance(data, (str, dict)):
             self._data = self.DATA_MODEL.import_model(data)
         elif not isinstance(data, self.DATA_MODEL):
@@ -137,15 +137,17 @@ def data(self, data: str | dict | TDataModel):
 
     def collect(
         self,
-        max_event_priority_level: EventPriority | str = EventPriority.CRITICAL,
-        system_interaction_level: SystemInteractionLevel | str = SystemInteractionLevel.INTERACTIVE,
+        max_event_priority_level: Optional[Union[EventPriority, str]] = EventPriority.CRITICAL,
+        system_interaction_level: Optional[
+            Union[SystemInteractionLevel, str]
+        ] = SystemInteractionLevel.INTERACTIVE,
         preserve_connection: bool = False,
-        collection_args: Optional[TCollectArg | dict] = None,
+        collection_args: Optional[Union[TCollectArg, dict]] = None,
     ) -> TaskResult:
         """Run data collector task
 
         Args:
-            max_event_priority_level (EventPriority | str, optional): priority limit for events. Defaults to EventPriority.CRITICAL.
+            max_event_priority_level (Union[EventPriority, str], optional): priority limit for events. Defaults to EventPriority.CRITICAL.
             system_interaction_level (SystemInteractionLevel | str, optional): system interaction level. Defaults to SystemInteractionLevel.INTERACTIVE.
             preserve_connection (bool, optional): whether we should close the connection after data collection. Defaults to False.
             collection_args (Optional[TCollectArg  |  dict], optional): args for data collection. Defaults to None.
@@ -227,16 +229,16 @@ def collect(
 
     def analyze(
         self,
-        max_event_priority_level: EventPriority | str = EventPriority.CRITICAL,
+        max_event_priority_level: Optional[Union[EventPriority, str]] = EventPriority.CRITICAL,
         analysis_args: Optional[TAnalyzeArg | dict] = None,
-        data: Optional[str | dict | TDataModel] = None,
+        data: Optional[Union[str, dict, TDataModel]] = None,
     ) -> TaskResult:
         """Run data analyzer task
 
         Args:
-            max_event_priority_level (EventPriority | str, optional): priority limit for events. Defaults to EventPriority.CRITICAL.
+            max_event_priority_level (Union[EventPriority, str], optional): priority limit for events. Defaults to EventPriority.CRITICAL.
             analysis_args (Optional[TAnalyzeArg  |  dict], optional): args for data analysis. Defaults to None.
-            data (Optional[str  |  dict  |  TDataModel], optional): data to analyze. Defaults to None.
+            data (Optional[Union[str, dict, TDataModel]], optional): data to analyze. Defaults to None.
 
         Returns:
             TaskResult: result of data analysis
@@ -276,10 +278,10 @@ def run(
         self,
         collection: bool = True,
         analysis: bool = True,
-        max_event_priority_level: EventPriority | str = EventPriority.CRITICAL,
+        max_event_priority_level: Union[EventPriority, str] = EventPriority.CRITICAL,
         system_interaction_level: SystemInteractionLevel | str = SystemInteractionLevel.INTERACTIVE,
         preserve_connection: bool = False,
-        data: Optional[str | dict | TDataModel] = None,
+        data: Optional[Union[str, dict, TDataModel]] = None,
         collection_args: Optional[TCollectArg | dict] = None,
         analysis_args: Optional[TAnalyzeArg | dict] = None,
     ) -> PluginResult:
@@ -288,10 +290,10 @@ def run(
         Args:
             collection (bool, optional): Enable data collection. Defaults to True.
             analysis (bool, optional): Enable data analysis. Defaults to True.
-            max_event_priority_level (EventPriority | str, optional): Max priority level to assign to events. Defaults to EventPriority.CRITICAL.
+            max_event_priority_level (Union[EventPriority, str], optional): Max priority level to assign to events. Defaults to EventPriority.CRITICAL.
             system_interaction_level (SystemInteractionLevel | str, optional): System interaction level. Defaults to SystemInteractionLevel.INTERACTIVE.
             preserve_connection (bool, optional): Whether to close the connection when data collection is complete. Defaults to False.
-            data (Optional[str  |  dict  |  TDataModel], optional): Input data. Defaults to None.
+            data (Optional[Union[str, dict, TDataModel]], optional): Input data. Defaults to None.
             collection_args (Optional[TCollectArg  |  dict], optional): Arguments for data collection. Defaults to None.
             analysis_args (Optional[TAnalyzeArg  |  dict], optional): Arguments for data analysis. Defaults to None.
 

nodescraper/interfaces/task.py

@@ -27,7 +27,7 @@
 import copy
 import datetime
 import logging
-from typing import Any, Optional
+from typing import Any, Optional, Union
 
 from nodescraper.constants import DEFAULT_LOGGER
 from nodescraper.enums import EventCategory, EventPriority
@@ -51,7 +51,7 @@ def __init__(
         self,
         system_info: SystemInfo,
         logger: Optional[logging.Logger] = None,
-        max_event_priority_level: EventPriority | str = EventPriority.CRITICAL,
+        max_event_priority_level: Union[EventPriority, str] = EventPriority.CRITICAL,
         parent: Optional[str] = None,
         task_result_hooks: Optional[list[TaskResultHook]] = None,
         **kwargs: dict[str, Any],
@@ -96,7 +96,7 @@ def __init_subclass__(cls, **kwargs) -> None:
 
     def _build_event(
         self,
-        category: EventCategory | str,
+        category: Union[EventCategory, str],
         description: str,
         priority: EventPriority,
         data: Optional[dict] = None,
@@ -133,7 +133,7 @@ def _build_event(
 
     def _log_event(
         self,
-        category: EventCategory | str,
+        category: Union[EventCategory, str],
         description: str,
         priority: EventPriority,
         data: Optional[dict] = None,

nodescraper/models/datamodel.py

@@ -27,7 +27,7 @@
 import json
 import os
 import tarfile
-from typing import TypeVar
+from typing import TypeVar, Union
 
 from pydantic import BaseModel, Field, field_validator
 
@@ -42,7 +42,7 @@ class FileModel(BaseModel):
 
     @field_validator("file_contents", mode="before")
     @classmethod
-    def file_contents_conformer(cls, value: io.BytesIO | str | bytes) -> bytes:
+    def file_contents_conformer(cls, value: Union[io.BytesIO, str, bytes]) -> bytes:
         if isinstance(value, io.BytesIO):
             return value.getvalue()
         if isinstance(value, str):
@@ -92,15 +92,15 @@ def merge_data(self, input_data: "DataModel") -> None:
         pass
 
     @classmethod
-    def import_model(cls: type[TDataModel], model_input: dict | str) -> TDataModel:
+    def import_model(cls: type[TDataModel], model_input: Union[dict, str]) -> TDataModel:
         """import a data model
         if the input is a string attempt to read data from file using the string as a file name
         if input is a dict, pass key value pairs directly to init function
 
 
         Args:
             cls (type[DataModel]): Data model class
-            model_input (dict | str): model data input
+            model_input (Union[dict, str]): model data input
 
         Raises:
             ValueError: if model_input has an invalid type

nodescraper/models/event.py

@@ -28,7 +28,7 @@
 import re
 import uuid
 from enum import Enum
-from typing import Optional
+from typing import Optional, Union
 
 from pydantic import BaseModel, Field, field_serializer, field_validator
 
@@ -80,7 +80,7 @@ def validate_timestamp(cls, timestamp: datetime.datetime) -> datetime.datetime:
 
     @field_validator("category", mode="before")
     @classmethod
-    def validate_category(cls, category: str | Enum) -> str:
+    def validate_category(cls, category: Optional[Union[str, Enum]]) -> str:
         """ensure category is has consistent formatting
         Args:
             category (str | Enum): category string
@@ -96,7 +96,7 @@ def validate_category(cls, category: str | Enum) -> str:
 
     @field_validator("priority", mode="before")
     @classmethod
-    def validate_priority(cls, priority: str | EventPriority) -> EventPriority:
+    def validate_priority(cls, priority: Optional[Union[str, EventPriority]]) -> EventPriority:
         """Allow priority to be set via string priority name
         Args:
             priority (str | EventPriority): event priority string or enum

nodescraper/models/pluginresult.py

@@ -23,7 +23,7 @@
 # SOFTWARE.
 #
 ###############################################################################
-from typing import Optional
+from typing import Optional, Union
 
 from pydantic import BaseModel
 
@@ -36,4 +36,4 @@ class PluginResult(BaseModel):
     status: ExecutionStatus
     source: str
     message: Optional[str] = None
-    result_data: Optional[dict | BaseModel] = None
+    result_data: Optional[Union[dict, BaseModel]] = None