Method upload_object

agentops/client/api/versions/v4.py:71–106

Upload an object to the V4 API. Args: body: The object body to upload Returns: Dictionary containing upload response data Raises: ApiServerException: If the upload fails

(self, body: Union[str, bytes])

        return HttpClient.get_session().post(url, json={"body": body}, headers=request_headers, timeout=30)

    def upload_object(self, body: Union[str, bytes]) -> Dict[str, Any]:
        """
        Upload an object to the V4 API.

        Args:
            body: The object body to upload

        Returns:
            Dictionary containing upload response data

        Raises:
            ApiServerException: If the upload fails
        """
        try:
            # Convert bytes to string for consistency with test expectations
            if isinstance(body, bytes):
                body = body.decode("utf-8")

            response = self.post("/v4/objects/upload/", body, self.prepare_headers())

            if response.status_code != 200:
                error_msg = f"Upload failed: {response.status_code}"
                try:
                    error_data = response.json()
                    if "error" in error_data:
                        error_msg = error_data["error"]
                except:
                    pass
                raise ApiServerException(error_msg)

            try:
                return response.json()
            except Exception as e:
                raise ApiServerException(f"Failed to process upload response: {str(e)}")
        except requests.exceptions.RequestException as e:
            raise ApiServerException(f"Failed to upload object: {e}")

    def upload_logfile(self, body: Union[str, bytes], trace_id: str) -> Dict[str, Any]:
        """