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

Completed

Push — testcase ( 8af9d3...56ee07 )

by Andreas

created 2017-09-16 22:15 UTC

APIConnect() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	7
Bugs	1	Features	0

Metric	Value
cc	3
c	7
b	1
f	0
dl	0
loc	15
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 send_request(method):
    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(conn, method, params, uri):
    """Transform Parameters to URL"""
    url = params_url(params, uri)
    header = set_header(url)
    try:
        r = send_request(method)
        # 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 send_request(method):
100			if method == 'GET':
101			r = requests.get(url, headers=(header),
102			stream=True, verify=False)
103			elif method == 'POST':
104			r = requests.post(url, headers=(header), data=encoded_string,
105			stream=True, verify=False)
106			elif method == 'DELETE':
107			r = requests.delete(url, headers=(header),
108			stream=True, verify=False)
109			return r
110
111			def APIConnect(conn, method, params, uri):
112			"""Transform Parameters to URL"""
113			url = params_url(params, uri)
114			header = set_header(url)
115			try:
116			r = send_request(method)
117			# Handle API Errors
118			if HandleAPIErrors(r):
119			# get results
120			result = r.json()
121			else:
122			result = {}
123			except requests.exceptions.RequestException as e:
124			HandleRequestsException(e)
125			return result
126
127
128			def showOrderbook(conn, OrderType, **args):
129			"""Search Orderbook for offers."""
130			# Build parameters
131			if OrderType == 'buy' or OrderType == 'sell':
132			params = {'type': OrderType}
133			else:
134			print('problem')
135			params.update(args)
136			return APIConnect(conn, 'GET', params, orderuri)
137
138
139			def createOrder(conn, OrderType, max_amount, price, **args):
140			"""Create a new Order."""
141			# Build parameters
142			params = {'type': OrderType, 'max_amount': max_amount, 'price': price}
143			params.update(args)
144			return APIConnect(conn, 'POST', params, orderuri)
145
146
147			def deleteOrder(conn, order_id):
148			"""Delete an Order."""
149			newuri = orderuri + "/" + order_id
150			params = {'order_id': order_id}
151			return APIConnect(conn, 'DELETE', params, newuri)
152
153
154			def showMyOrders(conn, **args):
155			"""Query and Filter own Orders."""
156			newuri = orderuri + '/my_own'
157			return APIConnect(conn, 'GET', args, newuri)
158
159
160			def showMyOrderDetails(conn, order_id):
161			"""Details to an own Order."""
162			newuri = orderuri + '/' + order_id
163			params = {'order_id': order_id}
164			return APIConnect(conn, 'GET', params, newuri)
165
166
167			def executeTrade(conn, order_id, OrderType, amount):
168			"""Buy/Sell on a specific Order."""
169			newuri = tradeuri + '/' + order_id
170			params = {'order_id': order_id, 'type': OrderType, 'amount': amount}
171			return APIConnect(conn, 'POST', params, newuri)
172
173
174			def showMyTrades(conn, **args):
175			"""Query and Filter on past Trades."""
176			return APIConnect(conn, 'GET', args, tradeuri)
177
178
179			def showMyTradeDetails(conn, trade_id):
180			"""Details to a specific Trade."""
181			newuri = tradeuri + '/' + trade_id
182			params = {'trade_id': trade_id}
183			return APIConnect(conn, 'GET', params, newuri)
184
185
186			def showAccountInfo(conn):
187			"""Query on Account Infos."""
188			params = {}
189			return APIConnect(conn, 'GET', params, accounturi)
190
191
192			def showOrderbookCompact(conn):
193			"""Bids and Asks in compact format."""
194			params = {}
195			return APIConnect(conn, 'GET', params, orderuri + '/compact')
196
197
198			def showPublicTradeHistory(conn, since_tid=None):
199			"""All successful trades of the las 7 days."""
200			if since_tid is not None:
201			params = {'since_tid': since_tid}
202			else:
203			params = {}
204			return APIConnect(conn, 'GET', params, tradeuri + '/history')
205
206
207			def showRates(conn):
208			"""Query of the average rate last 3 and 12 hours."""
209			newuri = apihost + '/' + apiversion + '/rates'
210			params = {}
211			return APIConnect(conn, 'GET', params, newuri)
212
213
214			def showAccountLedger(conn, **args):
215			"""Query on Account statement."""
216			return APIConnect(conn, 'GET', args, accounturi + '/ledger')
217

peshay / btcde

Push — testcase ( 8af9d3...56ee07 )

APIConnect() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like