Until

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=1, 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,47 @@ 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,69 @@ 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'))
+
+        self.assertRaises(ParseError, until.parse_partial, 'ssx')
+        self.assertRaises(ParseError, until.parse_partial, 'ssssssx')
+
     def test_optional(self):
         p = string("a").optional()
         self.assertEqual(p.parse("a"), "a")