(
self, method: str, endpoint: str, version: str = "v3", **kwargs
)
| 42 | self.users = UsersSDK(self) |
| 43 | |
| 44 | def _make_request( |
| 45 | self, method: str, endpoint: str, version: str = "v3", **kwargs |
| 46 | ) -> dict[str, Any] | BytesIO | None: |
| 47 | url = self._get_full_url(endpoint, version) |
| 48 | request_args = self._prepare_request_args(endpoint, **kwargs) |
| 49 | |
| 50 | try: |
| 51 | response = self.client.request(method, url, **request_args) |
| 52 | self._handle_response(response) |
| 53 | |
| 54 | if "application/json" in response.headers.get("Content-Type", ""): |
| 55 | return response.json() if response.content else None |
| 56 | else: |
| 57 | return BytesIO(response.content) |
| 58 | |
| 59 | except ConnectError as e: |
| 60 | raise R2RClientException( |
| 61 | message="Unable to connect to the server. Check your network connection and the server URL." |
| 62 | ) from e |
| 63 | |
| 64 | except RequestError as e: |
| 65 | raise R2RException( |
| 66 | message=f"Request failed: {str(e)}", |
| 67 | status_code=500, |
| 68 | ) from e |
| 69 | |
| 70 | def _make_streaming_request( |
| 71 | self, method: str, endpoint: str, version: str = "v3", **kwargs |
no test coverage detected