ParameterBuilder.create_url() - Code Metrics - Inspection of "add ParameterBuilder function" - peshay/btcde - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Branch — refactor (40cdd7)

by Andreas

created 2017-12-28 22:48 UTC

ParameterBuilder.create_url() A

↳ Parent: ParameterBuilder

Complexity

Conditions

Size

Total Lines

Duplication

Lines	10
Ratio	100 %

Importance

Changes	1
Bugs	0	Features	1

Metric	Value
cc	3
c	1
b	0
f	1
dl	10
loc	10
rs	9.4285

#! /usr/bin/env python
"""This Module provides the functions from bitcoin.de Trading API"""

import requests
import time
import json
import hmac
import hashlib
import logging
import codecs

logging.basicConfig()
logging.getLogger().setLevel(logging.DEBUG)
log = logging.getLogger(__name__)
requests_log = logging.getLogger("requests.packages.urllib3")
requests_log.setLevel(logging.DEBUG)
requests_log.propagate = True

__version__ = '2.0'

# disable unsecure SSL warning
requests.packages.urllib3.disable_warnings()

class ParameterBuilder(object):
    '''To verify given parameters for API.'''
    def __init__(self, avail_params, given_params, uri):
        self.verify_keys_and_values(avail_params, given_params)
        self.params = given_params
        self.create_url(uri)

    def verify_keys_and_values(self, avail_params, given_params):
        for k, v in given_params.items():
            if k not in avail_params:
                list_string = ', '.join(avail_params)
                raise KeyError("{} is not any of {}".format(k, list_string))
            if k == 'trading_pair':
                self.error_on_invalid_value(v, self.TRADING_PAIRS)
            elif k == 'type':
                self.error_on_invalid_value(v, self.ORDER_TYPES)
            elif k == 'currency':
                self.error_on_invalid_value(v, self.CURRENCIES)
            elif k == 'seat_of_bank':
                self.error_on_invalid_value(v, self.BANK_SEATS)
            elif k in ['min_trust_level', 'trust_level']:
                self.error_on_invalid_value(v, self.TRUST_LEVELS)
            elif k == 'payment_option':
                self.error_on_invalid_value(v, self.PAYMENT_OPTIONS)
            elif k == 'state':
                self.error_on_invalid_value(v, self.STATES)
    
    def error_on_invalid_value(self, value, list):
        if value not in list:
            list_string = ', '.join(list)
            raise ValueError("{} is not any of {}".format(value, list_string))

    def create_url(self, uri):

        if self.params:
            self.encoded_string = ''
            for key, value in sorted(self.params.items()):
                self.encoded_string += str(key) + '=' + str(value) + '&'
            self.encoded_string = self.encoded_string[:-1]
            self.url = uri + '?' + self.encoded_string
        else:
            self.encoded_string = ''
            self.url = uri

    
    TRADING_PAIRS = ['btceur', 'bcheur', 'etheur']
    ORDER_TYPES = ['buy', 'sell']
    CURRENCIES = ['btc', 'bch', 'eth']
    BANK_SEATS = ['AT', 'BE', 'BG', 'CH', 'CY', 'CZ',
                  'DE', 'DK', 'EE', 'ES', 'FI', 'FR',
                  'GB', 'GR', 'HR', 'HU', 'IE', 'IS',
                  'IT', 'LI', 'LT', 'LU', 'LV', 'MT',
                  'MQ', 'NL', 'NO', 'PL', 'PT', 'RO',
                  'SE', 'SI', 'SK']
    TRUST_LEVELS = ['bronze', 'silver', 'gold', 'platin']
    STATES = [-1, 0, 1]
    PAYMENT_OPTIONS = [1, 2, 3]
    TRADE_TYPES = ['all', 'buy', 'sell', 'inpayment',
                   'payout', 'affiliate', 'welcome_btc',
                   'buy_yubikey', 'buy_goldshop',
                   'buy_diamondshop', 'kickback',
                   'outgoing_fee_voluntary']

def HandleRequestsException(e):
    """Handle Exception from request."""
    log.warning(e)


def HandleAPIErrors(r):
    """To handle Errors from BTCDE API."""
    valid_status_codes = [200, 201, 204]
    if r.status_code not in valid_status_codes:
        reader = codecs.getreader("utf-8")
        content = json.load(reader(r.raw))
        errors = content.get('errors')
        log.warning('API Error Code: {}'.format(str(errors[0]['code'])))
        log.warning('API Error Message: {}'.format(errors[0]['message']))
        log.warning('API Error URL: {}'.format(r.url))
        return False
    else:
        return True


class Connection(object):
    """To provide connection credentials to the trading API"""
    def __init__(self, api_key, api_secret):
        self.api_key = api_key
        self.api_secret = api_secret
        # set initial self.nonce
        self.nonce = int(time.time())
        # Bitcoin.de API URI
        self.apihost = 'https://api.bitcoin.de'
        self.apiversion = 'v2'
        self.orderuri = self.apihost + '/' + self.apiversion + '/' + 'orders'
        self.tradeuri = self.apihost + '/' + self.apiversion + '/' + 'trades'
        self.accounturi = self.apihost + '/' + self.apiversion + '/' + 'account'

    def build_hmac_sign(self, md5string, method, url):
        hmac_data = '{method}#{url}#{key}#{nonce}#{md5}'\
                    .format(method=method, url=url,
                            key=self.api_key, nonce=str(self.nonce),
                            md5=md5string)
        hmac_signed = hmac.new(bytearray(self.api_secret.encode()), msg=hmac_data.encode(), digestmod=hashlib.sha256).hexdigest()
        return hmac_signed
        
    def set_header(self, url, method, encoded_string):
        # raise self.nonce before using
        self.nonce += 1
        if method == 'POST':
            md5_encoded_query_string = hashlib.md5(encoded_string.encode()).hexdigest()
        else:
            md5_encoded_query_string = hashlib.md5(b'').hexdigest()
        hmac_signed = self.build_hmac_sign(md5_encoded_query_string,
                                           method, url)
        # set header
        header = {'content-type':
                  'application/x-www-form-urlencoded; charset=utf-8',
                  'X-API-KEY': self.api_key,
                  'X-API-NONCE': str(self.nonce),
                  'X-API-SIGNATURE': hmac_signed }
        return header

    def send_request(self, url, method, header, encoded_string):
        if method == 'GET':
            r = requests.get(url, headers=(header),
                             stream=True, verify=False)
        elif method == 'POST':
            r = requests.post(url, headers=(header), data=encoded_string,
                              stream=True, verify=False)
        elif method == 'DELETE':
            r = requests.delete(url, headers=(header),
                                stream=True, verify=False)
        return r

    def APIConnect(self, method, params):
        """Transform Parameters to URL"""
        header = self.set_header(params.url, method,
                                 params.encoded_string)
        log.debug('Set Header: {}'.format(header))
        try:
            r = self.send_request(params.url, method, header,
                                  params.encoded_string)
            # Handle API Errors
            if HandleAPIErrors(r):
                # get results
                result = r.json()
            else:
                result = {}
        except requests.exceptions.RequestException as e:
            HandleRequestsException(e)
            result = {}
        return result

    def showOrderbook(self, order_type, trading_pair, **args):
        """Search Orderbook for offers."""
        params = {'type': order_type,
                  'trading_pair': trading_pair}
        params.update(args)
        avail_params = ['type', 'trading_pair', 'amount', 'price',
                        'order_requirements_fullfilled',
                        'only_kyc_full', 'only_express_orders',
                        'only_same_bankgroup', 'only_same_bic',
                        'seat_of_bank']
        p = ParameterBuilder(avail_params, params, self.orderuri)
        return self.APIConnect('GET', p)


    def createOrder(self, order_type, trading_pair, max_amount, price, **args):
        """Create a new Order."""
        # Build parameters
        params = {'type': order_type,
                  'max_amount': max_amount,
                  'price': price,
                  'trading_pair': trading_pair}
        params.update(args)
        avail_params = ['type', 'trading_pair', 'max_amount', 'price',
                        'min_amount', 'new_order_for_remaining_amount',
                        'min_trust_level', 'only_kyc_full', 'payment_option',
                        'seat_of_bank']
        p = ParameterBuilder(avail_params, params, self.orderuri)
        return self.APIConnect('POST', p)


    def deleteOrder(self, order_id, trading_pair):
        """Delete an Order."""
        # Build parameters
        params = {'order_id': order_id,
                  'trading_pair': trading_pair}
        avail_params = ['order_id', 'trading_pair']
        newuri = self.orderuri + "/" + order_id + "/" + trading_pair
        p = ParameterBuilder(avail_params, params, newuri)
        p.encoded_string = ''
        p.url = newuri
        return self.APIConnect('DELETE', p)


    def showMyOrders(self, **args):
        """Query and Filter own Orders."""
        # Build parameters
        params = args
        avail_params = ['type', 'trading_pair', 'state',
                        'date_start', 'date_end', 'page']
        newuri = self.orderuri + '/my_own'
        p = ParameterBuilder(avail_params, params, newuri)
        return self.APIConnect('GET', p)


    def showMyOrderDetails(self, order_id):
        """Details to an own Order."""
        newuri = self.orderuri + '/' + order_id
        p = ParameterBuilder({}, {}, newuri)
        return self.APIConnect('GET', p)


    def executeTrade(self, order_id, order_type, trading_pair, amount):
        """Buy/Sell on a specific Order."""
        newuri = self.tradeuri + '/' + order_id
        params = {'order_id': order_id,
                  'type': order_type,
                  'trading_pair': trading_pair,
                  'amount': amount}
        avail_params = ['order_id', 'type', 'trading_pair',
                        'amount']
        p = ParameterBuilder(avail_params, params, newuri)
        return self.APIConnect('POST', p)


    def showMyTrades(self, **args):
        """Query and Filter on past Trades."""
        # Build parameters
        params = args
        avail_params = ['type', 'trading_pair', 'state',
                        'date_start', 'date_end', 'page']
        p = ParameterBuilder(avail_params, params, self.tradeuri)
        return self.APIConnect('GET', p)


    def showMyTradeDetails(self, trade_id):
        """Details to a specific Trade."""
        newuri = self.tradeuri + '/' + trade_id
        params = {}
        p = ParameterBuilder({}, {}, newuri)
        return self.APIConnect('GET', p)


    def showAccountInfo(self):
        """Query on Account Infos."""
        p = ParameterBuilder({}, {}, self.accounturi)
        return self.APIConnect('GET', p)


    def showOrderbookCompact(self, trading_pair):
        """Bids and Asks in compact format."""
        params = {'trading_pair': trading_pair}
        # Build parameters
        avail_params = ['trading_pair']
        p = ParameterBuilder(avail_params, params,
                             self.orderuri + '/compact')
        return self.APIConnect('GET', p)


    def showPublicTradeHistory(self, trading_pair, **args):
        """All successful trades of the las 7 days."""
        params = {'trading_pair': trading_pair}
        params.update(args)
        avail_params = ['trading_pair', 'since_tid']
        p = ParameterBuilder(avail_params, params,
                             self.tradeuri + '/history')
        return self.APIConnect('GET', p)


    def showRates(self, trading_pair):
        """Query of the average rate last 3 and 12 hours."""
        newuri = self.apihost + '/' + self.apiversion + '/rates'
        params = {'trading_pair': trading_pair}
        avail_params = ['trading_pair']
        p = ParameterBuilder(avail_params, params, newuri)
        return self.APIConnect('GET', p)


    def showAccountLedger(self, currency, **args):
        """Query on Account statement."""
        params = {'currency': currency}
        params.update(args)
        avail_params = ['currency', 'type',
                        'date_start', 'date_end', 'page']
        p = ParameterBuilder(avail_params, params,
                             self.accounturi + '/ledger')
        return self.APIConnect('GET', p)


1		#! /usr/bin/env python
2		"""This Module provides the functions from bitcoin.de Trading API"""
3
4		import requests
5		import time
6		import json
7		import hmac
8		import hashlib
9		import logging
10		import codecs
11
12		logging.basicConfig()
13		logging.getLogger().setLevel(logging.DEBUG)
14		log = logging.getLogger(__name__)
15		requests_log = logging.getLogger("requests.packages.urllib3")
16		requests_log.setLevel(logging.DEBUG)
17		requests_log.propagate = True
18
19		__version__ = '2.0'
20
21		# disable unsecure SSL warning
22		requests.packages.urllib3.disable_warnings()
23
24		class ParameterBuilder(object):
25		'''To verify given parameters for API.'''
26		def __init__(self, avail_params, given_params, uri):
27		self.verify_keys_and_values(avail_params, given_params)
28		self.params = given_params
29		self.create_url(uri)
30
31		def verify_keys_and_values(self, avail_params, given_params):
32		for k, v in given_params.items():
33		if k not in avail_params:
34		list_string = ', '.join(avail_params)
35		raise KeyError("{} is not any of {}".format(k, list_string))
36		if k == 'trading_pair':
37		self.error_on_invalid_value(v, self.TRADING_PAIRS)
38		elif k == 'type':
39		self.error_on_invalid_value(v, self.ORDER_TYPES)
40		elif k == 'currency':
41		self.error_on_invalid_value(v, self.CURRENCIES)
42		elif k == 'seat_of_bank':
43		self.error_on_invalid_value(v, self.BANK_SEATS)
44		elif k in ['min_trust_level', 'trust_level']:
45		self.error_on_invalid_value(v, self.TRUST_LEVELS)
46		elif k == 'payment_option':
47		self.error_on_invalid_value(v, self.PAYMENT_OPTIONS)
48		elif k == 'state':
49		self.error_on_invalid_value(v, self.STATES)
50
51		def error_on_invalid_value(self, value, list):
52		if value not in list:
53		list_string = ', '.join(list)
54		raise ValueError("{} is not any of {}".format(value, list_string))
55
56	View Code Duplication	def create_url(self, uri):
		0 ignored issues – show Duplication introduced 2017-12-28 22:48 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
57		if self.params:
58		self.encoded_string = ''
59		for key, value in sorted(self.params.items()):
60		self.encoded_string += str(key) + '=' + str(value) + '&'
61		self.encoded_string = self.encoded_string[:-1]
62		self.url = uri + '?' + self.encoded_string
63		else:
64		self.encoded_string = ''
65		self.url = uri
66
67
68		TRADING_PAIRS = ['btceur', 'bcheur', 'etheur']
69		ORDER_TYPES = ['buy', 'sell']
70		CURRENCIES = ['btc', 'bch', 'eth']
71		BANK_SEATS = ['AT', 'BE', 'BG', 'CH', 'CY', 'CZ',
72		'DE', 'DK', 'EE', 'ES', 'FI', 'FR',
73		'GB', 'GR', 'HR', 'HU', 'IE', 'IS',
74		'IT', 'LI', 'LT', 'LU', 'LV', 'MT',
75		'MQ', 'NL', 'NO', 'PL', 'PT', 'RO',
76		'SE', 'SI', 'SK']
77		TRUST_LEVELS = ['bronze', 'silver', 'gold', 'platin']
78		STATES = [-1, 0, 1]
79		PAYMENT_OPTIONS = [1, 2, 3]
80		TRADE_TYPES = ['all', 'buy', 'sell', 'inpayment',
81		'payout', 'affiliate', 'welcome_btc',
82		'buy_yubikey', 'buy_goldshop',
83		'buy_diamondshop', 'kickback',
84		'outgoing_fee_voluntary']
85
86		def HandleRequestsException(e):
87		"""Handle Exception from request."""
88		log.warning(e)
89
90
91		def HandleAPIErrors(r):
92		"""To handle Errors from BTCDE API."""
93		valid_status_codes = [200, 201, 204]
94		if r.status_code not in valid_status_codes:
95		reader = codecs.getreader("utf-8")
96		content = json.load(reader(r.raw))
97		errors = content.get('errors')
98		log.warning('API Error Code: {}'.format(str(errors[0]['code'])))
99		log.warning('API Error Message: {}'.format(errors[0]['message']))
100		log.warning('API Error URL: {}'.format(r.url))
101		return False
102		else:
103		return True
104
105
106		class Connection(object):
107		"""To provide connection credentials to the trading API"""
108		def __init__(self, api_key, api_secret):
109		self.api_key = api_key
110		self.api_secret = api_secret
111		# set initial self.nonce
112		self.nonce = int(time.time())
113		# Bitcoin.de API URI
114		self.apihost = 'https://api.bitcoin.de'
115		self.apiversion = 'v2'
116		self.orderuri = self.apihost + '/' + self.apiversion + '/' + 'orders'
117		self.tradeuri = self.apihost + '/' + self.apiversion + '/' + 'trades'
118		self.accounturi = self.apihost + '/' + self.apiversion + '/' + 'account'
119
120		def build_hmac_sign(self, md5string, method, url):
121		hmac_data = '{method}#{url}#{key}#{nonce}#{md5}'\
122		.format(method=method, url=url,
123		key=self.api_key, nonce=str(self.nonce),
124		md5=md5string)
125		hmac_signed = hmac.new(bytearray(self.api_secret.encode()), msg=hmac_data.encode(), digestmod=hashlib.sha256).hexdigest()
126		return hmac_signed
127
128		def set_header(self, url, method, encoded_string):
129		# raise self.nonce before using
130		self.nonce += 1
131		if method == 'POST':
132		md5_encoded_query_string = hashlib.md5(encoded_string.encode()).hexdigest()
133		else:
134		md5_encoded_query_string = hashlib.md5(b'').hexdigest()
135		hmac_signed = self.build_hmac_sign(md5_encoded_query_string,
136		method, url)
137		# set header
138		header = {'content-type':
139		'application/x-www-form-urlencoded; charset=utf-8',
140		'X-API-KEY': self.api_key,
141		'X-API-NONCE': str(self.nonce),
142		'X-API-SIGNATURE': hmac_signed }
143		return header
144
145		def send_request(self, url, method, header, encoded_string):
146		if method == 'GET':
147		r = requests.get(url, headers=(header),
148		stream=True, verify=False)
149		elif method == 'POST':
150		r = requests.post(url, headers=(header), data=encoded_string,
151		stream=True, verify=False)
152		elif method == 'DELETE':
153		r = requests.delete(url, headers=(header),
154		stream=True, verify=False)
155		return r
156
157		def APIConnect(self, method, params):
158		"""Transform Parameters to URL"""
159		header = self.set_header(params.url, method,
160		params.encoded_string)
161		log.debug('Set Header: {}'.format(header))
162		try:
163		r = self.send_request(params.url, method, header,
164		params.encoded_string)
165		# Handle API Errors
166		if HandleAPIErrors(r):
167		# get results
168		result = r.json()
169		else:
170		result = {}
171		except requests.exceptions.RequestException as e:
172		HandleRequestsException(e)
173		result = {}
174		return result
175
176		def showOrderbook(self, order_type, trading_pair, **args):
177		"""Search Orderbook for offers."""
178		params = {'type': order_type,
179		'trading_pair': trading_pair}
180		params.update(args)
181		avail_params = ['type', 'trading_pair', 'amount', 'price',
182		'order_requirements_fullfilled',
183		'only_kyc_full', 'only_express_orders',
184		'only_same_bankgroup', 'only_same_bic',
185		'seat_of_bank']
186		p = ParameterBuilder(avail_params, params, self.orderuri)
187		return self.APIConnect('GET', p)
188
189
190		def createOrder(self, order_type, trading_pair, max_amount, price, **args):
191		"""Create a new Order."""
192		# Build parameters
193		params = {'type': order_type,
194		'max_amount': max_amount,
195		'price': price,
196		'trading_pair': trading_pair}
197		params.update(args)
198		avail_params = ['type', 'trading_pair', 'max_amount', 'price',
199		'min_amount', 'new_order_for_remaining_amount',
200		'min_trust_level', 'only_kyc_full', 'payment_option',
201		'seat_of_bank']
202		p = ParameterBuilder(avail_params, params, self.orderuri)
203		return self.APIConnect('POST', p)
204
205
206		def deleteOrder(self, order_id, trading_pair):
207		"""Delete an Order."""
208		# Build parameters
209		params = {'order_id': order_id,
210		'trading_pair': trading_pair}
211		avail_params = ['order_id', 'trading_pair']
212		newuri = self.orderuri + "/" + order_id + "/" + trading_pair
213		p = ParameterBuilder(avail_params, params, newuri)
214		p.encoded_string = ''
215		p.url = newuri
216		return self.APIConnect('DELETE', p)
217
218
219		def showMyOrders(self, **args):
220		"""Query and Filter own Orders."""
221		# Build parameters
222		params = args
223		avail_params = ['type', 'trading_pair', 'state',
224		'date_start', 'date_end', 'page']
225		newuri = self.orderuri + '/my_own'
226		p = ParameterBuilder(avail_params, params, newuri)
227		return self.APIConnect('GET', p)
228
229
230		def showMyOrderDetails(self, order_id):
231		"""Details to an own Order."""
232		newuri = self.orderuri + '/' + order_id
233		p = ParameterBuilder({}, {}, newuri)
234		return self.APIConnect('GET', p)
235
236
237		def executeTrade(self, order_id, order_type, trading_pair, amount):
238		"""Buy/Sell on a specific Order."""
239		newuri = self.tradeuri + '/' + order_id
240		params = {'order_id': order_id,
241		'type': order_type,
242		'trading_pair': trading_pair,
243		'amount': amount}
244		avail_params = ['order_id', 'type', 'trading_pair',
245		'amount']
246		p = ParameterBuilder(avail_params, params, newuri)
247		return self.APIConnect('POST', p)
248
249
250		def showMyTrades(self, **args):
251		"""Query and Filter on past Trades."""
252		# Build parameters
253		params = args
254		avail_params = ['type', 'trading_pair', 'state',
255		'date_start', 'date_end', 'page']
256		p = ParameterBuilder(avail_params, params, self.tradeuri)
257		return self.APIConnect('GET', p)
258
259
260		def showMyTradeDetails(self, trade_id):
261		"""Details to a specific Trade."""
262		newuri = self.tradeuri + '/' + trade_id
263		params = {}
264		p = ParameterBuilder({}, {}, newuri)
265		return self.APIConnect('GET', p)
266
267
268		def showAccountInfo(self):
269		"""Query on Account Infos."""
270		p = ParameterBuilder({}, {}, self.accounturi)
271		return self.APIConnect('GET', p)
272
273
274		def showOrderbookCompact(self, trading_pair):
275		"""Bids and Asks in compact format."""
276		params = {'trading_pair': trading_pair}
277		# Build parameters
278		avail_params = ['trading_pair']
279		p = ParameterBuilder(avail_params, params,
280		self.orderuri + '/compact')
281		return self.APIConnect('GET', p)
282
283
284		def showPublicTradeHistory(self, trading_pair, **args):
285		"""All successful trades of the las 7 days."""
286		params = {'trading_pair': trading_pair}
287		params.update(args)
288		avail_params = ['trading_pair', 'since_tid']
289		p = ParameterBuilder(avail_params, params,
290		self.tradeuri + '/history')
291		return self.APIConnect('GET', p)
292
293
294		def showRates(self, trading_pair):
295		"""Query of the average rate last 3 and 12 hours."""
296		newuri = self.apihost + '/' + self.apiversion + '/rates'
297		params = {'trading_pair': trading_pair}
298		avail_params = ['trading_pair']
299		p = ParameterBuilder(avail_params, params, newuri)
300		return self.APIConnect('GET', p)
301
302
303		def showAccountLedger(self, currency, **args):
304		"""Query on Account statement."""
305		params = {'currency': currency}
306		params.update(args)
307		avail_params = ['currency', 'type',
308		'date_start', 'date_end', 'page']
309		p = ParameterBuilder(avail_params, params,
310		self.accounturi + '/ledger')
311		return self.APIConnect('GET', p)
312

peshay / btcde

Branch — refactor (40cdd7)

ParameterBuilder.create_url() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like