Check document against regex rules asynchronously.
(self, document)
| 31 | return patterns |
| 32 | |
| 33 | async def check_document(self, document): |
| 34 | """Check document against regex rules asynchronously.""" |
| 35 | loop = asyncio.get_running_loop() |
| 36 | tasks = [ |
| 37 | loop.run_in_executor(self.executor, self.match_pattern, rule, document) |
| 38 | for rule in self.patterns |
| 39 | ] |
| 40 | results = await asyncio.gather(*tasks) |
| 41 | return [match for match in results if match] |
| 42 | |
| 43 | def match_pattern(self, rule, document): |
| 44 | """Match a single regex pattern synchronously (runs in a thread pool).""" |
no outgoing calls
no test coverage detected