Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""
2This module provides the :module:`hyperscan` backend for :class:`~pathspec.pathspec.PathSpec`.
4WARNING: The *pathspec._backends.hyperscan* package is not part of the public
5API. Its contents and structure are likely to change.
6"""
7from __future__ import annotations
9from collections.abc import (
10 Sequence)
11from typing import (
12 Any,
13 Callable, # Replaced by `collections.abc.Callable` in 3.9.2.
14 Optional) # Replaced by `X | None` in 3.10.
16try:
17 import hyperscan
18except ModuleNotFoundError:
19 hyperscan = None # type: ignore[assignment]
21from pathspec.backend import (
22 _Backend)
23from pathspec.pattern import (
24 RegexPattern)
25from pathspec._typing import (
26 override) # Added in 3.12.
28from .._utils import (
29 enumerate_patterns,
30 get_regex_flags,
31 has_unsupported_flags,
32 translate_end_anchor,
33 translate_regex_flags)
35from .base import (
36 hyperscan_error)
37from ._base import (
38 HS_FLAGS,
39 HS_VERSION,
40 HS_VERSION_0_9_1,
41 HyperscanExprDat,
42 HyperscanExprDebug,
43 compat_convert_utf8)
46class HyperscanPsBackend(_Backend):
47 """
48 The :class:`HyperscanPsBackend` class is the :module:`hyperscan`
49 implementation used by :class:`~pathspec.pathspec.PathSpec` for matching
50 files. The Hyperscan database uses block mode for matching files.
51 """
53 def __init__(
54 self,
55 patterns: Sequence[RegexPattern],
56 *,
57 _debug_exprs: Optional[bool] = None,
58 _test_sort: Optional[Callable[[list], None]] = None,
59 ) -> None:
60 """
61 Initialize the :class:`HyperscanPsBackend` instance.
63 *patterns* (:class:`Sequence` of :class:`.RegexPattern`) contains the
64 compiled patterns.
65 """
66 if hyperscan is None:
67 assert hyperscan_error is not None, (hyperscan, hyperscan_error)
68 raise hyperscan_error
70 if patterns and not isinstance(patterns[0], RegexPattern):
71 raise TypeError(f"{patterns[0]=!r} must be a RegexPattern.")
73 if has_unsupported_flags(patterns):
74 raise ValueError((
75 "The hyperscan backend cannot preserve the regex compile flags. "
76 "Use backend='simple'."
77 ))
79 use_patterns = enumerate_patterns(
80 patterns, filter=True, reverse=False,
81 )
83 debug_exprs = bool(_debug_exprs)
84 if use_patterns:
85 db = self._make_db()
86 expr_data = self._init_db(
87 db=db,
88 debug=debug_exprs,
89 patterns=use_patterns,
90 sort_ids=_test_sort,
91 )
92 else:
93 # WARNING: The hyperscan database cannot be initialized with zero
94 # patterns.
95 db = None
96 expr_data = []
98 self._db: Optional[hyperscan.Database] = db # type: ignore
99 """
100 *_db* (:class:`hyperscan.Database`) is the Hyperscan database.
101 """
103 self._debug_exprs = debug_exprs
104 """
105 *_debug_exprs* (:class:`bool`) is whether to include additional debugging
106 information for the expressions.
107 """
109 self._expr_data: list[HyperscanExprDat] = expr_data
110 """
111 *_expr_data* (:class:`list`) maps expression index (:class:`int`) to
112 expression data (:class:`:class:`HyperscanExprDat`).
113 """
115 self._out: tuple[Optional[bool], int] = (None, -1)
116 """
117 *_out* (:class:`tuple`) stores the current match:
119 - *0* (:class:`bool` or :data:`None`) is the match include.
121 - *1* (:class:`int`) is the match index.
122 """
124 self._patterns: dict[int, RegexPattern] = dict(use_patterns)
125 """
126 *_patterns* (:class:`dict`) maps pattern index (:class:`int`) to pattern
127 (:class:`RegexPattern`).
128 """
130 @staticmethod
131 def _init_db(
132 db: hyperscan.Database, # type: ignore
133 debug: bool,
134 patterns: list[tuple[int, RegexPattern]],
135 sort_ids: Optional[Callable[[list[int]], None]],
136 ) -> list[HyperscanExprDat]:
137 """
138 Initialize the Hyperscan database from the given patterns.
140 *db* (:class:`hyperscan.Hyperscan`) is the Hyperscan database.
142 *debug* (:class:`bool`) is whether to include additional debugging
143 information for the expressions.
145 *patterns* (:class:`~collections.abc.Sequence` of :class:`.RegexPattern`)
146 contains the patterns.
148 *sort_ids* (:class:`callable` or :data:`None`) is a function used to sort
149 the compiled expression ids. This is used during testing to ensure the order
150 of expressions is not accidentally relied on.
152 Returns a :class:`list` indexed by expression id (:class:`int`) to its data
153 (:class:`HyperscanExprDat`).
154 """
155 # WARNING: Hyperscan raises a `hyperscan.error` exception when compiled with
156 # zero elements.
157 assert patterns, patterns
159 # Prepare patterns.
160 expr_data: list[HyperscanExprDat] = []
161 exprs: list[bytes] = []
162 for pattern_index, pattern in patterns:
163 pattern_include = pattern.include
164 assert pattern_include is not None, (pattern_index, pattern)
165 assert pattern.regex is not None, (pattern_index, pattern)
167 # Encode regex.
168 assert isinstance(pattern, RegexPattern), pattern
169 regex = translate_end_anchor(pattern.regex.pattern)
170 regex = translate_regex_flags(
171 regex,
172 get_regex_flags(pattern.regex),
173 )
175 if isinstance(regex, bytes):
176 regex_bytes = regex
177 else:
178 assert isinstance(regex, str), regex
179 if HS_VERSION >= HS_VERSION_0_9_1:
180 regex_bytes = regex.encode('utf8')
181 else:
182 # Convert unicode for older version of hyperscan.
183 regex_bytes = compat_convert_utf8(regex)
185 if debug:
186 expr_data.append(HyperscanExprDebug(
187 include=pattern_include,
188 index=pattern_index,
189 is_dir_pattern=False,
190 regex=regex,
191 ))
192 else:
193 expr_data.append(HyperscanExprDat(
194 include=pattern_include,
195 index=pattern_index,
196 is_dir_pattern=False,
197 ))
199 exprs.append(regex_bytes)
201 # Sort expressions.
202 ids = list(range(len(exprs)))
203 if sort_ids is not None:
204 sort_ids(ids)
205 exprs = [exprs[__id] for __id in ids]
207 # Compile patterns.
208 db.compile(
209 expressions=exprs,
210 ids=ids,
211 elements=len(exprs),
212 flags=HS_FLAGS,
213 )
215 return expr_data
217 @override
218 def match_file(self, file: str) -> tuple[Optional[bool], Optional[int]]:
219 """
220 Check the file against the patterns.
222 *file* (:class:`str`) is the normalized file path to check.
224 Returns a :class:`tuple` containing whether to include *file* (:class:`bool`
225 or :data:`None`), and the index of the last matched pattern (:class:`int` or
226 :data:`None`).
227 """
228 # NOTICE: According to benchmarking, a method callback is 20% faster than
229 # using a closure here.
230 db = self._db
231 if db is None:
232 # Database was not initialized because there were no patterns. Return no
233 # match.
234 return (None, None)
236 self._out = (None, -1)
237 db.scan(file.encode('utf8'), match_event_handler=self.__on_match)
239 out_index: Optional[int]
240 out_include, out_index = self._out
241 if out_index == -1:
242 out_index = None
244 return (out_include, out_index)
246 @staticmethod
247 def _make_db() -> hyperscan.Database: # type: ignore
248 """
249 Create the Hyperscan database.
251 Returns the database (:class:`hyperscan.Database`).
252 """
253 assert hyperscan is not None, (hyperscan, hyperscan_error)
254 return hyperscan.Database(mode=hyperscan.HS_MODE_BLOCK)
256 def __on_match(
257 self,
258 expr_id: int,
259 _from: int,
260 _to: int,
261 _flags: int,
262 _context: Any,
263 ) -> Optional[bool]:
264 """
265 Called on each match.
267 *expr_id* (:class:`int`) is the expression id (index) of the matched
268 pattern.
269 """
270 # Store match.
271 # - WARNING: Hyperscan does not guarantee matches will be produced in order!
272 # Later expressions have higher priority.
273 expr_dat = self._expr_data[expr_id]
274 index = expr_dat.index
275 prev_index = self._out[1]
276 if index > prev_index:
277 self._out = (expr_dat.include, index)
279 return None