Class for misc CPU-intensive regex operations Offloads regex processing to other CPU cores via GIL release + thread pool For quick, one-off regexes, you don't need to use this helper. Only use this helper if you're searching large bodies of text or if your regex is CPU-intensi
| 4 | |
| 5 | |
| 6 | class RegexHelper: |
| 7 | """ |
| 8 | Class for misc CPU-intensive regex operations |
| 9 | |
| 10 | Offloads regex processing to other CPU cores via GIL release + thread pool |
| 11 | |
| 12 | For quick, one-off regexes, you don't need to use this helper. |
| 13 | Only use this helper if you're searching large bodies of text |
| 14 | or if your regex is CPU-intensive |
| 15 | """ |
| 16 | |
| 17 | def __init__(self, parent_helper): |
| 18 | self.parent_helper = parent_helper |
| 19 | |
| 20 | def ensure_compiled_regex(self, r): |
| 21 | """ |
| 22 | Make sure a regex has been compiled |
| 23 | """ |
| 24 | if not isinstance(r, re.Pattern): |
| 25 | raise ValueError("Regex must be compiled first!") |
| 26 | |
| 27 | def compile(self, *args, **kwargs): |
| 28 | return re.compile(*args, **kwargs) |
| 29 | |
| 30 | async def search(self, compiled_regex, *args, **kwargs): |
| 31 | self.ensure_compiled_regex(compiled_regex) |
| 32 | return await self.parent_helper.run_in_executor(compiled_regex.search, *args, **kwargs) |
| 33 | |
| 34 | async def match(self, compiled_regex, *args, **kwargs): |
| 35 | self.ensure_compiled_regex(compiled_regex) |
| 36 | return await self.parent_helper.run_in_executor(compiled_regex.match, *args, **kwargs) |
| 37 | |
| 38 | async def sub(self, compiled_regex, *args, **kwargs): |
| 39 | self.ensure_compiled_regex(compiled_regex) |
| 40 | return await self.parent_helper.run_in_executor(compiled_regex.sub, *args, **kwargs) |
| 41 | |
| 42 | async def findall(self, compiled_regex, *args, **kwargs): |
| 43 | self.ensure_compiled_regex(compiled_regex) |
| 44 | return await self.parent_helper.run_in_executor(compiled_regex.findall, *args, **kwargs) |
| 45 | |
| 46 | async def findall_multi(self, compiled_regexes, *args, threads=10, **kwargs): |
| 47 | """ |
| 48 | Same as findall() but with multiple regexes |
| 49 | """ |
| 50 | if not isinstance(compiled_regexes, dict): |
| 51 | raise ValueError('compiled_regexes must be a dictionary like this: {"regex_name": <compiled_regex>}') |
| 52 | for v in compiled_regexes.values(): |
| 53 | self.ensure_compiled_regex(v) |
| 54 | |
| 55 | tasks = {} |
| 56 | |
| 57 | def new_task(regex_name, r): |
| 58 | task = self.parent_helper.run_in_executor(r.findall, *args, **kwargs) |
| 59 | tasks[task] = regex_name |
| 60 | |
| 61 | compiled_regexes = dict(compiled_regexes) |
| 62 | for _ in range(threads): # Start initial batch of tasks |
| 63 | if compiled_regexes: # Ensure there are args to process |
no outgoing calls
no test coverage detected
searching dependent graphs…