ringcentral.http.api_response

  1# !/usr/bin/env python
  2# encoding: utf-8
  3import json
  4import requests
  5from email.feedparser import FeedParser
  6from .json_object import *
  7from ..core import is_third
  8
  9def get_prepared_request_details(request):
 10    if request is None:
 11        return "The request is None."
 12    
 13    try:
 14        method = request.method or "Unknown Method"
 15        url = request.url or "Unknown URL"
 16        headers = dict(request.headers) if request.headers else {}
 17        details = [
 18            f"Method: {method}",
 19            f"URL: {url}",
 20            f"Headers: {headers}"
 21        ]
 22        return "\n".join(details)
 23    except Exception as e:
 24        return f"Error generating request details: {str(e)}"
 25
 26
 27class ApiResponse:
 28    def __init__(self, request=None, response=None):
 29        self._request = request
 30        self._response = response
 31
 32    def ok(self):
 33        return self._response.ok
 34
 35    def raw(self):
 36        return self._response.raw
 37
 38    def body(self):
 39        return self._response.content
 40
 41    def text(self):
 42        return self._response.text
 43
 44    def json_dict(self):
 45        if not self._is_content_type('application/json'):
 46            raise Exception('Response is not JSON')
 47
 48        return self._response.json()
 49
 50    def json(self):
 51        return unfold(self._response.json())
 52
 53    def multipart(self):
 54        if not self._is_content_type('multipart/mixed'):
 55            raise Exception('Response is not Batch (Multipart)')
 56
 57        parts = self._break_into_parts()
 58
 59        if len(parts) < 1:
 60            raise Exception("Malformed Batch Response (not enough parts)")  # sic! not specific extension
 61
 62        statuses = json.loads(parts[0].get_payload())
 63
 64        if len(statuses["response"]) != len(parts) - 1:
 65            raise Exception("Malformed Batch Response (not-consistent number of parts)")
 66
 67        responses = []
 68
 69        for response, payload in zip(statuses["response"], parts[1:]):
 70            res = create_response(payload, response['status'])
 71
 72            responses.append(ApiResponse(response=res))
 73
 74        return responses
 75
 76    def error(self):
 77
 78        if self._response is None or self.ok():
 79            return None
 80
 81        message = 'HTTP ' + str(self._response.status_code) + ' ' + self._response.text
 82
 83        if self._request is not None:
 84            message = message + ' (request details: ' + get_prepared_request_details(self._request) + ')'
 85
 86        return message
 87
 88    def request(self):
 89        return self._request
 90
 91    def response(self):
 92        return self._response
 93
 94    def _is_content_type(self, content_type):
 95        return self._get_content_type().lower().find(content_type) >= 0
 96
 97    def _get_content_type(self):
 98        return self._response.headers.get('Content-Type') or ''
 99
100    def _break_into_parts(self):
101        p = FeedParser()
102
103        for h in self._response.headers:
104            p.feed(h + ':' + self._response.headers[h] + "\r\n")
105
106        p.feed("\r\n")
107        p.feed(self.text())
108
109        msg = p.close()
110
111        parts = msg.get_payload()
112
113        return parts
114
115
116def create_response(payload, status):
117    res = requests.Response()
118    res.headers = dict(payload)
119    if is_third():
120        res._content = bytes(payload.get_payload(), 'utf8')
121        res.encoding = 'utf8'
122    else:
123        res._content = str(payload.get_payload())
124    res.status_code = status
125    return res