# End-user documentation is in ../../doc/ and so is for the most part not
# duplicated here in the form of doc strings. Code comments and docstrings
# are mainly for internal use.
from __future__ import annotations
import enum
import operator
import re
from dataclasses import Field, dataclass, field, fields
from functools import reduce, wraps
from typing import (
Any,
Callable,
ClassVar,
Dict,
FrozenSet,
Generator,
Generic,
List,
Mapping,
Optional,
Pattern,
Tuple,
Type,
TypeVar,
Union,
overload,
)
from typing_extensions import Literal, ParamSpec, Protocol, TypeVarTuple, Unpack
from ts_lib_parsy.error import ParseError
OUT = TypeVar("OUT")
OUT1 = TypeVar("OUT1")
OUT2 = TypeVar("OUT2")
OUT3 = TypeVar("OUT3")
OUT4 = TypeVar("OUT4")
OUT5 = TypeVar("OUT5")
OUT6 = TypeVar("OUT6")
OUT_T = TypeVarTuple("OUT_T")
OUT_T2 = TypeVarTuple("OUT_T2")
OUT_co = TypeVar("OUT_co", covariant=True)
OUT2_co = TypeVar("OUT2_co", covariant=True)
P = ParamSpec("P")
T = TypeVar("T")
T_co = TypeVar("T_co", covariant=True)
_T_contra = TypeVar("_T_contra", contravariant=True)
_T_co = TypeVar("_T_co", covariant=True)
[docs]
class SupportsAdd(Protocol[_T_contra, _T_co]):
def __add__(self, __x: _T_contra) -> _T_co:
...
[docs]
def noop(val: T) -> T:
return val
[docs]
@dataclass
class Result(Generic[OUT_co]):
status: bool
index: int
value: OUT_co
furthest: int
expected: FrozenSet[str]
[docs]
@staticmethod
def success(index: int, value: OUT) -> Result[OUT]:
return Result(True, index, value, -1, frozenset())
# We don't handle types of failures yet, and always
# either:
# - don't return these values (e.g. choose another parser)
# - raise an exception.
# Therefore, I think it is safe here to use `Any` as type to keep type checker happy
# The same issue crops up in various branches that return parse failure results
[docs]
@staticmethod
def failure(index: int, expected: str) -> Result[Any]:
return Result(False, -1, None, index, frozenset([expected]))
# collect the furthest failure from self and other
[docs]
def aggregate(self: Result[OUT], other: Optional[Result[Any]]) -> Result[OUT]:
if not other:
return self
if self.furthest > other.furthest:
return self
elif self.furthest == other.furthest:
# if we both have the same failure index, we combine the expected messages.
return Result(
self.status,
self.index,
self.value,
self.furthest,
self.expected | other.expected,
)
else:
return Result(
self.status, self.index, self.value, other.furthest, other.expected
)
[docs]
class Parser(Generic[OUT_co]):
"""
A Parser is an object that wraps a function whose arguments are
a string to be parsed and the index on which to begin parsing.
The function should return either Result.success(next_index, value),
where the next index is where to continue the parse and the value is
the yielded value, or Result.failure(index, expected), where expected
is a string indicating what was expected, and the index is the index
of the failure.
"""
def __init__(self, wrapped_fn: Callable[[str, int], Result[OUT_co]]):
"""
This is a low level function to create new parsers that is used internally
but is rarely needed by users of the parsy library. It should be passed a
parsing function, which takes two arguments - a string/list to be parsed
and the current index into the list - and returns a :class:`Result` object.
"""
self.wrapped_fn: Callable[[str, int], Result[OUT_co]] = wrapped_fn
def __call__(self, stream: str, index: int) -> Result[OUT_co]:
return self.wrapped_fn(stream, index)
[docs]
def parse(self, stream: str) -> OUT_co:
"""
Attempts to parse the given string (or list). If the parse is successful
and consumes the entire string, the result is returned - otherwise, a
``ParseError`` is raised.
"""
"""Parse a string and return the result or raise a ParseError."""
(result, _) = (self << eof).parse_partial(stream)
return result
[docs]
def parse_partial(self, stream: str) -> Tuple[OUT_co, str]:
"""
Parse the longest possible prefix of a given string.
Return a tuple of the result and the rest of the string,
or raise a ParseError.
"""
result = self(stream, 0)
if result.status:
return (result.value, stream[result.index :])
else:
raise ParseError(result.expected, stream, result.furthest)
[docs]
def bind(
self: Parser[OUT1], bind_fn: Callable[[OUT1], Parser[OUT2]]
) -> Parser[OUT2]:
"""
Returns a parser which, if the initial parser is successful, passes the
result to ``bind_fn``, and continues with the parser returned from ``bind_fn``.
This is the monadic binding operation.
"""
@Parser
def bound_parser(stream: str, index: int) -> Result[OUT2]:
result: Result[OUT1] = self(stream, index)
if result.status:
next_parser = bind_fn(result.value)
return next_parser(stream, result.index).aggregate(result)
else:
return result # type: ignore
return bound_parser
[docs]
def map(self: Parser[OUT1], map_fn: Callable[[OUT1], OUT2]) -> Parser[OUT2]:
"""
Returns a parser that transforms the produced value of the initial parser
with ``map_fn``.
.. code:: python
>>> regex(r'[0-9]+').map(int).parse('1234')
1234
This is the simplest way to convert parsed strings into the data types
that you need.
"""
return self.bind(lambda res: success(map_fn(res)))
[docs]
def concat(self: Parser[List[str]]) -> Parser[str]:
"""
Returns a parser that concatenates together (as a string) the previously
produced values. Usually used after :meth:`~Parser.many` and similar
methods that produce multiple values.
.. code:: python
>>> letter.at_least(1).parse("hello")
['h', 'e', 'l', 'l', 'o']
>>> letter.at_least(1).concat().parse("hello")
'hello'
"""
return self.map("".join)
[docs]
def then(self: Parser[Any], other: Parser[OUT2]) -> Parser[OUT2]:
"""
Returns a parser which, if the initial parser succeeds, will continue parsing
with ``other_parser``. This will produce the value produced by
``other_parser``.
.. code:: python
>>> string('x').then(string('y')).parse('xy')
'y'
"""
return (self & other).map(lambda t: t[1])
[docs]
def skip(self: Parser[OUT1], other: Parser[Any]) -> Parser[OUT1]:
"""
Similar to :meth:`Parser.then`, except the resulting parser will use
the value produced by the first parser.
.. code:: python
>>> string('x').skip(string('y')).parse('xy')
'x'
"""
return (self & other).map(lambda t: t[0])
[docs]
def result(self: Parser[Any], res: OUT2) -> Parser[OUT2]:
"""
Returns a parser that, if the initial parser succeeds, always produces
``val``.
.. code:: python
>>> string('foo').result(42).parse('foo')
42
"""
return self >> success(res)
[docs]
def many(self: Parser[OUT_co]) -> Parser[List[OUT_co]]:
"""
Returns a parser that expects the initial parser 0 or more times, and
produces a list of the results. Note that this parser does not fail if
nothing matches, but instead consumes nothing and produces an empty list.
.. code:: python
>>> parser = regex(r'[a-z]').many()
>>> parser.parse('')
[]
>>> parser.parse('abc')
['a', 'b', 'c']
"""
return self.times(0, float("inf"))
[docs]
def times(
self: Parser[OUT_co], min: int, max: int | float | None = None
) -> Parser[List[OUT_co]]:
"""
Returns a parser that expects the initial parser at least ``min`` times,
and at most ``max`` times, and produces a list of the results. If only one
argument is given, the parser is expected exactly that number of times.
"""
the_max: int | float
if max is None:
the_max = min
else:
the_max = max
@Parser
def times_parser(stream: str, index: int) -> Result[List[OUT_co]]:
values: List[OUT_co] = []
times = 0
result = None
while times < the_max:
result = self(stream, index).aggregate(result)
if result.status:
values.append(result.value)
index = result.index
times += 1
elif times >= min:
break
else:
return result # type: ignore
return Result.success(index, values).aggregate(result)
return times_parser
[docs]
def at_most(self: Parser[OUT_co], n: int) -> Parser[List[OUT_co]]:
"""
Returns a parser that expects the initial parser at most ``n`` times, and
produces a list of the results.
"""
return self.times(0, n)
[docs]
def at_least(self: Parser[OUT_co], n: int) -> Parser[List[OUT_co]]:
"""
Returns a parser that expects the initial parser at least ``n`` times, and
produces a list of the results.
"""
return self.times(min=n, max=float("inf"))
[docs]
def optional(self: Parser[OUT1], default: OUT2 = None) -> Parser[OUT1 | OUT2]:
"""
Returns a parser that expects the initial parser zero or once, and maps
the result to a given default value in the case of no match. If no default
value is given, ``None`` is used.
.. code:: python
>>> string('A').optional().parse('A')
'A'
>>> string('A').optional().parse('')
None
>>> string('A').optional('Oops').parse('')
'Oops'
"""
return self.times(0, 1).map(lambda v: v[0] if v else default)
[docs]
def until(
self: Parser[OUT_co],
other: Parser[Any],
min: int = 0,
max: int | float = float("inf"),
) -> Parser[List[OUT_co]]:
"""
Returns a parser that expects the initial parser followed by ``other``.
The initial parser is expected at least ``min`` times and at most ``max`` times.
.. 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'
"""
@Parser
def until_parser(stream: str, index: int) -> Result[List[OUT_co]]:
values: List[OUT_co] = []
times = 0
while True:
# try parser first
res = other(stream, index)
if res.status and times >= min:
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
[docs]
def sep_by(
self: Parser[OUT],
sep: Parser[Any],
*,
min: int = 0,
max: int | float = float("inf"),
) -> Parser[List[OUT]]:
"""
Like :meth:`Parser.times`, this returns a new parser that repeats
the initial parser and collects the results in a list, but in this case
separated by the parser ``sep`` (whose return value is discarded). By default it
repeats with no limit, but minimum and maximum values can be supplied.
.. code:: python
>>> csv = letter.at_least(1).concat().sep_by(string(","))
>>> csv.parse("abc,def")
['abc', 'def']
"""
empty_result: List[OUT] = []
zero_times = success(empty_result)
if max == 0:
return zero_times
res = self.map(lambda x: [x]) + (sep >> self).times(min - 1, max - 1)
if min == 0:
res = res | zero_times
return res
[docs]
def desc(self, description: str) -> Parser[OUT_co]:
"""
Adds a description to the parser, which is used in the error message
if parsing fails.
>>> year = regex(r'[0-9]{4}').desc('4 digit year')
>>> year.parse('123')
ParseError: expected 4 digit year at 0:0
"""
@Parser
def desc_parser(stream: str, index: int) -> Result[OUT_co]:
result = self(stream, index)
if result.status:
return result
else:
return Result.failure(index, description)
return desc_parser
[docs]
def mark(
self: Parser[OUT_co],
) -> Parser[Tuple[Tuple[int, int], OUT_co, Tuple[int, int]]]:
"""
Returns a parser that wraps the initial parser's result in a value
containing column and line information of the match, as well as the
original value. The new value is a 3-tuple:
.. code:: python
((start_row, start_column),
original_value,
(end_row, end_column))
This is useful for being able to report problems with parsing more
accurately.
"""
return seq(line_info, self, line_info)
[docs]
def tag(self: Parser[OUT], name: str) -> Parser[Tuple[str, OUT]]:
"""
Returns a parser that wraps the produced value of the initial parser in a
2 tuple containing ``(name, value)``. This provides a very simple way to
label parsed components. e.g.:
.. code:: python
>>> day = regex(r'[0-9]+').map(int)
>>> month = string_from("January", "February", "March", "April", "May",
... "June", "July", "August", "September", "October",
... "November", "December")
>>> day.parse("10")
10
>>> day.tag("day").parse("10")
('day', 10)
>>> seq(day.tag("day") << whitespace,
... month.tag("month")
... ).parse("10 September")
[('day', 10), ('month', 'September')]
It also works well when combined with ``.map(dict)`` to get a dictionary
of values:
.. code:: python
>>> seq(day.tag("name") << whitespace,
... month.tag("month")
... ).map(dict).parse("10 September")
{'day': 10, 'month': 'September'}
"""
return self.map(lambda v: (name, v))
[docs]
def should_fail(self: Parser[OUT], description: str) -> Parser[Result[OUT]]:
"""
Returns a parser that fails when the initial parser succeeds, and succeeds
when the initial parser fails (consuming no input). A description must
be passed which is used in parse failure messages.
This is essentially a negative lookahead:
.. code:: python
>>> p = letter << string(" ").should_fail("not space")
>>> p.parse('A')
'A'
>>> p.parse('A ')
ParseError: expected 'not space' at 0:1
"""
@Parser
def fail_parser(stream: str, index: int) -> Result[Result[OUT]]:
res = self(stream, index)
if res.status:
return Result.failure(index, description)
return Result.success(index, res)
return fail_parser
# Special cases for adding tuples
# We have to unroll each number of tuple elements for `other` because PEP-646
# only allows one "Unpack" in a Tuple (if we could have two, the return
# type could use two Unpacks
@overload
def __add__(
self: Parser[Tuple[Unpack[OUT_T]]], other: Parser[Tuple[OUT1]]
) -> Parser[Tuple[Unpack[OUT_T], OUT1]]:
...
@overload
def __add__(
self: Parser[Tuple[Unpack[OUT_T]]], other: Parser[Tuple[OUT1, OUT2]]
) -> Parser[Tuple[Unpack[OUT_T], OUT1, OUT2]]:
...
@overload
def __add__(
self: Parser[Tuple[Unpack[OUT_T]]], other: Parser[Tuple[OUT1, OUT2, OUT3]]
) -> Parser[Tuple[Unpack[OUT_T], OUT1, OUT2, OUT3]]:
...
@overload
def __add__(
self: Parser[Tuple[Unpack[OUT_T]]], other: Parser[Tuple[OUT1, OUT2, OUT3, OUT4]]
) -> Parser[Tuple[Unpack[OUT_T], OUT1, OUT2, OUT3, OUT4]]:
...
@overload
def __add__(
self: Parser[Tuple[Unpack[OUT_T]]],
other: Parser[Tuple[OUT1, OUT2, OUT3, OUT4, OUT5]],
) -> Parser[Tuple[Unpack[OUT_T], OUT1, OUT2, OUT3, OUT4, OUT5]]:
...
# This covers tuples where `other` has more elements than the above overloads
# and the `self` and `other` tuples have the same homogeneous type
@overload
def __add__(
self: Parser[Tuple[OUT, ...]], other: Parser[Tuple[OUT, ...]]
) -> Parser[Tuple[OUT, ...]]:
...
# Cover the rest of cases which can't return a homogeneous tuple
@overload
def __add__(
self: Parser[Tuple[Any, ...]], other: Parser[Tuple[Any, ...]]
) -> Parser[Tuple[Any, ...]]:
...
# Addable parsers which return the same type
@overload
def __add__(
self: Parser[SupportsAdd[Any, _T_co]], other: Parser[SupportsAdd[Any, _T_co]]
) -> Parser[_T_co]:
...
def __add__(self: Parser[Any], other: Parser[Any]) -> Parser[Any]:
"""
Requires both parsers to match in order, and adds the two results together using
the + operator. This will only work if the results support the plus operator
(strings, lists, tuples, numeric types and so on):
.. code-block:: python
>>> (string("x") + regex("[0-9]")).parse("x1")
"x1"
>>> (string("x").many() + string("y").many()).parse("xxyyy")
['x', 'x', 'y', 'y', 'y']
"""
return (self & other).combine(operator.add)
def __mul__(self: Parser[OUT], other: range | int) -> Parser[List[OUT]]:
"""
This is a shortcut for doing :meth:`Parser.times`:
.. code-block:: python
>>> (string("x") * 3).parse("xxx")
["x", "x", "x"]
You can also set both upper and lower bounds by multiplying by a range:
.. code-block:: python
>>> (string("x") * range(0, 3)).parse("xxx")
ParseError: expected EOF at 0:2
(Note the normal semantics of ``range`` are respected - the second number is an
*exclusive* upper bound, not inclusive).
"""
if isinstance(other, range):
return self.times(other.start, other.stop - 1)
return self.times(other)
def __or__(self: Parser[OUT1], other: Parser[OUT2]) -> Parser[Union[OUT1, OUT2]]:
"""
Returns a parser that tries ``self`` and, if it fails, backtracks
and tries ``other``. These can be chained together.
``parser | other``
The resulting parser will produce the value produced by the first
successful parser.
.. code:: python
>>> parser = string('x') | string('y') | string('z')
>>> parser.parse('x')
'x'
>>> parser.parse('y')
'y'
>>> parser.parse('z')
'z'
Note that ``other`` will only be tried if ``parser`` cannot consume any
input and fails. ``other`` is not used in the case that **later** parser
components fail. This means that the order of the operands matters - for
example:
.. code:: python
>>> ((string('A') | string('AB')) + string('C')).parse('ABC')
ParseEror: expected 'C' at 0:1
>>> ((string('AB') | string('A')) + string('C')).parse('ABC')
'ABC'
>>> ((string('AB') | string('A')) + string('C')).parse('AC')
'AC'
"""
@Parser
def alt_parser(stream: str, index: int) -> Result[Union[OUT1, OUT2]]:
result0 = None
result1 = self(stream, index).aggregate(result0)
if result1.status:
return result1
result2 = other(stream, index).aggregate(result1)
return result2
return alt_parser
def __and__(self: Parser[OUT1], other: Parser[OUT2]) -> Parser[Tuple[OUT1, OUT2]]:
"""
Combine two parsers into a new parser whose result is a tuple of the results
of parsing with `self` then `other`. ``parser & other``.
.. code:: python
>>> parser = string('x') & string('y')
>>> parser.parse('xy')
('x', 'y')
"""
@Parser
def and_parser(stream: str, index: int) -> Result[Tuple[OUT1, OUT2]]:
self_result = self(stream, index)
if not self_result.status:
return self_result # type: ignore
other_result = other(stream, self_result.index).aggregate(self_result)
if not other_result.status:
return other_result # type: ignore
return Result.success(
other_result.index, (self_result.value, other_result.value)
).aggregate(other_result)
return and_parser
[docs]
def join(self: Parser[OUT1], other: Parser[OUT2]) -> Parser[Tuple[OUT1, OUT2]]:
"""
An alternative name for the ``&`` parser (``Parser.__and__``).
"""
return self & other
[docs]
def as_tuple(self: Parser[OUT]) -> Parser[Tuple[OUT]]:
"""
Modify the result of a parser, wrapping it in a tuple.
Several combinators work on parsers which return tuples, like ``append``
and ``combine``, so this parser can be used before calling other parsers
which expect a tuple result type.
.. code:: python
>>> parser = string('x').as_tuple()
>>> parser.parse('x')
('x',)
>>> parser.append(string('y')).parse('xy')
('x', 'y')
"""
return self.map(lambda value: (value,))
[docs]
def append(
self: Parser[Tuple[Unpack[OUT_T]]], other: Parser[OUT2]
) -> Parser[Tuple[Unpack[OUT_T], OUT2]]:
"""
Take a parser whose result is a tuple, and add the result of ``other``
to the end of that tuple.
.. code:: python
>>> parser = string('x').as_tuple()
>>> parser.parse('x')
('x',)
>>> parser.append(string('y')).parse('xy')
('x', 'y')
"""
return self.bind(
lambda self_value: other.bind(
lambda other_value: success((*self_value, other_value))
)
)
[docs]
def combine(
self: Parser[Tuple[Unpack[OUT_T]]], combine_fn: Callable[[Unpack[OUT_T]], OUT2]
) -> Parser[OUT2]:
"""
Returns a parser that transforms the produced values of the initial parser
with ``combine_fn``, passing the arguments using ``*args`` syntax.
Where the current parser produces an iterable of values, this can be a
more convenient way to combine them than :meth:`~Parser.map`.
Example 1 - the argument order of our callable already matches:
.. code:: python
>>> from datetime import date
>>> yyyymmdd = seq(regex(r'[0-9]{4}').map(int),
... regex(r'[0-9]{2}').map(int),
... regex(r'[0-9]{2}').map(int)).combine(date)
>>> yyyymmdd.parse('20140506')
datetime.date(2014, 5, 6)
"""
return self.bind(lambda value: success(combine_fn(*value)))
# haskelley operators, for fun #
# >>
def __rshift__(self, other: Parser[OUT]) -> Parser[OUT]:
"""
The same as ``parser.then(other_parser)`` - see :meth:`Parser.then`.
``self >> other``. (Hint - the arrows point at the important parser!)
.. code-block:: python
>>> (string('x') >> string('y')).parse('xy')
'y'
"""
return self.then(other)
# <<
def __lshift__(self, other: Parser[Any]) -> Parser[OUT_co]:
"""
The same as ``self.skip(other)`` - see :meth:`Parser.skip`. ``self << other``.
(Hint - the arrows point at the important parser!)
.. code:: python
>>> (string('x') << string('y')).parse('xy')
'x'
"""
return self.skip(other)
[docs]
def generate(fn: Callable[[], Generator[Parser[Any], Any, OUT]]) -> Parser[OUT]:
"""
``generate`` converts a generator function (one that uses the ``yield`` keyword)
into a parser. The generator function must yield parsers. These parsers are
applied successively and their results are sent back to the generator using the
``.send()`` protocol. The generator function should return the final result of
the parsing. Alternatively it can return another parser, which is equivalent to
applying it and returning its result.
"""
@Parser
@wraps(fn)
def generated(stream: str, index: int) -> Result[OUT]:
# start up the generator
iterator = fn()
result = None
value = None
try:
while True:
next_parser = iterator.send(value)
result = next_parser(stream, index).aggregate(result)
if not result.status:
return result
value = result.value
index = result.index
except StopIteration as stop:
returnVal = stop.value
return Result.success(index, returnVal).aggregate(result)
return generated
# A convenience type for defining forward references to parsers using a generator
ParserReference = Generator[Parser[T], T, T]
#: A parser that consumes no input and always just returns the current stream
#: index. This is normally useful when wanting to build more debugging
#: information into parse failure error messages.
index = Parser(lambda _, index: Result.success(index, index))
[docs]
def line_info_at(stream: str, index: int) -> Tuple[int, int]:
"""Given text and an index, return the line and column indices, zero-indexed."""
if index > len(stream):
raise ValueError("invalid index")
line = stream.count("\n", 0, index)
last_nl = stream.rfind("\n", 0, index)
col = index - (last_nl + 1)
return (line, col)
#: A parser that consumes no input and always just returns the current line
#: information, a tuple of (line, column), zero-indexed, where lines are
#: terminated by ``\n``. This is normally useful when wanting to build more
#: debugging information into parse failure error messages.
line_info = Parser(
lambda stream, index: Result.success(index, line_info_at(stream, index))
)
[docs]
def success(val: OUT) -> Parser[OUT]:
"""
Returns a parser that does not consume any of the stream, but
produces ``val``.
"""
return Parser(lambda _, index: Result.success(index, val))
[docs]
def fail(expected: str) -> Parser[None]:
"""
Returns a parser that always fails with the provided error message.
"""
return Parser(lambda _, index: Result.failure(index, expected))
[docs]
def string(s: str, transform: Callable[[str], str] = noop) -> Parser[str]:
"""
Returns a parser that expects the ``expected_string`` and produces
that string value.
Optionally, a transform function can be passed, which will be used on both
the expected string and tested string. This allows things like case
insensitive matches to be done. This function must not change the length of
the string (as determined by ``len``). The returned value of the parser will
always be ``expected_string`` in its un-transformed state.
.. code-block:: python
>>> parser = string("Hello", transform=lambda s: s.upper())
>>> parser.parse("Hello")
'Hello'
>>> parser.parse("hello")
'Hello'
>>> parser.parse("HELLO")
'Hello'
"""
slen = len(s)
transformed_s = transform(s)
@Parser
def string_parser(stream: str, index: int) -> Result[str]:
if transform(stream[index : index + slen]) == transformed_s:
return Result.success(index + slen, s)
else:
return Result.failure(index, s)
return string_parser
PatternType = Union[str, Pattern[str]]
@overload
def regex(
pattern: PatternType,
*,
flags: re.RegexFlag = re.RegexFlag(0),
group: Literal[0] = 0,
) -> Parser[str]:
...
@overload
def regex(
pattern: PatternType, *, flags: re.RegexFlag = re.RegexFlag(0), group: str | int
) -> Parser[str]:
...
@overload
def regex(
pattern: PatternType,
*,
flags: re.RegexFlag = re.RegexFlag(0),
group: Tuple[str | int],
) -> Parser[Tuple[str]]:
...
@overload
def regex(
pattern: PatternType,
*,
flags: re.RegexFlag = re.RegexFlag(0),
group: Tuple[str | int, str | int],
) -> Parser[Tuple[str, str]]:
...
@overload
def regex(
pattern: PatternType,
*,
flags: re.RegexFlag = re.RegexFlag(0),
group: Tuple[str | int, str | int, str | int],
) -> Parser[Tuple[str, str, str]]:
...
@overload
def regex(
pattern: PatternType,
*,
flags: re.RegexFlag = re.RegexFlag(0),
group: Tuple[str | int, str | int, str | int, str | int],
) -> Parser[Tuple[str, str, str, str]]:
...
@overload
def regex(
pattern: PatternType,
*,
flags: re.RegexFlag = re.RegexFlag(0),
group: Tuple[str | int, str | int, str | int, str | int, str | int],
) -> Parser[Tuple[str, str, str, str, str]]:
...
[docs]
def regex(
pattern: PatternType,
*,
flags: re.RegexFlag = re.RegexFlag(0),
group: str | int | Tuple[str | int, ...] = 0,
) -> Parser[str | Tuple[str, ...]]:
"""
Returns a parser that expects the given ``exp``, and produces the
matched string. ``exp`` can be a compiled regular expression, or a
string which will be compiled with the given ``flags``.
Optionally, accepts ``group``, which is passed to `re.Match.group
<https://docs.python.org/3/library/re.html#re.Match.group>`_ to
return the text from a capturing group in the regex instead of the
entire match.
Using a regex parser for small building blocks, instead of building up
parsers from primitives like :func:`string`, :func:`test_char` and
:meth:`Parser.times` combinators etc., can have several advantages,
including:
* It can be more succinct e.g. compare:
.. code-block:: python
>>> (string('a') | string('b')).times(1, 4)
>>> regex(r'[ab]{1,4}')
* It can return the entire matched string as a single item,
so you don't need to use :meth:`Parser.concat`.
* It can return a part of the matched string using a capturing group
from the regex, so you don't need to split the string yourself.
You can use named or numbered groups, just like with `re.Match.group
<https://docs.python.org/3/library/re.html#re.Match.group>`_.
Tuples also work, and return the captured text from multiple groups.
.. code-block:: python
>>> regex(r'([0-9]{4})-([0-9]{2})', group=1).parse('2020-03')
'2020'
>>> regex(r'(?P<year>[0-9]{4})-(?P<month>[0-9]{2})', group='month').parse('2020-03')
'03'
>>> regex(r'([0-9]{4})-([0-9]{2})', group=(1,2)).parse('2020-03')
('2020', '03')
* It can be much faster.
"""
if isinstance(pattern, str):
exp = re.compile(pattern, flags)
else:
exp = pattern
if isinstance(group, tuple) and len(group) >= 2:
first_group, second_group, *groups = group
@Parser
def regex_parser_tuple(stream: str, index: int) -> Result[Tuple[str, ...]]:
match = exp.match(stream, index)
if match:
return Result.success(
match.end(), match.group(first_group, second_group, *groups)
)
else:
return Result.failure(index, exp.pattern)
return regex_parser_tuple
if isinstance(group, tuple) and len(group) == 1:
target_group = group[0]
elif isinstance(group, tuple):
target_group = 0
else:
target_group = group
@Parser
def regex_parser(stream: str, index: int) -> Result[str]:
match = exp.match(stream, index)
if match:
return Result.success(match.end(), match.group(target_group))
else:
return Result.failure(index, exp.pattern)
return regex_parser
# Each number of args needs to be typed separately
@overload
def seq(
__parser_1: Parser[OUT1],
__parser_2: Parser[OUT2],
__parser_3: Parser[OUT3],
__parser_4: Parser[OUT4],
__parser_5: Parser[OUT5],
__parser_6: Parser[OUT6],
) -> Parser[Tuple[OUT1, OUT2, OUT3, OUT4, OUT5, OUT6]]:
...
@overload
def seq(
__parser_1: Parser[OUT1],
__parser_2: Parser[OUT2],
__parser_3: Parser[OUT3],
__parser_4: Parser[OUT4],
__parser_5: Parser[OUT5],
) -> Parser[Tuple[OUT1, OUT2, OUT3, OUT4, OUT5]]:
...
@overload
def seq(
__parser_1: Parser[OUT1],
__parser_2: Parser[OUT2],
__parser_3: Parser[OUT3],
__parser_4: Parser[OUT4],
) -> Parser[Tuple[OUT1, OUT2, OUT3, OUT4]]:
...
@overload
def seq(
__parser_1: Parser[OUT1], __parser_2: Parser[OUT2], __parser_3: Parser[OUT3]
) -> Parser[Tuple[OUT1, OUT2, OUT3]]:
...
@overload
def seq(
__parser_1: Parser[OUT1], __parser_2: Parser[OUT2]
) -> Parser[Tuple[OUT1, OUT2]]:
...
@overload
def seq(__parser_1: Parser[OUT1]) -> Parser[Tuple[OUT1]]:
...
@overload
def seq(*parsers: Parser[Any]) -> Parser[Tuple[Any, ...]]:
...
[docs]
def seq(*parsers: Parser[Any]) -> Parser[Tuple[Any, ...]]:
"""
Creates a parser that applies a sequence of parsers in sequence and combines
their results into a tuple.
.. code-block:: python
>>> x_bottles_of_y_on_the_z = \
... seq(regex(r"[0-9]+").map(int) << string(" bottles of "),
... regex(r"\S+") << string(" on the "),
... regex(r"\S+")
... )
>>> x_bottles_of_y_on_the_z.parse("99 bottles of beer on the wall")
(99, 'beer', 'wall')
"""
if not parsers:
raise ValueError("`seq` must receive at least one parser")
first, *remainder = parsers
parser = first.as_tuple()
for p in remainder:
parser = parser.append(p) # type: ignore
return parser
[docs]
def test_char(func: Callable[[str], bool], description: str) -> Parser[str]:
"""
Returns a parser that tests a single character with the callable
``func``. If ``func`` returns ``True``, the parse succeeds, otherwise
the parse fails with the description ``description``.
.. code-block:: python
>>> ascii = test_char(lambda c: ord(c) < 128,
... 'ascii character')
>>> ascii.parse('A')
'A'
"""
@Parser
def test_char_parser(stream: str, index: int) -> Result[str]:
if index < len(stream):
if func(stream[index]):
return Result.success(index + 1, stream[index])
return Result.failure(index, description)
return test_char_parser
[docs]
def match_char(char: str, description: Optional[str] = None) -> Parser[str]:
if description is None:
description = char
return test_char(lambda i: char == i, description)
[docs]
def string_from(*strings: str, transform: Callable[[str], str] = noop) -> Parser[str]:
"""
Accepts a sequence of strings as positional arguments, and returns a parser
that matches and returns one string from the list. The list is first sorted
in descending length order, so that overlapping strings are handled correctly
by checking the longest one first.
.. code-block:: python
>>> string_from('y', 'yes').parse('yes')
'yes'
Optionally accepts ``transform``, which is passed to :func:`string` (see the
documentation there).
"""
# Sort longest first, so that overlapping options work correctly
return reduce(
operator.or_,
[string(s, transform) for s in sorted(strings, key=len, reverse=True)],
)
[docs]
def char_from(string: str) -> Parser[str]:
"""
Accepts a string and returns a parser that matches and returns one character
from the string.
.. code-block:: python
>>> char_from('abc').parse('a')
'a'
"""
return test_char(lambda c: c in string, "[" + string + "]")
[docs]
def peek(parser: Parser[OUT]) -> Parser[OUT]:
"""
Returns a lookahead parser that parse the input stream without consuming
chars.
.. code-block: python
>>> peek(any_char).parse_partial("ABC")
('A', 'ABC')
"""
@Parser
def peek_parser(stream: str, index: int) -> Result[OUT]:
result = parser(stream, index)
if result.status:
return Result.success(index, result.value)
else:
return result
return peek_parser
#: A parser that matches any single character.
any_char = test_char(lambda c: True, "any character")
#: A parser that matches and returns one or more whitespace characters.
whitespace = regex(r"\s+")
#: A parser that matches and returns a single letter, as defined by
#: `str.isalpha <https://docs.python.org/3/library/stdtypes.html#str.isalpha>`_.
letter = test_char(lambda c: c.isalpha(), "a letter")
#: A parser that matches and returns a single digit, as defined by `str.isdigit
#: <https://docs.python.org/3/library/stdtypes.html#str.isdigit>`_. Note that
#: this includes various unicode characters outside of the normal 0-9 range,
#: such as ¹²³.
digit = test_char(lambda c: c.isdigit(), "a digit")
#: A parser that matches and returns a single decimal digit, one of
#: "0123456789".
decimal_digit = char_from("0123456789")
@Parser
def eof(stream: str, index: int) -> Result[None]:
"""
A parser that only succeeds if the end of the stream has been reached.
>>> eof.parse_partial("")
(None, '')
>>> eof.parse_partial("123")
Traceback (most recent call last):
...
parsy.ParseError: expected 'EOF' at 0:0
"""
if index >= len(stream):
return Result.success(index, None)
else:
return Result.failure(index, "EOF")
E = TypeVar("E", bound=enum.Enum)
[docs]
def from_enum(enum_cls: type[E], transform: Callable[[str], str] = noop) -> Parser[E]:
"""
Given a class that is an `enum.Enum
<https://docs.python.org/3/library/enum.html>`_ class, returns a parser that
will parse the values (or the string representations of the values) and
return the corresponding enum item.
.. code-block:: python
>>> from enum import Enum
>>> class Pet(Enum):
... CAT = "cat"
... DOG = "dog"
>>> pet = from_enum(Pet)
>>> pet.parse("cat")
<Pet.CAT: 'cat'>
``str`` is first run on the values (for the case of values that are integers
etc.) to create the strings which are turned into parsers using
:func:`string`.
If ``transform`` is provided, it is passed to :func:`string` when creating
the parser (allowing for things like case insensitive parsing).
"""
items = sorted(
((str(enum_item.value), enum_item) for enum_item in enum_cls),
key=lambda t: len(t[0]),
reverse=True,
)
return reduce(
operator.or_,
[
string(value, transform=transform).result(enum_item)
for value, enum_item in items
],
)
# Dataclass parsers
[docs]
def parser_field(
parser: Parser[OUT],
*,
init: bool = True,
repr: bool = True,
hash: Union[bool, None] = None,
compare: bool = True,
metadata: Optional[Mapping[Any, Any]] = None,
) -> OUT:
"""
A field descriptor for dataclass fields which adds a parser to the metadata of the
field. This parser can later be used by ``dataclass_parser`` to convert the
dataclass into a parser whose result is the dataclass.
"""
if metadata is None:
metadata = {}
return field(
init=init,
repr=repr,
hash=hash,
compare=compare,
metadata={**metadata, "parser": parser},
)
[docs]
class DataClassProtocol(Protocol):
"""Protocol used to annotate a generic dataclass type."""
__dataclass_fields__: ClassVar[Dict[str, Field[Any]]]
__init__: Callable[..., None]
OUT_D = TypeVar("OUT_D", bound=DataClassProtocol)
[docs]
def dataclass_parser(datatype: Type[OUT_D]) -> Parser[OUT_D]:
"""
Take a dataclass with field descriptors defined using ``parser_field``, and
call each field's ``parser`` sequentially. Take the results of those parsers
and use them to create an instance of the `datatype` dataclass, resulting
in a parser with a type of ``Parser[datatype]``.
"""
field_parsers: Dict[str, Parser[Any]] = {
dataclass_field.name: dataclass_field.metadata["parser"]
for dataclass_field in fields(datatype)
if "parser" in dataclass_field.metadata
}
@Parser
def data_parser(stream: str, index: int) -> Result[OUT_D]:
parsed_fields: Dict[str, Any] = {}
for field_name, parser in field_parsers.items():
result = parser(stream, index)
if not result.status:
# A parser did not match - return the unsuccessful result
return result # type: ignore
index = result.index
parsed_fields[field_name] = result.value
return Result.success(index, datatype(**parsed_fields))
return data_parser