fix typing

logprep/processor/base/rule.py

@@ -127,9 +127,7 @@
 """
 
 import hashlib
-import json
 from functools import cached_property
-from os.path import basename, splitext
 from typing import Dict, List, Optional, Set
 
 from attrs import define, field, validators
@@ -140,7 +138,6 @@
 from logprep.filter.lucene_filter import LuceneFilter
 from logprep.metrics.metrics import CounterMetric, HistogramMetric
 from logprep.processor.base.exceptions import InvalidRuleDefinitionError
-from logprep.util.getter import GetterFactory
 from logprep.util.helper import camel_to_snake
 
 yaml = YAML(typ="safe", pure=True)
@@ -318,26 +315,6 @@ def lucene_filter(self):
 
     # pylint: enable=C0111
 
-    @classmethod
-    def create_rules_from_target(cls, rule_target: str, processor_name: str = None) -> list:
-        """Create a rule from a file."""
-        if isinstance(rule_target, dict):
-            return [cls._create_from_dict(rule_target, processor_name)]
-        content = GetterFactory.from_string(rule_target).get()
-        try:
-            rule_data = json.loads(content)
-        except ValueError:
-            rule_data = yaml.load_all(content)
-        try:
-            rules = [cls._create_from_dict(rule, processor_name) for rule in rule_data]
-        except InvalidRuleDefinitionError as error:
-            raise InvalidRuleDefinitionError(f"{rule_target}: {error}") from error
-        if len(rules) == 0:
-            raise InvalidRuleDefinitionError(f"no rules in file {rule_target}")
-        for rule in rules:
-            rule.file_name = splitext(basename(rule_target))[0]
-        return rules
-
     @classmethod
     def normalize_rule_dict(cls, rule: dict) -> None:
         """normalizes rule dict before create rule config object

logprep/util/rule_loader.py

@@ -31,7 +31,7 @@ class DirectoryRuleLoader(RuleLoader):
       Returns the list of rules loaded from the directory.
     """
 
-    def __init__(self, directory: str, rule_class: type = Rule):
+    def __init__(self, directory: str, rule_class: Rule = Rule):
         self.source = directory
         self.rule_class = rule_class
 
@@ -65,7 +65,7 @@ class FileRuleLoader(RuleLoader):
       Returns the list of rules loaded from the file.
     """
 
-    def __init__(self, file: str, rule_class: type = Rule):
+    def __init__(self, file: str, rule_class: Rule = Rule):
         self.rule_class = rule_class
         self.source = file
 
@@ -98,7 +98,7 @@ class ListRuleLoader(RuleLoader):
       Returns a list of Rule objects created from the source list of dictionaries.
     """
 
-    def __init__(self, rules: List[dict], rule_class: type = Rule):
+    def __init__(self, rules: List[dict], rule_class: Rule = Rule):
         self.rule_class = rule_class
         self.source = rules
 

tests/unit/processor/labeler/test_labeler_rule.py

@@ -11,7 +11,6 @@
 from logprep.processor.base.exceptions import InvalidRuleDefinitionError
 from logprep.processor.labeler.labeling_schema import LabelingSchema
 from logprep.processor.labeler.rule import LabelerRule
-from tests.testdata.FilledTempFile import JsonTempFile
 from tests.testdata.ruledata import simple_rule_dict
 
 
@@ -30,19 +29,6 @@ def get_parent_labels(self, category, label):
 
 
 class TestRule:
-    def test_create_from_file_fails_if_document_does_not_contain_filter_and_label(self):
-        with pytest.raises(InvalidRuleDefinitionError):
-            with JsonTempFile({}) as rule_path:
-                LabelerRule.create_rules_from_target(rule_path)
-
-        for missing_field in ["filter", "labeler"]:
-            invalid_rule_dict = deepcopy(simple_rule_dict)
-            del invalid_rule_dict[missing_field]
-
-            with pytest.raises(InvalidRuleDefinitionError):
-                with JsonTempFile([invalid_rule_dict]) as rule_path:
-                    LabelerRule.create_rules_from_target(rule_path)
-
     def test_create_from_dict_creates_expected_rule(self):
         rule_definition = {
             "filter": 'applyrule: "yes"',