Skip to content

rest

😸 Cheshire-Cat API

Production ready AI assistant framework

The version of the OpenAPI document: 1.3.1 Generated by OpenAPI Generator (https://openapi-generator.tech)

Do not edit the class manually.

RESTClientObject

Source code in cheshire_cat_api/rest.py 
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
class RESTClientObject:

    def __init__(self, configuration) -> None:
        # urllib3.PoolManager will pass all kw parameters to connectionpool
        # https://github.com/shazow/urllib3/blob/f9409436f83aeb79fbaf090181cd81b784f1b8ce/urllib3/poolmanager.py#L75  # noqa: E501
        # https://github.com/shazow/urllib3/blob/f9409436f83aeb79fbaf090181cd81b784f1b8ce/urllib3/connectionpool.py#L680  # noqa: E501
        # Custom SSL certificates and client certificates: http://urllib3.readthedocs.io/en/latest/advanced-usage.html  # noqa: E501

        # cert_reqs
        if configuration.verify_ssl:
            cert_reqs = ssl.CERT_REQUIRED
        else:
            cert_reqs = ssl.CERT_NONE

        addition_pool_args = {}
        if configuration.assert_hostname is not None:
            addition_pool_args['assert_hostname'] = (
                configuration.assert_hostname
            )

        if configuration.retries is not None:
            addition_pool_args['retries'] = configuration.retries

        if configuration.tls_server_name:
            addition_pool_args['server_hostname'] = configuration.tls_server_name


        if configuration.socket_options is not None:
            addition_pool_args['socket_options'] = configuration.socket_options

        # https pool manager
        if configuration.proxy:
            self.pool_manager = urllib3.ProxyManager(
                cert_reqs=cert_reqs,
                ca_certs=configuration.ssl_ca_cert,
                cert_file=configuration.cert_file,
                key_file=configuration.key_file,
                proxy_url=configuration.proxy,
                proxy_headers=configuration.proxy_headers,
                **addition_pool_args
            )
        else:
            self.pool_manager = urllib3.PoolManager(
                cert_reqs=cert_reqs,
                ca_certs=configuration.ssl_ca_cert,
                cert_file=configuration.cert_file,
                key_file=configuration.key_file,
                **addition_pool_args
            )

    def request(
        self,
        method,
        url,
        headers=None,
        body=None,
        post_params=None,
        _request_timeout=None
    ):
        """Perform requests.

        :param method: http request method
        :param url: http request url
        :param headers: http request headers
        :param body: request json body, for `application/json`
        :param post_params: request post parameters,
                            `application/x-www-form-urlencoded`
                            and `multipart/form-data`
        :param _request_timeout: timeout setting for this request. If one
                                 number provided, it will be total request
                                 timeout. It can also be a pair (tuple) of
                                 (connection, read) timeouts.
        """
        method = method.upper()
        assert method in [
            'GET',
            'HEAD',
            'DELETE',
            'POST',
            'PUT',
            'PATCH',
            'OPTIONS'
        ]

        if post_params and body:
            raise ApiValueError(
                "body parameter cannot be used with post_params parameter."
            )

        post_params = post_params or {}
        headers = headers or {}

        timeout = None
        if _request_timeout:
            if isinstance(_request_timeout, (int, float)):
                timeout = urllib3.Timeout(total=_request_timeout)
            elif (
                    isinstance(_request_timeout, tuple)
                    and len(_request_timeout) == 2
                ):
                timeout = urllib3.Timeout(
                    connect=_request_timeout[0],
                    read=_request_timeout[1]
                )

        try:
            # For `POST`, `PUT`, `PATCH`, `OPTIONS`, `DELETE`
            if method in ['POST', 'PUT', 'PATCH', 'OPTIONS', 'DELETE']:

                # no content type provided or payload is json
                content_type = headers.get('Content-Type')
                if (
                    not content_type
                    or re.search('json', content_type, re.IGNORECASE)
                ):
                    request_body = None
                    if body is not None:
                        request_body = json.dumps(body)
                    r = self.pool_manager.request(
                        method,
                        url,
                        body=request_body,
                        timeout=timeout,
                        headers=headers,
                        preload_content=False
                    )
                elif content_type == 'application/x-www-form-urlencoded':
                    r = self.pool_manager.request(
                        method,
                        url,
                        fields=post_params,
                        encode_multipart=False,
                        timeout=timeout,
                        headers=headers,
                        preload_content=False
                    )
                elif content_type == 'multipart/form-data':
                    # must del headers['Content-Type'], or the correct
                    # Content-Type which generated by urllib3 will be
                    # overwritten.
                    del headers['Content-Type']
                    r = self.pool_manager.request(
                        method,
                        url,
                        fields=post_params,
                        encode_multipart=True,
                        timeout=timeout,
                        headers=headers,
                        preload_content=False
                    )
                # Pass a `string` parameter directly in the body to support
                # other content types than Json when `body` argument is
                # provided in serialized form
                elif isinstance(body, str) or isinstance(body, bytes):
                    request_body = body
                    r = self.pool_manager.request(
                        method,
                        url,
                        body=request_body,
                        timeout=timeout,
                        headers=headers,
                        preload_content=False
                    )
                else:
                    # Cannot generate the request from given parameters
                    msg = """Cannot prepare a request message for provided
                             arguments. Please check that your arguments match
                             declared content type."""
                    raise ApiException(status=0, reason=msg)
            # For `GET`, `HEAD`
            else:
                r = self.pool_manager.request(
                    method,
                    url,
                    fields={},
                    timeout=timeout,
                    headers=headers,
                    preload_content=False
                )
        except urllib3.exceptions.SSLError as e:
            msg = "\n".join([type(e).__name__, str(e)])
            raise ApiException(status=0, reason=msg)

        return RESTResponse(r)

request(method, url, headers=None, body=None, post_params=None, _request_timeout=None)

Perform requests.

Parameters:

Name Type Description Default
method

http request method

 required
url

http request url

 required
headers

http request headers

 None
body

request json body, for application/json

 None
post_params

request post parameters, application/x-www-form-urlencoded and multipart/form-data

 None
_request_timeout

timeout setting for this request. If one number provided, it will be total request timeout. It can also be a pair (tuple) of (connection, read) timeouts.

 None
Source code in cheshire_cat_api/rest.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
def request(
    self,
    method,
    url,
    headers=None,
    body=None,
    post_params=None,
    _request_timeout=None
):
    """Perform requests.

    :param method: http request method
    :param url: http request url
    :param headers: http request headers
    :param body: request json body, for `application/json`
    :param post_params: request post parameters,
                        `application/x-www-form-urlencoded`
                        and `multipart/form-data`
    :param _request_timeout: timeout setting for this request. If one
                             number provided, it will be total request
                             timeout. It can also be a pair (tuple) of
                             (connection, read) timeouts.
    """
    method = method.upper()
    assert method in [
        'GET',
        'HEAD',
        'DELETE',
        'POST',
        'PUT',
        'PATCH',
        'OPTIONS'
    ]

    if post_params and body:
        raise ApiValueError(
            "body parameter cannot be used with post_params parameter."
        )

    post_params = post_params or {}
    headers = headers or {}

    timeout = None
    if _request_timeout:
        if isinstance(_request_timeout, (int, float)):
            timeout = urllib3.Timeout(total=_request_timeout)
        elif (
                isinstance(_request_timeout, tuple)
                and len(_request_timeout) == 2
            ):
            timeout = urllib3.Timeout(
                connect=_request_timeout[0],
                read=_request_timeout[1]
            )

    try:
        # For `POST`, `PUT`, `PATCH`, `OPTIONS`, `DELETE`
        if method in ['POST', 'PUT', 'PATCH', 'OPTIONS', 'DELETE']:

            # no content type provided or payload is json
            content_type = headers.get('Content-Type')
            if (
                not content_type
                or re.search('json', content_type, re.IGNORECASE)
            ):
                request_body = None
                if body is not None:
                    request_body = json.dumps(body)
                r = self.pool_manager.request(
                    method,
                    url,
                    body=request_body,
                    timeout=timeout,
                    headers=headers,
                    preload_content=False
                )
            elif content_type == 'application/x-www-form-urlencoded':
                r = self.pool_manager.request(
                    method,
                    url,
                    fields=post_params,
                    encode_multipart=False,
                    timeout=timeout,
                    headers=headers,
                    preload_content=False
                )
            elif content_type == 'multipart/form-data':
                # must del headers['Content-Type'], or the correct
                # Content-Type which generated by urllib3 will be
                # overwritten.
                del headers['Content-Type']
                r = self.pool_manager.request(
                    method,
                    url,
                    fields=post_params,
                    encode_multipart=True,
                    timeout=timeout,
                    headers=headers,
                    preload_content=False
                )
            # Pass a `string` parameter directly in the body to support
            # other content types than Json when `body` argument is
            # provided in serialized form
            elif isinstance(body, str) or isinstance(body, bytes):
                request_body = body
                r = self.pool_manager.request(
                    method,
                    url,
                    body=request_body,
                    timeout=timeout,
                    headers=headers,
                    preload_content=False
                )
            else:
                # Cannot generate the request from given parameters
                msg = """Cannot prepare a request message for provided
                         arguments. Please check that your arguments match
                         declared content type."""
                raise ApiException(status=0, reason=msg)
        # For `GET`, `HEAD`
        else:
            r = self.pool_manager.request(
                method,
                url,
                fields={},
                timeout=timeout,
                headers=headers,
                preload_content=False
            )
    except urllib3.exceptions.SSLError as e:
        msg = "\n".join([type(e).__name__, str(e)])
        raise ApiException(status=0, reason=msg)

    return RESTResponse(r)

RESTResponse

Bases: IOBase

Source code in cheshire_cat_api/rest.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
class RESTResponse(io.IOBase):

    def __init__(self, resp) -> None:
        self.response = resp
        self.status = resp.status
        self.reason = resp.reason
        self.data = None

    def read(self):
        if self.data is None:
            self.data = self.response.data
        return self.data

    def getheaders(self):
        """Returns a dictionary of the response headers."""
        return self.response.headers

    def getheader(self, name, default=None):
        """Returns a given response header."""
        return self.response.headers.get(name, default)

getheader(name, default=None)

Returns a given response header.

Source code in cheshire_cat_api/rest.py 
43
44
45
def getheader(self, name, default=None):
    """Returns a given response header."""
    return self.response.headers.get(name, default)

getheaders()

Returns a dictionary of the response headers.

Source code in cheshire_cat_api/rest.py 
39
40
41
def getheaders(self):
    """Returns a dictionary of the response headers."""
    return self.response.headers