Skip to content

Commit 04bfca2

Browse files
committed
feat(rules) add support of proper yaml arrays in rules
1 parent 04bb0c3 commit 04bfca2

2 files changed

Lines changed: 301 additions & 303 deletions

File tree

pacta/rules/parser.py

Lines changed: 24 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -79,23 +79,11 @@ def parse_file(self, path: str | Path) -> RulesDocumentAst:
7979
def parse_text(self, text: str, *, filename: str | None = None) -> RulesDocumentAst:
8080
filename = filename or "<rules>"
8181

82-
blocks = self._split_rule_blocks(text)
83-
rules: list[RuleAst] = []
84-
85-
for i, block in enumerate(blocks, start=1):
86-
try:
87-
rules.append(self._parse_rule_block(block, filename=filename))
88-
except RulesParseError:
89-
raise
90-
except Exception as e:
91-
raise RulesParseError(
92-
message=f"Failed to parse rule block #{i}: {e}",
93-
file=filename,
94-
) from e
95-
96-
return RulesDocumentAst(rules=tuple(rules), span=SourceSpan(file=filename))
97-
98-
def _parse_rule_block(self, block: str, *, filename: str) -> RuleAst:
82+
parsed_rules = self._parse_rule_block(text,filename=filename)
83+
84+
return RulesDocumentAst(rules=tuple(parsed_rules), span=SourceSpan(file=filename))
85+
86+
def _parse_rule_block(self, block: str, *, filename: str) -> list[RuleAst]:
9987
try:
10088
import yaml
10189
except ImportError as e:
@@ -109,13 +97,26 @@ def _parse_rule_block(self, block: str, *, filename: str) -> RuleAst:
10997
except Exception as e:
11098
raise RulesParseError(message=f"Invalid YAML in rule block: {e}", file=filename) from e
11199

112-
if not isinstance(doc, Mapping) or doc.get("rule") is None:
113-
raise RulesParseError(message="Rule block must contain top-level 'rule:' mapping", file=filename)
100+
if not isinstance(doc, Mapping) or doc.get("rules") is None:
101+
raise RulesParseError(message="Rule block must contain top-level 'rules:' list", file=filename)
102+
103+
rules_list = doc["rules"]
104+
if not isinstance(rules_list, list):
105+
raise RulesParseError(message="'rules list' must be a list", file=filename)
106+
107+
parsed_rules = []
114108

115-
root = doc["rule"]
116-
if not isinstance(root, Mapping):
117-
raise RulesParseError(message="'rule' must be a mapping", file=filename)
109+
for root in rules_list:
110+
if not isinstance(root, Mapping):
111+
raise RulesParseError(message="'rule' must be a mapping", file=filename)
112+
rule_item = self._add_rule_in_list(root, filename = filename)
113+
parsed_rules.append(rule_item)
118114

115+
return parsed_rules
116+
117+
118+
119+
def _add_rule_in_list(self, root: Mapping, *, filename:str) -> RuleAst:
119120
# Required fields
120121
rid = self._req_str(root, "id", filename)
121122
name = self._req_str(root, "name", filename)
@@ -152,10 +153,6 @@ def _parse_rule_block(self, block: str, *, filename: str) -> RuleAst:
152153
metadata={"parser": "dsl-v0+pyyaml"},
153154
)
154155

155-
# Keep your existing _split_rule_blocks, _parse_when_predicate, _parse_pred_item,
156-
# _parse_inline_compare, _parse_literal, _req_str, _opt_str as-is (they already handle
157-
# strings like "from.layer == domain" and nested any/all/not).
158-
159156
def _split_rule_blocks(self, text: str) -> list[str]:
160157
lines = [ln.rstrip("\n") for ln in text.splitlines()]
161158
blocks: list[list[str]] = []
@@ -177,7 +174,7 @@ def flush():
177174
filtered = []
178175
for b in blocks:
179176
first = next((ln for ln in b if ln.strip() and not ln.strip().startswith("#")), "")
180-
if first.strip() != "rule:":
177+
if first.strip() != "rules:":
181178
continue
182179
filtered.append("\n".join(b))
183180
return filtered

0 commit comments

Comments
 (0)