| 23 | self.telemetry = self._validate_telemetry(telemetry) if telemetry else None |
| 24 | |
| 25 | def add_telemetry(self, telemetry: Optional[Union[TelemetryCreate, dict]] = None): |
| 26 | telemetry_payload = ( |
| 27 | self._validate_telemetry(telemetry) if telemetry else self.telemetry |
| 28 | ) |
| 29 | if telemetry_payload is None: |
| 30 | logger.error("TelemetryClient.add_telemetry() needs a telemetry payload") |
| 31 | return None |
| 32 | payload = telemetry_payload.model_dump(mode="json", exclude_none=True) |
| 33 | |
| 34 | try: |
| 35 | response = requests.post( |
| 36 | url=self.telemetry_url, |
| 37 | json=payload, |
| 38 | timeout=2, |
| 39 | headers=self.headers, |
| 40 | ) |
| 41 | if response.status_code != 201: |
| 42 | self._log_error(payload, response) |
| 43 | return None |
| 44 | return response.json() |
| 45 | except Exception as e: |
| 46 | logger.error(e, exc_info=True) |
| 47 | return None |
| 48 | |
| 49 | @staticmethod |
| 50 | def _validate_telemetry(telemetry: Union[TelemetryCreate, dict]) -> TelemetryCreate: |