# -*- coding: utf-8 -*-
from typing import Optional, Any, TypeVar
from .exceptions import YaDiskError
from .compat import Dict
from .utils import get_exception
from .objects import ErrorObject
from .types import ConsumeCallback, JSON, HTTPMethod, Payload, TimeoutParameter
__all__ = ["Session", "Response"]
[docs]
class Response:
"""
Represents an HTTP response.
In case an error occurs, methods of this class should throw one of exceptions
derived from :any:`YaDiskError`.
:ivar status: `int`, HTTP status code
"""
_Self = TypeVar("_Self", bound="Response")
status: int
[docs]
def json(self) -> JSON:
"""
Returns JSON-content of the response (parses JSON).
.. note::
This is an abstract method that needs to be implemented.
:returns: `dict`, `list`, `str`, `int`, `float` or `None`
"""
raise NotImplementedError
[docs]
def download(self, consume_callback: ConsumeCallback) -> None:
"""
Downloads response's content.
.. note::
This is an abstract method that needs to be implemented.
:param consume_callback: function, takes one parameter - chunk of data (bytes),
consumes the chunk (e.g. by writing to a file)
"""
raise NotImplementedError
[docs]
def get_exception(self) -> YaDiskError:
"""
Convenience wrapper for :any:`yadisk.utils.get_exception`.
"""
try:
js = self.json()
except (ValueError, RuntimeError):
js = None
error = ErrorObject(js)
return get_exception(self, error)
[docs]
def close(self) -> None:
"""
Closes the response and releases the underlying connection into the pool
.. note::
This is an abstract method that needs to be implemented.
"""
raise NotImplementedError
def __enter__(self: _Self) -> _Self:
return self
def __exit__(self, *args, **kwargs) -> None:
"""Closes the response and releases the underlying connection into the pool"""
self.close()
[docs]
class Session:
"""
HTTP session class. Maintains open connections, stores headers and
some other request parameters.
Must be explicitly closed (can be done using the `with` statement).
"""
_Self = TypeVar("_Self", bound="Session")
[docs]
def set_token(self, token: str) -> None:
"""
Sets token for the session by setting the Authorization header.
:param token: `str`, API token
"""
self.set_headers({"Authorization": "OAuth " + token})
[docs]
def send_request(self,
method: HTTPMethod,
url: str,
*,
params: Optional[Dict[str, Any]] = None,
data: Optional[Payload] = None,
timeout: TimeoutParameter = None,
stream: bool = False,
**kwargs) -> Response:
"""
Sends an HTTP request with given parameters.
In case an error occurs, the method should throw one of exceptions
derived from :any:`YaDiskError`.
Additional keyword arguments may be passed, they may be forwarded
to the underlying HTTP client without modification.
.. note::
This is an abstract method that needs to be implemented.
:param method: `str`, HTTP method
:param url: `str`, URL
:param params: `dict`, GET parameters
:param data: `bytes`, an iterator or a file-like object, data to be sent in the request body
:param headers: `dict`, additional headers to be set
:param timeout: request timeout, a `tuple` of `(read timeout, connect timeout)`, `float` or `None` (no timeout)
:param stream: `bool`, if `False`, the response content will be immediately downloaded
:returns: :any:`Response`, response object
"""
raise NotImplementedError
[docs]
def close(self) -> None:
"""
Closes the session.
.. note::
This is an abstract method that needs to be implemented.
"""
raise NotImplementedError
def __enter__(self: _Self) -> _Self:
return self
def __exit__(self, *args, **kwargs) -> None:
"""Closes the session."""
return self.close()