POST a batch of JSONL records to the collection endpoint. Returns True if the server accepted the batch. This is the ONLY function that makes network calls.
(lines: list[str])
| 270 | |
| 271 | |
| 272 | def _send_batch(lines: list[str]) -> bool: |
| 273 | """POST a batch of JSONL records to the collection endpoint. |
| 274 | |
| 275 | Returns True if the server accepted the batch. This is the ONLY |
| 276 | function that makes network calls. |
| 277 | """ |
| 278 | try: |
| 279 | import httpx |
| 280 | payload = "\n".join(lines) |
| 281 | resp = httpx.post( |
| 282 | _ENDPOINT_URL, |
| 283 | content=payload.encode("utf-8"), |
| 284 | headers={"Content-Type": "application/jsonl"}, |
| 285 | timeout=10.0, |
| 286 | ) |
| 287 | return resp.status_code in (200, 201, 202, 204) |
| 288 | except Exception as e: |
| 289 | logger.debug("Telemetry send failed: %s", e) |
| 290 | return False |
| 291 | |
| 292 | |
| 293 | def get_sent_records() -> list[dict[str, Any]]: |