aboutsummaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--edify/builder/mixins/matcher.py70
-rw-r--r--edify/errors/backend.py19
-rw-r--r--edify/result/regex.py86
-rw-r--r--tests/builder/timeout.test.py109
4 files changed, 255 insertions, 29 deletions
diff --git a/edify/builder/mixins/matcher.py b/edify/builder/mixins/matcher.py
index 3ed74fa..6bf22ae 100644
--- a/edify/builder/mixins/matcher.py
+++ b/edify/builder/mixins/matcher.py
@@ -8,7 +8,8 @@ regex exactly once.
The signatures mirror :class:`re.Pattern` exactly so IDE autocomplete and
type inference stay unchanged. :meth:`test` is the one non-:mod:`re` method:
a boolean shortcut that returns ``True`` when the pattern matches anywhere
-in the input, ``False`` otherwise.
+in the input, ``False`` otherwise. Every match method also accepts a per-call
+``timeout=`` kwarg that only applies under ``engine="regex"``.
"""
from __future__ import annotations
@@ -27,50 +28,89 @@ class MatcherMixin(BuilderProtocol):
"""Return ``True`` when the pattern matches anywhere in ``string``, else ``False``."""
return self._lazy_regex().search(string, pos, endpos) is not None
- def match(self, string: str, pos: int = 0, endpos: int = sys.maxsize) -> re.Match[str] | None:
+ def match(
+ self,
+ string: str,
+ pos: int = 0,
+ endpos: int = sys.maxsize,
+ *,
+ timeout: float | None = None,
+ ) -> re.Match[str] | None:
"""Delegate to :meth:`re.Pattern.match`."""
- return self._lazy_regex().match(string, pos, endpos)
+ return self._lazy_regex().match(string, pos, endpos, timeout=timeout)
- def search(self, string: str, pos: int = 0, endpos: int = sys.maxsize) -> re.Match[str] | None:
+ def search(
+ self,
+ string: str,
+ pos: int = 0,
+ endpos: int = sys.maxsize,
+ *,
+ timeout: float | None = None,
+ ) -> re.Match[str] | None:
"""Delegate to :meth:`re.Pattern.search`."""
- return self._lazy_regex().search(string, pos, endpos)
+ return self._lazy_regex().search(string, pos, endpos, timeout=timeout)
def fullmatch(
- self, string: str, pos: int = 0, endpos: int = sys.maxsize
+ self,
+ string: str,
+ pos: int = 0,
+ endpos: int = sys.maxsize,
+ *,
+ timeout: float | None = None,
) -> re.Match[str] | None:
"""Delegate to :meth:`re.Pattern.fullmatch`."""
- return self._lazy_regex().fullmatch(string, pos, endpos)
+ return self._lazy_regex().fullmatch(string, pos, endpos, timeout=timeout)
def findall(
- self, string: str, pos: int = 0, endpos: int = sys.maxsize
+ self,
+ string: str,
+ pos: int = 0,
+ endpos: int = sys.maxsize,
+ *,
+ timeout: float | None = None,
) -> list[str] | list[tuple[str, ...]]:
"""Delegate to :meth:`re.Pattern.findall`."""
- return self._lazy_regex().findall(string, pos, endpos)
+ return self._lazy_regex().findall(string, pos, endpos, timeout=timeout)
def finditer(
- self, string: str, pos: int = 0, endpos: int = sys.maxsize
+ self,
+ string: str,
+ pos: int = 0,
+ endpos: int = sys.maxsize,
+ *,
+ timeout: float | None = None,
) -> Iterator[re.Match[str]]:
"""Delegate to :meth:`re.Pattern.finditer`."""
- return self._lazy_regex().finditer(string, pos, endpos)
+ return self._lazy_regex().finditer(string, pos, endpos, timeout=timeout)
def sub(
self,
replacement: str | Callable[[re.Match[str]], str],
string: str,
count: int = 0,
+ *,
+ timeout: float | None = None,
) -> str:
"""Delegate to :meth:`re.Pattern.sub`."""
- return self._lazy_regex().sub(replacement, string, count=count)
+ return self._lazy_regex().sub(replacement, string, count=count, timeout=timeout)
def subn(
self,
replacement: str | Callable[[re.Match[str]], str],
string: str,
count: int = 0,
+ *,
+ timeout: float | None = None,
) -> tuple[str, int]:
"""Delegate to :meth:`re.Pattern.subn`."""
- return self._lazy_regex().subn(replacement, string, count=count)
+ return self._lazy_regex().subn(replacement, string, count=count, timeout=timeout)
- def split(self, string: str, maxsplit: int = 0) -> list[str | None]:
+ def split(
+ self,
+ string: str,
+ maxsplit: int = 0,
+ *,
+ timeout: float | None = None,
+ ) -> list[str | None]:
"""Delegate to :meth:`re.Pattern.split`."""
- return self._lazy_regex().split(string, maxsplit=maxsplit)
+ return self._lazy_regex().split(string, maxsplit=maxsplit, timeout=timeout)
diff --git a/edify/errors/backend.py b/edify/errors/backend.py
index cd7650e..3102c55 100644
--- a/edify/errors/backend.py
+++ b/edify/errors/backend.py
@@ -21,6 +21,25 @@ class MissingRegexBackendError(EdifySyntaxError):
super().__init__(message)
+class TimeoutNotSupportedByEngineError(EdifySyntaxError):
+ """Raised when ``timeout=`` is passed to a match call on an ``engine='re'`` pattern."""
+
+ def __init__(self) -> None:
+ message = compose_annotated_message(
+ summary="the timeout= kwarg is only supported under engine='regex'",
+ trigger_hint="match/search/... called with timeout= here",
+ note=(
+ "the stdlib re module has no per-call timeout facility; only the "
+ "third-party regex engine exposes one."
+ ),
+ help_line=(
+ "help: re-compile the pattern with .to_regex(engine='regex') to use timeout=, "
+ "or drop the kwarg."
+ ),
+ )
+ super().__init__(message)
+
+
class VariableWidthLookbehindNotSupportedError(EdifySyntaxError):
"""Raised when ``engine='re'`` compiles a lookbehind whose body is variable-width."""
diff --git a/edify/result/regex.py b/edify/result/regex.py
index 2627729..75cf3e3 100644
--- a/edify/result/regex.py
+++ b/edify/result/regex.py
@@ -9,6 +9,7 @@ from typing import Any, cast
from edify.builder.types.engine import Engine
from edify.elements.types.base import BaseElement
+from edify.errors.backend import TimeoutNotSupportedByEngineError
from edify.introspect.explain import explain_elements
from edify.introspect.verbose import verbose_elements
from edify.introspect.visualize import visualize_elements
@@ -62,53 +63,102 @@ class Regex:
"""The engine identifier this pattern was compiled with."""
return self._engine
- def match(self, string: str, pos: int = 0, endpos: int = sys.maxsize) -> re.Match[str] | None:
+ def match(
+ self,
+ string: str,
+ pos: int = 0,
+ endpos: int = sys.maxsize,
+ *,
+ timeout: float | None = None,
+ ) -> re.Match[str] | None:
"""Delegate to the compiled pattern's ``match`` method."""
- return self._compiled.match(string, pos, endpos)
+ return self._compiled.match(string, pos, endpos, **_timeout_kwargs(self._engine, timeout))
- def search(self, string: str, pos: int = 0, endpos: int = sys.maxsize) -> re.Match[str] | None:
+ def search(
+ self,
+ string: str,
+ pos: int = 0,
+ endpos: int = sys.maxsize,
+ *,
+ timeout: float | None = None,
+ ) -> re.Match[str] | None:
"""Delegate to the compiled pattern's ``search`` method."""
- return self._compiled.search(string, pos, endpos)
+ return self._compiled.search(string, pos, endpos, **_timeout_kwargs(self._engine, timeout))
def fullmatch(
- self, string: str, pos: int = 0, endpos: int = sys.maxsize
+ self,
+ string: str,
+ pos: int = 0,
+ endpos: int = sys.maxsize,
+ *,
+ timeout: float | None = None,
) -> re.Match[str] | None:
"""Delegate to the compiled pattern's ``fullmatch`` method."""
- return self._compiled.fullmatch(string, pos, endpos)
+ return self._compiled.fullmatch(
+ string, pos, endpos, **_timeout_kwargs(self._engine, timeout)
+ )
def findall(
- self, string: str, pos: int = 0, endpos: int = sys.maxsize
+ self,
+ string: str,
+ pos: int = 0,
+ endpos: int = sys.maxsize,
+ *,
+ timeout: float | None = None,
) -> list[str] | list[tuple[str, ...]]:
"""Delegate to the compiled pattern's ``findall`` method."""
- return self._compiled.findall(string, pos, endpos)
+ return self._compiled.findall(string, pos, endpos, **_timeout_kwargs(self._engine, timeout))
def finditer(
- self, string: str, pos: int = 0, endpos: int = sys.maxsize
+ self,
+ string: str,
+ pos: int = 0,
+ endpos: int = sys.maxsize,
+ *,
+ timeout: float | None = None,
) -> Iterator[re.Match[str]]:
"""Delegate to the compiled pattern's ``finditer`` method."""
- return self._compiled.finditer(string, pos, endpos)
+ return self._compiled.finditer(
+ string, pos, endpos, **_timeout_kwargs(self._engine, timeout)
+ )
def sub(
self,
replacement: str | Callable[[re.Match[str]], str],
string: str,
count: int = 0,
+ *,
+ timeout: float | None = None,
) -> str:
"""Delegate to the compiled pattern's ``sub`` method."""
- return self._compiled.sub(replacement, string, count=count)
+ return self._compiled.sub(
+ replacement, string, count=count, **_timeout_kwargs(self._engine, timeout)
+ )
def subn(
self,
replacement: str | Callable[[re.Match[str]], str],
string: str,
count: int = 0,
+ *,
+ timeout: float | None = None,
) -> tuple[str, int]:
"""Delegate to the compiled pattern's ``subn`` method."""
- return self._compiled.subn(replacement, string, count=count)
+ return self._compiled.subn(
+ replacement, string, count=count, **_timeout_kwargs(self._engine, timeout)
+ )
- def split(self, string: str, maxsplit: int = 0) -> list[str | None]:
+ def split(
+ self,
+ string: str,
+ maxsplit: int = 0,
+ *,
+ timeout: float | None = None,
+ ) -> list[str | None]:
"""Delegate to the compiled pattern's ``split`` method."""
- return self._compiled.split(string, maxsplit=maxsplit)
+ return self._compiled.split(
+ string, maxsplit=maxsplit, **_timeout_kwargs(self._engine, timeout)
+ )
def explain(self) -> str:
"""Return a human-readable narrative describing what the pattern matches."""
@@ -164,3 +214,11 @@ class Regex:
def __hash__(self) -> int:
"""Return a hash derived from the source, engine, and compiled flags."""
return hash((self._source, self._engine, self._compiled.flags))
+
+
+def _timeout_kwargs(engine: Engine, timeout: float | None) -> Mapping[str, float]:
+ if timeout is None:
+ return {}
+ if engine == "re":
+ raise TimeoutNotSupportedByEngineError()
+ return {"timeout": timeout}
diff --git a/tests/builder/timeout.test.py b/tests/builder/timeout.test.py
new file mode 100644
index 0000000..466ffb3
--- /dev/null
+++ b/tests/builder/timeout.test.py
@@ -0,0 +1,109 @@
+"""Tests for the per-call ``timeout=`` kwarg on match methods."""
+
+import pytest
+
+from edify import RegexBuilder
+from edify.errors.backend import TimeoutNotSupportedByEngineError
+
+
+def _regex_pattern():
+ return RegexBuilder().string("a").to_regex(engine="regex")
+
+
+def _re_pattern():
+ return RegexBuilder().string("a").to_regex(engine="re")
+
+
+def test_timeout_flows_through_search_under_regex_engine():
+ assert _regex_pattern().search("aaa", timeout=1.0) is not None
+
+
+def test_timeout_flows_through_match_under_regex_engine():
+ assert _regex_pattern().match("aaa", timeout=1.0) is not None
+
+
+def test_timeout_flows_through_fullmatch_under_regex_engine():
+ assert _regex_pattern().fullmatch("a", timeout=1.0) is not None
+
+
+def test_timeout_flows_through_findall_under_regex_engine():
+ assert _regex_pattern().findall("aaa", timeout=1.0) == ["a", "a", "a"]
+
+
+def test_timeout_flows_through_finditer_under_regex_engine():
+ matches = list(_regex_pattern().finditer("aaa", timeout=1.0))
+ assert len(matches) == 3
+
+
+def test_timeout_flows_through_sub_under_regex_engine():
+ assert _regex_pattern().sub("b", "aaa", timeout=1.0) == "bbb"
+
+
+def test_timeout_flows_through_subn_under_regex_engine():
+ assert _regex_pattern().subn("b", "aaa", timeout=1.0) == ("bbb", 3)
+
+
+def test_timeout_flows_through_split_under_regex_engine():
+ assert _regex_pattern().split("a1a2a3", timeout=1.0) == ["", "1", "2", "3"]
+
+
+def test_timeout_under_re_engine_raises_annotated_error():
+ with pytest.raises(TimeoutNotSupportedByEngineError) as excinfo:
+ _re_pattern().search("aaa", timeout=1.0)
+ text = str(excinfo.value)
+ assert "timeout=" in text
+ assert "engine='regex'" in text
+ assert "= note:" in text
+
+
+def test_matcher_search_forwards_timeout_to_the_cached_regex():
+ builder = RegexBuilder().string("a")
+ with pytest.raises(TimeoutNotSupportedByEngineError):
+ builder.search("aaa", timeout=1.0)
+
+
+def test_matcher_match_forwards_timeout_to_the_cached_regex():
+ builder = RegexBuilder().string("a")
+ with pytest.raises(TimeoutNotSupportedByEngineError):
+ builder.match("aaa", timeout=1.0)
+
+
+def test_matcher_fullmatch_forwards_timeout_to_the_cached_regex():
+ builder = RegexBuilder().string("a")
+ with pytest.raises(TimeoutNotSupportedByEngineError):
+ builder.fullmatch("a", timeout=1.0)
+
+
+def test_matcher_findall_forwards_timeout_to_the_cached_regex():
+ builder = RegexBuilder().string("a")
+ with pytest.raises(TimeoutNotSupportedByEngineError):
+ builder.findall("aaa", timeout=1.0)
+
+
+def test_matcher_finditer_forwards_timeout_to_the_cached_regex():
+ builder = RegexBuilder().string("a")
+ with pytest.raises(TimeoutNotSupportedByEngineError):
+ list(builder.finditer("aaa", timeout=1.0))
+
+
+def test_matcher_sub_forwards_timeout_to_the_cached_regex():
+ builder = RegexBuilder().string("a")
+ with pytest.raises(TimeoutNotSupportedByEngineError):
+ builder.sub("b", "aaa", timeout=1.0)
+
+
+def test_matcher_subn_forwards_timeout_to_the_cached_regex():
+ builder = RegexBuilder().string("a")
+ with pytest.raises(TimeoutNotSupportedByEngineError):
+ builder.subn("b", "aaa", timeout=1.0)
+
+
+def test_matcher_split_forwards_timeout_to_the_cached_regex():
+ builder = RegexBuilder().string("a")
+ with pytest.raises(TimeoutNotSupportedByEngineError):
+ builder.split("aaa", timeout=1.0)
+
+
+def test_no_timeout_kwarg_never_raises_on_either_engine():
+ assert _regex_pattern().search("aaa") is not None
+ assert _re_pattern().search("aaa") is not None
generated by cgit v1.2.3 (git 2.25.1) at 2026-07-27 08:51:57 +0000