set_header() - Code Metrics - Inspection of "Make own func for header" - peshay/btcde - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — testcase ( d1c203...8af9d3 )

by Andreas

created 2017-09-16 22:10 UTC

set_header() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	2
c	1
b	0
f	0
dl	0
loc	19
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

# these two lines enable debugging at httplib level
# (requests->urllib3->httplib)
# you will see the REQUEST, including HEADERS and DATA, and RESPONSE with
# HEADERS but without DATA.
# the only thing missing will be the response.body which is not logged.
import http.client
http.client.HTTPConnection.debuglevel = 1


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

__version__ = '1.4'


class Connection:
    """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

# Bitcoin.de API URI
apihost = 'https://api.bitcoin.de'
apiversion = 'v1'
orderuri = apihost + '/' + apiversion + '/' + 'orders'
tradeuri = apihost + '/' + apiversion + '/' + 'trades'
accounturi = apihost + '/' + apiversion + '/' + 'account'
# set initial nonce
nonce = int(time.time())
# disable unsecure SSL warning
requests.packages.urllib3.disable_warnings()


def HandleRequestsException(e):
    """Handle Exception from request."""
    print(e[0][0])
    print(e[0][1])


def HandleAPIErrors(r):
    """To handle Errors from BTCDE API."""
    if r.status_code != 200 and r.status_code != 201 and r.status_code != 204:
        reader = codecs.getreader("utf-8")
        content = json.load(reader(r.raw))
        errors = content.get('errors')
        print('Code:     ' + str(errors[0]['code']))
        print('Message:  ' + errors[0]['message'])
        print('With URL: ' + r.url)
        return False
    else:
        return True


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

def set_header(url):
    global nonce
    # raise nonce before using
    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_data = method + '#' + \
        url + '#' + conn.api_key + \
        '#' + str(nonce) + '#' + md5_encoded_query_string
    hmac_signed = hmac.new(bytearray(conn.api_secret.encode()), msg=hmac_data.encode(), digestmod=hashlib.sha256).hexdigest()
    # set header
    header = {'content-type':
              'application/x-www-form-urlencoded; charset=utf-8',
              'X-API-KEY': conn.api_key,
                   'X-API-NONCE': str(nonce),
                   'X-API-SIGNATURE': hmac_signed }
    return header
    
def APIConnect(conn, method, params, uri):
    """Transform Parameters to URL"""

    url = params_url(params, uri)
    header = set_header(url)
    try:
        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)
        # Handle API Errors
        if HandleAPIErrors(r):
            # get results
            result = r.json()
        else:
            result = {}
    except requests.exceptions.RequestException as e:
        HandleRequestsException(e)
    return result


def showOrderbook(conn, OrderType, **args):
    """Search Orderbook for offers."""
    # Build parameters
    if OrderType == 'buy' or OrderType == 'sell':
        params = {'type': OrderType}
    else:
        print('problem')
    params.update(args)
    return APIConnect(conn, 'GET', params, orderuri)


def createOrder(conn, OrderType, max_amount, price, **args):
    """Create a new Order."""
    # Build parameters
    params = {'type': OrderType, 'max_amount': max_amount, 'price': price}
    params.update(args)
    return APIConnect(conn, 'POST', params, orderuri)


def deleteOrder(conn, order_id):
    """Delete an Order."""
    newuri = orderuri + "/" + order_id
    params = {'order_id': order_id}
    return APIConnect(conn, 'DELETE', params, newuri)


def showMyOrders(conn, **args):
    """Query and Filter own Orders."""
    newuri = orderuri + '/my_own'
    return APIConnect(conn, 'GET', args, newuri)


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


def executeTrade(conn, order_id, OrderType, amount):
    """Buy/Sell on a specific Order."""
    newuri = tradeuri + '/' + order_id
    params = {'order_id': order_id, 'type': OrderType, 'amount': amount}
    return APIConnect(conn, 'POST', params, newuri)


def showMyTrades(conn, **args):
    """Query and Filter on past Trades."""
    return APIConnect(conn, 'GET', args, tradeuri)


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


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


def showOrderbookCompact(conn):
    """Bids and Asks in compact format."""
    params = {}
    return APIConnect(conn, 'GET', params, orderuri + '/compact')


def showPublicTradeHistory(conn, since_tid=None):
    """All successful trades of the las 7 days."""
    if since_tid is not None:
        params = {'since_tid': since_tid}
    else:
        params = {}
    return APIConnect(conn, 'GET', params, tradeuri + '/history')


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


def showAccountLedger(conn, **args):
    """Query on Account statement."""
    return APIConnect(conn, 'GET', args, accounturi + '/ledger')


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			# these two lines enable debugging at httplib level
13			# (requests->urllib3->httplib)
14			# you will see the REQUEST, including HEADERS and DATA, and RESPONSE with
15			# HEADERS but without DATA.
16			# the only thing missing will be the response.body which is not logged.
17			import http.client
18			http.client.HTTPConnection.debuglevel = 1
19
20
21			logging.basicConfig()
22			logging.getLogger().setLevel(logging.DEBUG)
23			requests_log = logging.getLogger("requests.packages.urllib3")
24			requests_log.setLevel(logging.DEBUG)
25			requests_log.propagate = True
26
27			__version__ = '1.4'
28
29
30			class Connection:
31			"""To provide connection credentials to the trading API"""
32			def __init__(self, api_key, api_secret):
33			self.api_key = api_key
34			self.api_secret = api_secret
35
36			# Bitcoin.de API URI
37			apihost = 'https://api.bitcoin.de'
38			apiversion = 'v1'
39			orderuri = apihost + '/' + apiversion + '/' + 'orders'
40			tradeuri = apihost + '/' + apiversion + '/' + 'trades'
41			accounturi = apihost + '/' + apiversion + '/' + 'account'
42			# set initial nonce
43			nonce = int(time.time())
44			# disable unsecure SSL warning
45			requests.packages.urllib3.disable_warnings()
46
47
48			def HandleRequestsException(e):
49			"""Handle Exception from request."""
50			print(e[0][0])
51			print(e[0][1])
52
53
54			def HandleAPIErrors(r):
55			"""To handle Errors from BTCDE API."""
56			if r.status_code != 200 and r.status_code != 201 and r.status_code != 204:
57			reader = codecs.getreader("utf-8")
58			content = json.load(reader(r.raw))
59			errors = content.get('errors')
60			print('Code: ' + str(errors[0]['code']))
61			print('Message: ' + errors[0]['message'])
62			print('With URL: ' + r.url)
63			return False
64			else:
65			return True
66
67
68			def params_url(params, uri):
69			encoded_string = ''
70			if params:
71			for key, value in sorted(params.items()):
72			encoded_string += str(key) + '=' + str(value) + '&'
73			encoded_string = encoded_string[:-1]
74			url = uri + '?' + encoded_string
75			else:
76			url = uri
77			return url
78
79			def set_header(url):
80			global nonce
81			# raise nonce before using
82			nonce += 1
83			if method == 'POST':
84			md5_encoded_query_string = hashlib.md5(encoded_string.encode()).hexdigest()
85			else:
86			md5_encoded_query_string = hashlib.md5(b'').hexdigest()
87			hmac_data = method + '#' + \
88			url + '#' + conn.api_key + \
89			'#' + str(nonce) + '#' + md5_encoded_query_string
90			hmac_signed = hmac.new(bytearray(conn.api_secret.encode()), msg=hmac_data.encode(), digestmod=hashlib.sha256).hexdigest()
91			# set header
92			header = {'content-type':
93			'application/x-www-form-urlencoded; charset=utf-8',
94			'X-API-KEY': conn.api_key,
95			'X-API-NONCE': str(nonce),
96			'X-API-SIGNATURE': hmac_signed }
97			return header
98
99			def APIConnect(conn, method, params, uri):
100			"""Transform Parameters to URL"""
101
102			url = params_url(params, uri)
103			header = set_header(url)
104			try:
105			if method == 'GET':
106			r = requests.get(url, headers=(header),
107			stream=True, verify=False)
108			elif method == 'POST':
109			r = requests.post(url, headers=(header), data=encoded_string,
110			stream=True, verify=False)
111			elif method == 'DELETE':
112			r = requests.delete(url, headers=(header),
113			stream=True, verify=False)
114			# Handle API Errors
115			if HandleAPIErrors(r):
116			# get results
117			result = r.json()
118			else:
119			result = {}
120			except requests.exceptions.RequestException as e:
121			HandleRequestsException(e)
122			return result
123
124
125			def showOrderbook(conn, OrderType, **args):
126			"""Search Orderbook for offers."""
127			# Build parameters
128			if OrderType == 'buy' or OrderType == 'sell':
129			params = {'type': OrderType}
130			else:
131			print('problem')
132			params.update(args)
133			return APIConnect(conn, 'GET', params, orderuri)
134
135
136			def createOrder(conn, OrderType, max_amount, price, **args):
137			"""Create a new Order."""
138			# Build parameters
139			params = {'type': OrderType, 'max_amount': max_amount, 'price': price}
140			params.update(args)
141			return APIConnect(conn, 'POST', params, orderuri)
142
143
144			def deleteOrder(conn, order_id):
145			"""Delete an Order."""
146			newuri = orderuri + "/" + order_id
147			params = {'order_id': order_id}
148			return APIConnect(conn, 'DELETE', params, newuri)
149
150
151			def showMyOrders(conn, **args):
152			"""Query and Filter own Orders."""
153			newuri = orderuri + '/my_own'
154			return APIConnect(conn, 'GET', args, newuri)
155
156
157			def showMyOrderDetails(conn, order_id):
158			"""Details to an own Order."""
159			newuri = orderuri + '/' + order_id
160			params = {'order_id': order_id}
161			return APIConnect(conn, 'GET', params, newuri)
162
163
164			def executeTrade(conn, order_id, OrderType, amount):
165			"""Buy/Sell on a specific Order."""
166			newuri = tradeuri + '/' + order_id
167			params = {'order_id': order_id, 'type': OrderType, 'amount': amount}
168			return APIConnect(conn, 'POST', params, newuri)
169
170
171			def showMyTrades(conn, **args):
172			"""Query and Filter on past Trades."""
173			return APIConnect(conn, 'GET', args, tradeuri)
174
175
176			def showMyTradeDetails(conn, trade_id):
177			"""Details to a specific Trade."""
178			newuri = tradeuri + '/' + trade_id
179			params = {'trade_id': trade_id}
180			return APIConnect(conn, 'GET', params, newuri)
181
182
183			def showAccountInfo(conn):
184			"""Query on Account Infos."""
185			params = {}
186			return APIConnect(conn, 'GET', params, accounturi)
187
188
189			def showOrderbookCompact(conn):
190			"""Bids and Asks in compact format."""
191			params = {}
192			return APIConnect(conn, 'GET', params, orderuri + '/compact')
193
194
195			def showPublicTradeHistory(conn, since_tid=None):
196			"""All successful trades of the las 7 days."""
197			if since_tid is not None:
198			params = {'since_tid': since_tid}
199			else:
200			params = {}
201			return APIConnect(conn, 'GET', params, tradeuri + '/history')
202
203
204			def showRates(conn):
205			"""Query of the average rate last 3 and 12 hours."""
206			newuri = apihost + '/' + apiversion + '/rates'
207			params = {}
208			return APIConnect(conn, 'GET', params, newuri)
209
210
211			def showAccountLedger(conn, **args):
212			"""Query on Account statement."""
213			return APIConnect(conn, 'GET', args, accounturi + '/ledger')
214

peshay / btcde

Push — testcase ( d1c203...8af9d3 )

set_header() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like