Return the current aggregate content as a payload without clearing. Returns None when there is no content buffered.
(self)
| 66 | self.last_updated = time.monotonic() |
| 67 | |
| 68 | def snapshot_payload(self) -> Optional[BaseResponseDataPayload]: |
| 69 | """Return the current aggregate content as a payload without clearing. |
| 70 | |
| 71 | Returns None when there is no content buffered. |
| 72 | """ |
| 73 | if not self.parts: |
| 74 | return None |
| 75 | content = "".join(self.parts) |
| 76 | return BaseResponseDataPayload(content=content) |
| 77 | |
| 78 | |
| 79 | class ResponseBuffer: |