diff options
Diffstat (limited to 'src/couch/resource.py')
| -rw-r--r-- | src/couch/resource.py | 60 |
1 files changed, 34 insertions, 26 deletions
diff --git a/src/couch/resource.py b/src/couch/resource.py index 364bff4..f110c8d 100644 --- a/src/couch/resource.py +++ b/src/couch/resource.py @@ -1,17 +1,25 @@ # -*- coding: utf-8 -*- # Based on py-couchdb (https://github.com/histrio/py-couchdb) - +from __future__ import annotations from __future__ import unicode_literals -from typing import Union, Tuple +from typing import Union, Tuple, Dict, Any import json import requests - - -from couch import utils -from couch import exceptions +from .utils import ( + urljoin, + as_json, + force_bytes, +) +from .exceptions import ( + GenericError, + NotFound, + BadRequest, + Conflict, + AuthenticationFailed, +) class Resource: @@ -40,12 +48,11 @@ class Resource: if method == "session": data = {"name": credentials[0], "password": credentials[1]} - data = utils.force_bytes(json.dumps(data)) - post_url = utils.urljoin(self.base_url, "_session") - r = self.session.post(post_url, data=data) - if r.status_code != 200: - raise exceptions.AuthenticationFailed() + post_url = urljoin(self.base_url, "_session") + r = self.session.post(post_url, data=force_bytes(json.dumps(data))) + if r and r.status_code != 200: + raise AuthenticationFailed() elif method == "basic": self.session.auth = credentials @@ -53,8 +60,8 @@ class Resource: else: raise RuntimeError("Invalid authentication method") - def __call__(self, *path: str): - base_url = utils.urljoin(self.base_url, *path) + def __call__(self, *path: str) -> Resource: + base_url = urljoin(self.base_url, *path) return self.__class__(base_url, session=self.session) def _check_result(self, response, result) -> None: @@ -68,17 +75,18 @@ class Resource: # This is here because couchdb can return http 201 # but containing a list of conflict errors if error == 'conflict' or error == "file_exists": - raise exceptions.Conflict(reason or "Conflict") + raise Conflict(reason or "Conflict") if response.status_code > 205: if response.status_code == 404 or error == 'not_found': - raise exceptions.NotFound(reason or 'Not found') + raise NotFound(reason or 'Not found') elif error == 'bad_request': - raise exceptions.BadRequest(reason or "Bad request") - raise exceptions.GenericError(result) + raise BadRequest(reason or "Bad request") + raise GenericError(result) - def request(self, method, path: str, params=None, data=None, - headers=None, stream=False, **kwargs): + + def request(self, method, path: Union[str, None], params=None, data=None, + headers=None, stream=False, **kwargs) -> Tuple[requests.models.Response, Union[Dict[str, Any], None]]: if headers is None: headers = {} @@ -88,7 +96,7 @@ class Resource: if path: if not isinstance(path, (list, tuple)): path = [path] - url = utils.urljoin(self.base_url, *path) + url = urljoin(self.base_url, *path) else: url = self.base_url @@ -102,7 +110,7 @@ class Resource: result = None self._check_result(response, result) else: - result = utils.as_json(response) + result = as_json(response) if result is None: return response, result @@ -115,17 +123,17 @@ class Resource: return response, result - def get(self, path: Union[str, None] = None, **kwargs): + def get(self, path: Union[str, None] = None, **kwargs: Any) -> Tuple[requests.models.Response, Union[Dict[str, Any], None]]: return self.request("GET", path, **kwargs) - def put(self, path: Union[str, None] = None, **kwargs): + def put(self, path: Union[str, None] = None, **kwargs: Any) -> Tuple[requests.models.Response, Union[Dict[str, Any], None]]: return self.request("PUT", path, **kwargs) - def post(self, path: Union[str, None] = None, **kwargs): + def post(self, path: Union[str, None] = None, **kwargs: Any) -> Tuple[requests.models.Response, Union[Dict[str, Any], None]]: return self.request("POST", path, **kwargs) - def delete(self, path: Union[str, None] = None, **kwargs): + def delete(self, path: Union[str, None] = None, **kwargs: Any) -> Tuple[requests.models.Response, Union[Dict[str, Any], None]]: return self.request("DELETE", path, **kwargs) - def head(self, path: Union[str, None] = None, **kwargs): + def head(self, path: Union[str, None] = None, **kwargs: Any) -> Tuple[requests.models.Response, Union[Dict[str, Any], None]]: return self.request("HEAD", path, **kwargs) |