Create an event for the scan. Raises a validation error if the event could not be created, unless raise_error is set to False. Args: *args: Positional arguments to be passed to the scan's make_event method. **kwargs: Keyword arguments to be passed to the sca
(self, *args, **kwargs)
| 498 | return submitted |
| 499 | |
| 500 | def make_event(self, *args, **kwargs): |
| 501 | """Create an event for the scan. |
| 502 | |
| 503 | Raises a validation error if the event could not be created, unless raise_error is set to False. |
| 504 | |
| 505 | Args: |
| 506 | *args: Positional arguments to be passed to the scan's make_event method. |
| 507 | **kwargs: Keyword arguments to be passed to the scan's make_event method. |
| 508 | raise_error (bool, optional): Whether to raise a validation error if the event could not be created. Defaults to False. |
| 509 | |
| 510 | Examples: |
| 511 | >>> new_event = self.make_event("1.2.3.4", parent=event) |
| 512 | >>> await self.emit_event(new_event) |
| 513 | |
| 514 | Returns: |
| 515 | Event or None: The created event, or None if a validation error occurred and raise_error was False. |
| 516 | |
| 517 | Raises: |
| 518 | ValidationError: If the event could not be validated and raise_error is True. |
| 519 | """ |
| 520 | raise_error = kwargs.pop("raise_error", False) |
| 521 | module = kwargs.pop("module", None) |
| 522 | if module is None: |
| 523 | if (not args) or getattr(args[0], "module", None) is None: |
| 524 | kwargs["module"] = self |
| 525 | try: |
| 526 | if args and is_event(args[0]): |
| 527 | raise ValidationError( |
| 528 | f"{self.__class__.__name__}.make_event() does not accept an existing event " |
| 529 | f"({type(args[0]).__name__}) as the first argument. " |
| 530 | "Use update_event(event, ...) or emit_event(event, ...) instead." |
| 531 | ) |
| 532 | event = self.scan.make_event(*args, **kwargs) |
| 533 | except ValidationError as e: |
| 534 | if raise_error: |
| 535 | raise |
| 536 | self.warning(f"{e}") |
| 537 | return |
| 538 | return event |
| 539 | |
| 540 | def update_event(self, event, **kwargs): |
| 541 | """Update an existing event for the scan. |