Source code for pyrate.main
from base64 import b64encode
import json
import requests
import sys
__docformat__ = 'sphinx en'
[docs]class Pyrate(object):
"""This is the main class
:param list http_methods: List of available HTTP methods for this service
:param list return_formats: List of available return formats for this service
:param default_header_content: Default content for the request header
:param default_body_content: Default content for the request body
:param string default_http_method: Default HTTP method (will be used if none else is specified in request)
:param string default_return_format: Default return format (will be used if none else is specified in request)
:param string connection_check_method: Used by :func:`check_connection`
:param string auth_type: The authentification type. Obsolete.
:param string base_url: The base url for all api requests
:param bool send_json: Whether the request body should be encoded with json
"""
http_methods = ['GET', 'POST', 'PUT', 'DELETE', 'OPTIONS']
return_formats = ['json']
default_header_content = None
default_body_content = None
default_http_method = None
default_return_format = None
connection_check_method = None
auth_type = None
base_url = None
send_json = False
def __init__(self):
self.default_http_method = self.http_methods[0]
try:
self.default_return_format = self.return_formats[0]
except IndexError:
self.default_return_format = ''
[docs] def create_basic_auth(self, user, password):
"""Creates the header content for HTTP Basic Authentification.
:param user: Username
:param password: Password
:rtype: Base64-encoded auth string
"""
# Messing around with Python3's strictness about strings
if sys.version_info >= (3, 0):
if not isinstance(user, str):
user = user.decode('utf-8')
if not isinstance(password, str):
password = password.decode('utf-8')
return 'Basic ' + b64encode((user + ":" + password).encode('utf-8')).decode('utf-8')
else:
return 'Basic ' + b64encode(user + ":" + password).rstrip()
[docs] def get_oauth(self):
raise NotImplementedError("OAuth methods need to be implemented by subclasses!")
[docs] def check_connection(self):
return self.do(self.connection_check_method[1], http_method=self.connection_check_method[0])
[docs] def build_content(self, args):
# takes a dictionary, filters out all the empty stuff
if 'self' in args:
del args['self']
new_args = args.copy()
for key in args:
if not args[key]:
del new_args[key]
return new_args
[docs] def check_response_success(self, response):
raise NotImplementedError('Please implement in subclass')
[docs] def parse_errors(self, response):
raise NotImplementedError('Please implement in subclass')
[docs] def do(self, method, content=None, headers=None, http_method=None, return_format=None):
request_body = self.default_body_content
if content is not None:
request_body.update(content)
request_headers = self.default_header_content
if headers is not None:
request_headers.update(headers)
if http_method is None:
http_method = self.default_http_method
if return_format is None:
if self.default_return_format:
return_format = "." + self.default_return_format
else:
return_format = ''
request_url = self.base_url + method + return_format
return self.do_request(http_method, request_url, request_headers, request_body, return_format)
[docs] def do_request(self, http_method, url, headers, body, return_format):
if self.auth_type == 'OAUTH1':
auth_data = self.get_oauth()
else:
auth_data = None
if self.send_json:
# We need to make sure that body is jsonified
try:
body = json.dumps(body)
except TypeError or ValueError:
pass
if http_method.upper() == 'GET':
r = requests.get(url, headers=headers, auth=auth_data)
elif http_method.upper() == 'POST':
r = requests.post(url, data=body, headers=headers, auth=auth_data)
elif http_method.upper() == 'PUT':
r = requests.put(url, data=body, headers=headers, auth=auth_data)
elif http_method.upper() == 'DELETE':
r = requests.delete(url, data=body, headers=headers, auth=auth_data)
elif http_method.upper() == 'OPTIONS':
r = requests.options(url, data=body, headers=headers, auth=auth_data)
else:
raise Exception("Invalid request method")
return self.handle_response(r, return_format)
[docs] def handle_response(self, response, return_format):
try:
return response.json()
except ValueError:
return response.content