Make python models a bit more sane (#10)

Author: glossawy

README.md

@@ -6,7 +6,7 @@ Sometimes you want to modify the content or structure of an Atom or RSS feed, sa
 
 ## Concepts
 
-At the foundation, transformation rules are constructed of a `Condition` (which can be made up of many conditions) and a set of `Mutation`s to perform. Together a condition + mutations makes a `Rule`. A feed URL and a set of rules is a feed transformation. Elements of a feed to apply a condition or rule to are determined using [XPath queries](https://developer.mozilla.org/en-US/docs/Web/XPath), an XPath starting with a slack (`/`) indicates an absolute XPath, an XPath without a leading slash is assumed to be relative to the rule's XPath. XPaths are optional for conditions and mutations, if they are missing they are applied to the elements matched by the rule's XPath.
+At the foundation, transformation rules are constructed of a `Condition` (which can be made up of many conditions) and a set of `Mutation`s to perform. Together a condition + mutations makes a `Rule`. A feed URL and a set of rules is a `Feed Transform`. Elements of a feed to apply a condition or rule to are determined using [XPath queries](https://developer.mozilla.org/en-US/docs/Web/XPath), an XPath starting with a slash (`/`) indicates an absolute XPath, an XPath without a leading slash is assumed to be relative to the rule's XPath. XPaths are optional for conditions and mutations, if they are missing they are applied to the elements matched by the rule's XPath.
 
 Currently the app supports the following conditions:
 
@@ -15,7 +15,7 @@ contains:
    Tests whether or not the text of the element contains a substring
 
    args:
-      value: Substring to match
+      pattern: Regular expression to match
 ```
 
 and the following mutations:
@@ -53,4 +53,3 @@ This uses `docker compose` to spin up the application locally hosted at `localho
 
 - For frontend only work: `make dev-frontend` or `bun run dev`
 - For API work: `make dev-backend`
-

app/feed_editor/rewrite/compression.py

@@ -7,18 +7,18 @@
 from feed_editor.rewrite.rules.mutations import MutationArgs
 from feed_editor.rewrite.rules.types import (
     ConditionDict,
-    FeedRulesDict,
+    FeedTransformDict,
     MutationDict,
-    MutationDictWithoutXPath,
     RuleDict,
-    SingleCondition,
-    SingleConditionWithoutXPath,
+    SingleConditionDict,
 )
 from feed_editor.utils.dict_validation import validate_dict
 
 # Each key in the feed dict must have a unique mapping here
 # otherwise it wont be minified. If there is a collision then
 # decoding will return invalid results.
+# used:
+# abcdefgmnoqrstuwyx
 KEY_MINIFY_MAP = {
     "feed_url": "f",
     "rules": "q",
@@ -28,6 +28,10 @@
     "xpath": "x",
     "name": "n",
     "args": "b",
+    # Feed Transform
+    "version": "g",
+    # Rules
+    "rid": "e",
     # Conditions
     "contains": "c",
     "all_of": "a",
@@ -36,8 +40,6 @@
     "remove": "r",
     "replace": "s",
     "changeTag": "t",
-    # contains
-    "value": "v",
     # replace
     "pattern": "p",
     "replacement": "u",
@@ -49,15 +51,15 @@
 KEY_EXPAND_MAP = {v: k for k, v in KEY_MINIFY_MAP.items()}
 
 
-def compress_and_encode(rules: FeedRulesDict) -> str:
+def compress_and_encode(rules: FeedTransformDict) -> str:
     """
     Takes a feed url and its transforms, minifies the dict, compresses it
     and urlsafe b64 encodes it
     """
     return _gzip_encode(_simplify_feed_dict(rules))
 
 
-def decode_and_decompress(encoded: str) -> FeedRulesDict:
+def decode_and_decompress(encoded: str) -> FeedTransformDict:
     """
     Takes a minified version of the feed rules dict and decodes it, decompresses it,
     and un-minifies it.
@@ -87,11 +89,11 @@ def _json_dumps(data: Mapping) -> str:
     return json.dumps(data, separators=(",", ":"), indent=None)
 
 
-def _simplify_feed_dict(rules: FeedRulesDict) -> dict:
+def _simplify_feed_dict(rules: FeedTransformDict) -> dict:
     def simplify_dict(args: Mapping) -> dict:
         return {KEY_MINIFY_MAP[k]: v for k, v in args.items()}
 
-    def simplify_typed_dict(typed: Union[MutationDict, SingleCondition]) -> dict:
+    def simplify_typed_dict(typed: Union[MutationDict, SingleConditionDict]) -> dict:
         simple_dict = {
             KEY_MINIFY_MAP["name"]: typed["name"],
             KEY_MINIFY_MAP["args"]: simplify_dict(typed["args"]),
@@ -122,6 +124,8 @@ def simplify_mutation(mut: MutationDict) -> dict:
 
     def simplify_rule(rule: RuleDict):
         return {
+            KEY_MINIFY_MAP["rid"]: rule["rid"],
+            KEY_MINIFY_MAP["name"]: rule["name"],
             KEY_MINIFY_MAP["xpath"]: rule["xpath"],
             KEY_MINIFY_MAP["condition"]: simplify_condition(rule["condition"]),
             KEY_MINIFY_MAP["mutations"]: [
@@ -130,12 +134,13 @@ def simplify_rule(rule: RuleDict):
         }
 
     return {
+        KEY_MINIFY_MAP["version"]: rules["version"],
         KEY_MINIFY_MAP["feed_url"]: rules["feed_url"],
         KEY_MINIFY_MAP["rules"]: [simplify_rule(rule) for rule in rules["rules"]],
     }
 
 
-def _feedify_simple_dict(simple_dict: dict) -> FeedRulesDict:
+def _feedify_simple_dict(simple_dict: dict) -> FeedTransformDict:
     def feedify_condition_args_dict(simple_args: dict) -> ConditionArgs:
         return cast(
             ConditionArgs, {KEY_EXPAND_MAP[k]: v for k, v in simple_args.items()}
@@ -161,31 +166,33 @@ def feedify_condition(simple_condition: dict) -> ConditionDict:
                     for sc in simple_condition[KEY_MINIFY_MAP["any_of"]]
                 ]
             }
-        cond_dict: SingleConditionWithoutXPath = {
+        cond_dict: SingleConditionDict = {
             "name": simple_condition[KEY_MINIFY_MAP["name"]],
             "args": feedify_condition_args_dict(
                 simple_condition[KEY_MINIFY_MAP["args"]]
             ),
         }
 
         if KEY_MINIFY_MAP["xpath"] in simple_condition:
-            return {**cond_dict, "xpath": simple_condition[KEY_MINIFY_MAP["xpath"]]}
+            cond_dict["xpath"] = simple_condition[KEY_MINIFY_MAP["xpath"]]
 
         return cond_dict
 
     def feedify_mutation(simple_mutation: dict) -> MutationDict:
-        mut_dict: MutationDictWithoutXPath = {
+        mut_dict: MutationDict = {
             "name": simple_mutation[KEY_MINIFY_MAP["name"]],
             "args": feedify_mutation_args_dict(simple_mutation[KEY_MINIFY_MAP["args"]]),
         }
 
         if KEY_MINIFY_MAP["xpath"] in simple_mutation:
-            return {**mut_dict, "xpath": simple_mutation[KEY_MINIFY_MAP["xpath"]]}
+            mut_dict["xpath"] = simple_mutation[KEY_MINIFY_MAP["xpath"]]
 
         return mut_dict
 
     def feedify_rule(simple_rule: dict) -> RuleDict:
         return {
+            "rid": simple_rule[KEY_MINIFY_MAP["rid"]],
+            "name": simple_rule[KEY_MINIFY_MAP["name"]],
             "xpath": simple_rule[KEY_MINIFY_MAP["xpath"]],
             "condition": feedify_condition(simple_rule[KEY_MINIFY_MAP["condition"]]),
             "mutations": [
@@ -198,11 +205,12 @@ def feedify_rule(simple_rule: dict) -> RuleDict:
         KEY_MINIFY_MAP["feed_url"] not in simple_dict
         or KEY_MINIFY_MAP["rules"] not in simple_dict
     ):
-        return validate_dict(FeedRulesDict, simple_dict)
+        return validate_dict(FeedTransformDict, simple_dict)
 
     return validate_dict(
-        FeedRulesDict,
+        FeedTransformDict,
         {
+            "version": simple_dict[KEY_MINIFY_MAP["version"]],
             "feed_url": simple_dict[KEY_MINIFY_MAP["feed_url"]],
             "rules": [
                 feedify_rule(rule) for rule in simple_dict[KEY_MINIFY_MAP["rules"]]

app/feed_editor/rewrite/rewriter.py

@@ -5,7 +5,7 @@
 import validators  # type: ignore
 
 from feed_editor.rewrite.rules import run_rule
-from feed_editor.rewrite.rules.types import FeedRulesDict
+from feed_editor.rewrite.rules.types import FeedTransformDict
 from feed_editor.rss.errors import FeedError
 from feed_editor.rss.fetch import Feed
 from feed_editor.rss.fetch import fetch_feed as rss_fetch
@@ -19,7 +19,7 @@ class FeedRewriter:
     to the rewritten feed
     """
 
-    feed_rules: FeedRulesDict
+    feed_transform: FeedTransformDict
 
     @property
     def is_valid_feed(self) -> bool:
@@ -29,7 +29,7 @@ def is_valid_feed(self) -> bool:
     @property
     def feed_url(self) -> str:
         """Get the URL of the feed being rewritten"""
-        return self.feed_rules["feed_url"]
+        return self.feed_transform["feed_url"]
 
     @cached_property
     def feed(self) -> Feed:
@@ -41,7 +41,7 @@ def rewritten_feed(self) -> Feed:
         """A rewritten version of the original feed after all rules are applied"""
         new_feed = self.feed.copy()
 
-        for rule_dict in self.feed_rules["rules"]:
+        for rule_dict in self.feed_transform["rules"]:
             run_rule(new_feed.tree, rule_dict)
 
         return new_feed

app/feed_editor/rewrite/routes.py

@@ -10,7 +10,7 @@
 )
 from feed_editor.rewrite.rewriter import FeedRewriter
 from feed_editor.rewrite.rules import validate_dict, validate_xpaths
-from feed_editor.rewrite.rules.types import FeedRulesDict
+from feed_editor.rewrite.rules.types import FeedTransformDict
 
 rewrite_api = Blueprint("rewrite", __name__, url_prefix="/rewrite")
 
@@ -64,7 +64,7 @@ def url():
     return compress_and_encode(feed_dict)
 
 
-def _parse_args(params: Mapping[str, str]) -> FeedRulesDict:
+def _parse_args(params: Mapping[str, str]) -> FeedTransformDict:
     feed_data_gzipped: str | None = params.get("r", None)
 
     if feed_data_gzipped:

app/feed_editor/rewrite/rules/__init__.py

@@ -13,24 +13,24 @@
 from .types import (
     AndDict,
     ConditionDict,
-    FeedRulesDict,
+    FeedTransformDict,
     MutationDict,
     OrDict,
     RuleDict,
 )
 
 
-def validate_dict(test_dict: Mapping) -> FeedRulesDict:
+def validate_dict(test_dict: Mapping) -> FeedTransformDict:
     """Validate that a dictionary represents a feed and rules to apply to it (FeedRulesDict)"""
-    return generic_validate_dict(FeedRulesDict, test_dict)
+    return generic_validate_dict(FeedTransformDict, test_dict)
 
 
-def validate_xpaths(feed_rules: FeedRulesDict) -> bool:
+def validate_xpaths(feed_transform: FeedTransformDict) -> bool:
     """Traverses the entire dictionary and ensures that all rules and their
     mutations and conditions have valid xpaths, primarily that they are non-empty.
 
     Args:
-        feed_rules (FeedRulesDict): Feed URL and transformation rules
+        feed_transform (FeedRulesDict): Feed URL and transformation rules
 
     Returns:
         bool: True if all xpaths are valid, False otherwise
@@ -63,7 +63,7 @@ def validate_rule_xpaths(rule: RuleDict) -> bool:
             )
         )
 
-    return all(validate_rule_xpaths(rule) for rule in feed_rules["rules"])
+    return all(validate_rule_xpaths(rule) for rule in feed_transform["rules"])
 
 
 def test_conditions_element(
@@ -95,8 +95,7 @@ def test_condition(condition: ConditionDict) -> bool:
             return test_disjunction(condition)
 
         if condition["name"] in conditions_map:
-            cond_dict = conditions_map[condition["name"]]
-            cond = cond_dict["definition"]
+            cond = conditions_map[condition["name"]]
 
             if test_element is not None and test_element.text is not None:
                 return cond(test_element.text, condition["args"])
@@ -137,7 +136,7 @@ def run_mutations_element(
         mut = mutation_map[mutation_dict["name"]]
         args = mutation_dict["args"]
 
-        mut["definition"](target_element, args)
+        mut(target_element, args)
 
 
 def run_rule(tree: etree._ElementTree, rule: RuleDict) -> None:

app/feed_editor/rewrite/rules/conditions.py

@@ -1,73 +1,50 @@
 # pylint: disable=too-few-public-methods,missing-class-docstring
-import functools
-from typing import TYPE_CHECKING, Callable, Protocol, TypedDict
-
-from feed_editor.utils.dict_validation import _TypedDict_T, validate_dict
+import re
+from typing import (
+    TYPE_CHECKING,
+    Generic,
+    Optional,
+    Protocol,
+    TypedDict,
+    TypeVar,
+)
 
 if TYPE_CHECKING:
     from .types import ConditionDict
 
 
-def _require_args(dict_type: type[_TypedDict_T]):
-    def decorator(
-        predicate: Callable[[str, _TypedDict_T], bool]
-    ) -> Callable[[str, "ConditionArgs"], bool]:
-        @functools.wraps(predicate)
-        def decorated(value, args: "ConditionArgs"):
-            return predicate(value, validate_dict(dict_type, args))
-
-        return decorated
-
-    return decorator
-
-
 class ContainsArgs(TypedDict):
-    """Arguments to the contains condition"""
-
-    value: str
-
-
-@_require_args(ContainsArgs)
-def _contains(feed_value: str, args: ContainsArgs, /) -> bool:
-    return args["value"] in feed_value
-
-
-def _contains_testval(xpath: str) -> "ConditionDict":
-    return {"xpath": xpath, "name": "contains", "args": {"value": "test value"}}
+    pattern: str
 
 
 ConditionArgs = ContainsArgs
+ConditionArgsT = TypeVar("ConditionArgsT", bound=ConditionArgs)
 
 
-class ConditionFn(Protocol):  # pylint: disable=too-few-public-methods
-    """Required signature for a condition"""
-
-    @staticmethod
-    def __call__(value: str, args: ConditionArgs, /) -> bool: ...
+class Condition(Protocol, Generic[ConditionArgsT]):
+    ArgSpec: type[ConditionArgsT]
+    name: str
 
+    def __call__(self, feed_value: str, args: ConditionArgsT) -> bool: ...
 
-class TestFactory(Protocol):
-    def __call__(self, xpath: str) -> "ConditionDict": ...
+    def __test_factory__(
+        self, xpath: str, args: Optional[ConditionArgsT] = None
+    ) -> "ConditionDict": ...
 
 
-class Condition(TypedDict):
-    """Base TypedDict for a Condition"""
+class Contains(Condition[ContainsArgs]):
+    ArgSpec = ContainsArgs
+    name = "contains"
 
-    display_name: str
-    definition: ConditionFn
-    arg_spec: type[ConditionArgs]
-    test_factory: TestFactory
+    def __call__(self, feed_value: str, args: ContainsArgs) -> bool:
+        return re.search(args["pattern"], feed_value) is not None
 
+    def __test_factory__(
+        self, xpath: str, args: ContainsArgs | None = None
+    ) -> "ConditionDict":
+        args = args or {"pattern": ".+?"}
+        return {"xpath": xpath, "name": self.name, "args": args}
 
-all_conditions: list[Condition] = [
-    {
-        "display_name": "contains",
-        "definition": _contains,
-        "arg_spec": ContainsArgs,
-        "test_factory": _contains_testval,
-    }
-]
 
-conditions_map: dict[str, Condition] = {
-    cond["display_name"]: cond for cond in all_conditions
-}
+all_conditions: list[Condition] = [condcls() for condcls in Condition.__subclasses__()]
+conditions_map: dict[str, Condition] = {cond.name: cond for cond in all_conditions}