Added method 'until'

Refs #14
Refs #31

docs/ref/methods_and_combinators.rst

@@ -111,6 +111,24 @@ can be used and manipulated as below.
  Returns a parser that expects the initial parser at least ``n`` times, and
  produces a list of the results.
 
+ .. method:: until(other_parser, [min=0, max=inf, consume_other=False])
+
+ Returns a parser that expects the initial parser followed by ``other_parser``.
+ The initial parser is expected at least ``min`` times and at most ``max`` times.
+ By default, it does not consume ``other_parser`` and it produces a list of the
+ results excluding ``other_parser``. If ``consume_other`` is ``True`` then
+ ``other_parser`` is consumed and its result is included in the list of results.
+
+ .. code:: python
+
+ >>> seq(string('A').until(string('B')), string('BC')).parse('AAABC')
+ [['A','A','A'], 'BC']
+ >>> string('A').until(string('B')).then(string('BC')).parse('AAABC')
+ 'BC'
+ >>> string('A').until(string('BC'), consume_other=True).parse('AAABC')
+ ['A', 'A', 'A', 'BC']
+
+
  .. method:: optional()
 
  Returns a parser that expects the initial parser zero or once, and maps

src/parsy/__init__.py

@@ -181,6 +181,43 @@ def at_least(self, n):
  def optional(self):
  return self.times(0, 1).map(lambda v: v[0] if v else None)
 
+ def until(self, other, min=0, max=float("inf"), consume_other=False):
+ @Parser
+ def until_parser(stream, index):
+ values = []
+ times = 0
+ while True:
+
+ # try parser first
+ res = other(stream, index)
+ if res.status and times >= min:
+ if consume_other:
+ # consume other
+ values.append(res.value)
+ index = res.index
+ return Result.success(index, values)
+
+ # exceeded max?
+ if times >= max:
+ # return failure, it matched parser more than max times
+ return Result.failure(index, f"at most {max} items")
+
+ # failed, try parser
+ result = self(stream, index)
+ if result.status:
+ # consume
+ values.append(result.value)
+ index = result.index
+ times += 1
+ elif times >= min:
+ # return failure, parser is not followed by other
+ return Result.failure(index, "did not find other parser")
+ else:
+ # return failure, it did not match parser at least min times
+ return Result.failure(index, f"at least {min} items; got {times} item(s)")
+
+ return until_parser
+
  def sep_by(self, sep, *, min=0, max=float("inf")):
  zero_times = success([])
  if max == 0:

tests/test_parsy.py

@@ -347,6 +347,73 @@ def test_at_most(self):
  self.assertEqual(ab.at_most(2).parse("abab"), ["ab", "ab"])
  self.assertRaises(ParseError, ab.at_most(2).parse, "ababab")
 
+ def test_until(self):
+
+ until = string("s").until(string("x"))
+
+ s = "ssssx"
+ self.assertEqual(until.parse_partial(s), (4 * ["s"], "x"))
+ self.assertEqual(seq(until, string("x")).parse(s), [4 * ["s"], "x"])
+ self.assertEqual(until.then(string("x")).parse(s), "x")
+
+ s = "ssssxy"
+ self.assertEqual(until.parse_partial(s), (4 * ["s"], "xy"))
+ self.assertEqual(seq(until, string("x")).parse_partial(s), ([4 * ["s"], "x"], "y"))
+ self.assertEqual(until.then(string("x")).parse_partial(s), ("x", "y"))
+
+ self.assertRaises(ParseError, until.parse, "ssssy")
+ self.assertRaises(ParseError, until.parse, "xssssxy")
+
+ self.assertEqual(until.parse_partial("xxx"), ([], "xxx"))
+
+ until = regex(".").until(string("x"))
+ self.assertEqual(until.parse_partial("xxxx"), ([], "xxxx"))
+
+ def test_until_with_consume_other(self):
+
+ until = string("s").until(string("x"), consume_other=True)
+
+ self.assertEqual(until.parse("ssssx"), 4 * ["s"] + ["x"])
+ self.assertEqual(until.parse_partial("ssssxy"), (4 * ["s"] + ["x"], "y"))
+
+ self.assertEqual(until.parse_partial("xxx"), (["x"], "xx"))
+
+ self.assertRaises(ParseError, until.parse, "ssssy")
+ self.assertRaises(ParseError, until.parse, "xssssxy")
+
+ def test_until_with_min(self):
+
+ until = string("s").until(string("x"), min=3)
+
+ self.assertEqual(until.parse_partial("sssx"), (3 * ["s"], "x"))
+ self.assertEqual(until.parse_partial("sssssx"), (5 * ["s"], "x"))
+
+ self.assertRaises(ParseError, until.parse_partial, "ssx")
+
+ def test_until_with_max(self):
+
+ # until with max
+ until = string("s").until(string("x"), max=3)
+
+ self.assertEqual(until.parse_partial("ssx"), (2 * ["s"], "x"))
+ self.assertEqual(until.parse_partial("sssx"), (3 * ["s"], "x"))
+
+ self.assertRaises(ParseError, until.parse_partial, "ssssx")
+
+ def test_until_with_min_max(self):
+
+ until = string("s").until(string("x"), min=3, max=5)
+
+ self.assertEqual(until.parse_partial("sssx"), (3 * ["s"], "x"))
+ self.assertEqual(until.parse_partial("sssssx"), (5 * ["s"], "x"))
+
+ with self.assertRaises(ParseError) as cm:
+ until.parse_partial("ssx")
+ assert cm.exception.args[0] == frozenset({"at least 3 items; got 2 item(s)"})
+ with self.assertRaises(ParseError) as cm:
+ until.parse_partial("ssssssx")
+ assert cm.exception.args[0] == frozenset({"at most 5 items"})
+
  def test_optional(self):
  p = string("a").optional()
  self.assertEqual(p.parse("a"), "a")