#351

pending completion

Build # #351

Build Type

push

coveralls-python

Committed by

web-flow

Commit Message

Add devcontainer (#178)

* Remove Travis in favor of GHA

* Add Devcontainer for development

* Remove the pin on arrow lib

* Use flake8 inside tox

* Remove support for Python 3.6 and 3.7

Run Details

341 of 374 relevant lines covered (91.18%)

0.91 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

90.48

/tapioca/adapters.py

# coding: utf-8

import json
import xmltodict
from collections.abc import Mapping

from .tapioca import TapiocaInstantiator
from .exceptions import (
    ResponseProcessException, ClientError, ServerError)
from .serializers import SimpleSerializer


def generate_wrapper_from_adapter(adapter_class):
    return TapiocaInstantiator(adapter_class)


class TapiocaAdapter(object):
    serializer_class = SimpleSerializer

    def __init__(self, serializer_class=None, *args, **kwargs):
        if serializer_class:
            self.serializer = serializer_class()
        else:
            self.serializer = self.get_serializer()

    def _get_to_native_method(self, method_name, value):
        if not self.serializer:
            raise NotImplementedError("This client does not have a serializer")

        def to_native_wrapper(**kwargs):
            return self._value_to_native(method_name, value, **kwargs)

        return to_native_wrapper

    def _value_to_native(self, method_name, value, **kwargs):
        return self.serializer.deserialize(method_name, value, **kwargs)

    def get_serializer(self):
        if self.serializer_class:
            return self.serializer_class()

    def get_api_root(self, api_params, **kwargs):
        return self.api_root

    def fill_resource_template_url(self, template, params):
        return template.format(**params)

    def get_request_kwargs(self, api_params, *args, **kwargs):
        serialized = self.serialize_data(kwargs.get('data'))

        kwargs.update({
            'data': self.format_data_to_request(serialized),
        })
        return kwargs

    def get_error_message(self, data, response=None):
        return str(data)

    def process_response(self, response):
        if 500 <= response.status_code < 600:
            raise ResponseProcessException(ServerError, None)

        data = self.response_to_native(response)

        if 400 <= response.status_code < 500:
            raise ResponseProcessException(ClientError, data)

        return data

    def serialize_data(self, data):
        if self.serializer:
            return self.serializer.serialize(data)

        return data

    def format_data_to_request(self, data):
        raise NotImplementedError()

    def response_to_native(self, response):
        raise NotImplementedError()

    def get_iterator_list(self, response_data):
        raise NotImplementedError()

    def get_iterator_next_request_kwargs(self, iterator_request_kwargs,
                                         response_data, response):
        raise NotImplementedError()

    def is_authentication_expired(self, exception, *args, **kwargs):
        return False

    def refresh_authentication(self, api_params, *args, **kwargs):
        raise NotImplementedError()


class FormAdapterMixin(object):

    def format_data_to_request(self, data):
        return data

    def response_to_native(self, response):
        return {'text': response.text}


class JSONAdapterMixin(object):

    def get_request_kwargs(self, api_params, *args, **kwargs):
        arguments = super(JSONAdapterMixin, self).get_request_kwargs(
            api_params, *args, **kwargs)

        if 'headers' not in arguments:
            arguments['headers'] = {}
        arguments['headers']['Content-Type'] = 'application/json'
        return arguments

    def format_data_to_request(self, data):
        if data:
            return json.dumps(data)

    def response_to_native(self, response):
        if response.content.strip():
            return response.json()

    def get_error_message(self, data, response=None):
        if not data and response.content.strip():
            data = json.loads(response.content.decode('utf-8'))

        if data:
            return data.get('error', None)


class XMLAdapterMixin(object):

    def _input_branches_to_xml_bytestring(self, data):
        if isinstance(data, Mapping):
            return xmltodict.unparse(
                data, **self._xmltodict_unparse_kwargs).encode('utf-8')
        try:
            return data.encode('utf-8')
        except Exception as e:
            raise type(e)('Format not recognized, please enter an XML as string or a dictionary'
                          'in xmltodict spec: \n%s' % e.message)

    def get_request_kwargs(self, api_params, *args, **kwargs):
        # stores kwargs prefixed with 'xmltodict_unparse__' for use by xmltodict.unparse
        self._xmltodict_unparse_kwargs = {k[len('xmltodict_unparse__'):]: kwargs.pop(k)
                                          for k in kwargs.copy().keys()
                                          if k.startswith('xmltodict_unparse__')}
        # stores kwargs prefixed with 'xmltodict_parse__' for use by xmltodict.parse
        self._xmltodict_parse_kwargs = {k[len('xmltodict_parse__'):]: kwargs.pop(k)
                                        for k in kwargs.copy().keys()
                                        if k.startswith('xmltodict_parse__')}

        arguments = super(XMLAdapterMixin, self).get_request_kwargs(
            api_params, *args, **kwargs)

        if 'headers' not in arguments:
            arguments['headers'] = {}
        arguments['headers']['Content-Type'] = 'application/xml'
        return arguments

    def format_data_to_request(self, data):
        if data:
            return self._input_branches_to_xml_bytestring(data)

    def response_to_native(self, response):
        if response.content.strip():
            if 'xml' in response.headers['content-type']:
                return xmltodict.parse(response.content, **self._xmltodict_parse_kwargs)
            return {'text': response.text}

1	# coding: utf-8
2
3	import json	1✔
4	import xmltodict	1✔
5	from collections.abc import Mapping	1✔
6
7	from .tapioca import TapiocaInstantiator	1✔
8	from .exceptions import (	1✔
9	ResponseProcessException, ClientError, ServerError)
10	from .serializers import SimpleSerializer	1✔
11
12
13	def generate_wrapper_from_adapter(adapter_class):	1✔
14	return TapiocaInstantiator(adapter_class)	1✔
15
16
17	class TapiocaAdapter(object):	1✔
18	serializer_class = SimpleSerializer	1✔
19
20	def __init__(self, serializer_class=None, args, *kwargs):	1✔
21	if serializer_class:	1✔
22	self.serializer = serializer_class()	1✔
23	else:
24	self.serializer = self.get_serializer()	1✔
25
26	def _get_to_native_method(self, method_name, value):	1✔
27	if not self.serializer:	1✔
28	raise NotImplementedError("This client does not have a serializer")	1✔
29
30	def to_native_wrapper(**kwargs):	1✔
31	return self._value_to_native(method_name, value, **kwargs)	1✔
32
33	return to_native_wrapper	1✔
34
35	def _value_to_native(self, method_name, value, **kwargs):	1✔
36	return self.serializer.deserialize(method_name, value, **kwargs)	1✔
37
38	def get_serializer(self):	1✔
39	if self.serializer_class:	1✔
40	return self.serializer_class()	1✔
41
42	def get_api_root(self, api_params, **kwargs):	1✔
43	return self.api_root	1✔
44
45	def fill_resource_template_url(self, template, params):	1✔
46	return template.format(**params)	1✔
47
48	def get_request_kwargs(self, api_params, args, *kwargs):	1✔
49	serialized = self.serialize_data(kwargs.get('data'))	1✔
50
51	kwargs.update({	1✔
52	'data': self.format_data_to_request(serialized),
53	})
54	return kwargs	1✔
55
56	def get_error_message(self, data, response=None):	1✔
57	return str(data)	×
58
59	def process_response(self, response):	1✔
60	if 500 <= response.status_code < 600:	1✔
61	raise ResponseProcessException(ServerError, None)	1✔
62
63	data = self.response_to_native(response)	1✔
64
65	if 400 <= response.status_code < 500:	1✔
66	raise ResponseProcessException(ClientError, data)	1✔
67
68	return data	1✔
69
70	def serialize_data(self, data):	1✔
71	if self.serializer:	1✔
72	return self.serializer.serialize(data)	1✔
73
74	return data	1✔
75
76	def format_data_to_request(self, data):	1✔
77	raise NotImplementedError()	×
78
79	def response_to_native(self, response):	1✔
80	raise NotImplementedError()	×
81
82	def get_iterator_list(self, response_data):	1✔
83	raise NotImplementedError()	×
84
85	def get_iterator_next_request_kwargs(self, iterator_request_kwargs,	1✔
86	response_data, response):
87	raise NotImplementedError()	×
88
89	def is_authentication_expired(self, exception, args, *kwargs):	1✔
90	return False	1✔
91
92	def refresh_authentication(self, api_params, args, *kwargs):	1✔
93	raise NotImplementedError()	×
94
95
96	class FormAdapterMixin(object):	1✔
97
98	def format_data_to_request(self, data):	1✔
99	return data	×
100
101	def response_to_native(self, response):	1✔
102	return {'text': response.text}	×
103
104
105	class JSONAdapterMixin(object):	1✔
106
107	def get_request_kwargs(self, api_params, args, *kwargs):	1✔
108	arguments = super(JSONAdapterMixin, self).get_request_kwargs(	1✔
109	api_params, args, *kwargs)
110
111	if 'headers' not in arguments:	1✔
112	arguments['headers'] = {}	1✔
113	arguments['headers']['Content-Type'] = 'application/json'	1✔
114	return arguments	1✔
115
116	def format_data_to_request(self, data):	1✔
117	if data:	1✔
118	return json.dumps(data)	1✔
119
120	def response_to_native(self, response):	1✔
121	if response.content.strip():	1✔
122	return response.json()	1✔
123
124	def get_error_message(self, data, response=None):	1✔
125	if not data and response.content.strip():	1✔
126	data = json.loads(response.content.decode('utf-8'))	1✔
127
128	if data:	1✔
129	return data.get('error', None)	1✔
130
131
132	class XMLAdapterMixin(object):	1✔
133
134	def _input_branches_to_xml_bytestring(self, data):	1✔
135	if isinstance(data, Mapping):	1✔
136	return xmltodict.unparse(	1✔
137	data, **self._xmltodict_unparse_kwargs).encode('utf-8')
138	try:	1✔
139	return data.encode('utf-8')	1✔
140	except Exception as e:	×
141	raise type(e)('Format not recognized, please enter an XML as string or a dictionary'	×
142	'in xmltodict spec: \n%s' % e.message)
143
144	def get_request_kwargs(self, api_params, args, *kwargs):	1✔
145	# stores kwargs prefixed with 'xmltodict_unparse__' for use by xmltodict.unparse
146	self._xmltodict_unparse_kwargs = {k[len('xmltodict_unparse__'):]: kwargs.pop(k)	1✔
147	for k in kwargs.copy().keys()
148	if k.startswith('xmltodict_unparse__')}
149	# stores kwargs prefixed with 'xmltodict_parse__' for use by xmltodict.parse
150	self._xmltodict_parse_kwargs = {k[len('xmltodict_parse__'):]: kwargs.pop(k)	1✔
151	for k in kwargs.copy().keys()
152	if k.startswith('xmltodict_parse__')}
153
154	arguments = super(XMLAdapterMixin, self).get_request_kwargs(	1✔
155	api_params, args, *kwargs)
156
157	if 'headers' not in arguments:	1✔
158	arguments['headers'] = {}	1✔
159	arguments['headers']['Content-Type'] = 'application/xml'	1✔
160	return arguments	1✔
161
162	def format_data_to_request(self, data):	1✔
163	if data:	1✔
164	return self._input_branches_to_xml_bytestring(data)	1✔
165
166	def response_to_native(self, response):	1✔
167	if response.content.strip():	1✔
168	if 'xml' in response.headers['content-type']:	1✔
169	return xmltodict.parse(response.content, **self._xmltodict_parse_kwargs)	1✔
170	return {'text': response.text}	1✔

vintasoftware / tapioca-wrapper / #351

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous