Here we override RadixTarget's get() method, which normally only accepts hosts, to also accept events for convenience.
(self, event, **kwargs)
| 38 | return set(e.input for e in self.event_seeds) |
| 39 | |
| 40 | def get(self, event, **kwargs): |
| 41 | """ |
| 42 | Here we override RadixTarget's get() method, which normally only accepts hosts, to also accept events for convenience. |
| 43 | """ |
| 44 | host = None |
| 45 | raise_error = kwargs.get("raise_error", False) |
| 46 | # if it's already an event or event seed, use its host |
| 47 | if is_event(event) or isinstance(event, BaseEventSeed): |
| 48 | host = event.host |
| 49 | # save resources by checking if the event is an IP or DNS name |
| 50 | elif is_ip(event, include_network=True) or is_dns_name(event): |
| 51 | host = event |
| 52 | # if it's a string, autodetect its type and parse out its host |
| 53 | elif isinstance(event, str): |
| 54 | event_seed = self._make_event_seed(event, raise_error=raise_error) |
| 55 | host = event_seed.host |
| 56 | if not host: |
| 57 | return |
| 58 | else: |
| 59 | raise ValueError(f"Invalid target type for {self.__class__.__name__}: {type(event)}") |
| 60 | if not host: |
| 61 | msg = f"Host not found: '{event}'" |
| 62 | if raise_error: |
| 63 | raise KeyError(msg) |
| 64 | else: |
| 65 | log.warning(msg) |
| 66 | return |
| 67 | results = super().get(host, **kwargs) |
| 68 | return results |
| 69 | |
| 70 | def _make_event_seed(self, target, raise_error=False): |
| 71 | try: |
nothing calls this directly
no test coverage detected