bamboo.request
Response
Response returned by request functions of http
or https
modules.
Examples:
from bamboo.request import http
with http.get("http://localhost:8000/hello") as res:
headers = res.headers
if res.ok:
body = res.body
print(body)
else:
print(f"Error occured. Response status: {res.status}")
# The session will be automatically closed by leaving the block.
content_length: t.Optional[int]
property
readonly
Content length of the response if existing, None otherwise.
fileno: int
property
readonly
File number of the socket.
headers: http.client.HTTPMessage
property
readonly
All response headers.
is_closed: bool
property
readonly
If the session is closed or not.
ok: bool
property
readonly
If request succeeded or not.
status: int
property
readonly
Response status.
uri: str
property
readonly
Requested URI.
version: int
property
readonly
HTTP version of the session.
__init__(self, conn, res, uri, datacls=<class 'bamboo.api.base.BinaryApiData'>)
special
Parameters:
Name | Type | Description | Default |
---|---|---|---|
conn |
http.client.HTTPConnection |
connection object of a session. |
required |
res |
http.client.HTTPResponse |
HTTPResponse of a request. |
required |
uri |
str |
Requested URI. |
required |
datacls |
t.Type[ResponseData_t] |
ApiData class to attach raw response body. |
<class 'bamboo.api.base.BinaryApiData'> |
Source code in bamboo/request/response.py
def __init__(
self,
conn: http.client.HTTPConnection,
res: http.client.HTTPResponse,
uri: str,
datacls: t.Type[ResponseData_t] = BinaryApiData,
) -> None:
"""
Args:
conn: connection object of a session.
res: HTTPResponse of a request.
uri: Requested URI.
datacls: ApiData class to attach raw response body.
"""
self._conn = conn
self._res = res
self._uri = uri
self._datacls = datacls
self._is_read = False
attach(self, datacls=None)
Generate new object of specified ApiData from response body.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
datacls |
t.Optional[t.Type[ResponseData_t]] |
ApiData class or its subclass. |
None |
Returns:
Type | Description |
---|---|
ResponseData_t |
ResponseData_t: Generated object. |
Source code in bamboo/request/response.py
def attach(
self,
datacls: t.Optional[t.Type[ResponseData_t]] = None,
) -> ResponseData_t:
"""Generate new object of specified ApiData from response body.
Args:
datacls: ApiData class or its subclass.
Returns:
ResponseData_t: Generated object.
"""
content_type_raw = self.get_header("Content-Type")
if content_type_raw:
content_type = ContentType.parse(content_type_raw)
if datacls is None:
if content_type_raw is None:
content_type = self._datacls.__content_type__
return self._datacls.__validate__(self.body, content_type)
else:
if content_type_raw is None:
content_type = datacls.__content_type__
return datacls.__validate__(self.body, content_type)
close(self)
Close the session.
Source code in bamboo/request/response.py
def close(self) -> None:
"""Close the session.
"""
self._conn.close()
get_header(self, name)
Retrive header value from response headers.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
name |
str |
Header name. |
required |
Returns:
Type | Description |
---|---|
t.Optional[str] |
Value of header if existing, None otherwise. |
Source code in bamboo/request/response.py
def get_header(self, name: str) -> t.Optional[str]:
"""Retrive header value from response headers.
Args:
name: Header name.
Returns:
Value of header if existing, None otherwise.
"""
return self._res.getheader(name)
read(self, amt=None)
Reads and returns the response body.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
amt |
t.Optional[int] |
Amount of the binary, all of it if None. |
None |
Returns:
Type | Description |
---|---|
bytes |
Results of reading. Its length may be less than |
Source code in bamboo/request/response.py
def read(self, amt: t.Optional[int] = None) -> bytes:
"""Reads and returns the response body.
Args:
amt: Amount of the binary, all of it if None.
Returns:
Results of reading. Its length may be less than `amt`.
"""
if not self._is_read:
self._is_read = True
return self._res.read(amt)
ResponseBodyAlreadyReadError
Response body has already been read and data consistensy would be broken.