Export multiple events. Args: events: Events to export Returns: Tuple of (successful, failed) counts
(self, events: list[SIEMEvent])
| 133 | |
| 134 | async def export_batch(self, events: list[SIEMEvent]) -> tuple[int, int]: |
| 135 | """Export multiple events. |
| 136 | |
| 137 | Args: |
| 138 | events: Events to export |
| 139 | |
| 140 | Returns: |
| 141 | Tuple of (successful, failed) counts |
| 142 | """ |
| 143 | success = 0 |
| 144 | failed = 0 |
| 145 | |
| 146 | for event in events: |
| 147 | try: |
| 148 | if await self.export(event): |
| 149 | success += 1 |
| 150 | else: |
| 151 | failed += 1 |
| 152 | except Exception as e: |
| 153 | self.logger.error(f"Export failed: {e}") |
| 154 | failed += 1 |
| 155 | |
| 156 | return success, failed |
| 157 | |
| 158 | async def export_scan_result(self, result: ScanResult) -> tuple[int, int]: |
| 159 | """Export a scan result as events. |
| 160 |
no test coverage detected