Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -7,3 +7,4 @@ build/
dist/
*.egg-info/
.eggs/
.coverage
6 changes: 6 additions & 0 deletions .travis.yml
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,13 @@ python:

install:
- easy_install -U setuptools
- pip uninstall -y six
- pip install six>=1.11.0
- python setup.py -q install
- pip install codecov

script:
- python setup.py test

after_success:
- codecov
73 changes: 73 additions & 0 deletions examples/readonly.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,73 @@
import os
import time

from luno_python.client import Client


if __name__ == '__main__':
c = Client(api_key_id=os.getenv('LUNO_API_KEY_ID'),
api_key_secret=os.getenv('LUNO_API_KEY_SECRET'))

res = c.get_tickers()
print(res)
time.sleep(0.5)

res = c.get_ticker(pair='XBTZAR')
print(res)
time.sleep(0.5)

res = c.get_order_book(pair='XBTZAR')
print(res)
time.sleep(0.5)

since = int(time.time()*1000)-24*60*60*1000
res = c.list_trades(pair='XBTZAR', since=since)
print(res)
time.sleep(0.5)

res = c.get_balances()
print(res)
time.sleep(0.5)

aid = ''
if res['balance']:
aid = res['balance'][0]['account_id']

if aid:
res = c.list_transactions(id=aid, min_row=1, max_row=10)
print(res)
time.sleep(0.5)

if aid:
res = c.list_pending_transactions(id=aid)
print(res)
time.sleep(0.5)

res = c.list_orders()
print(res)
time.sleep(0.5)

res = c.list_user_trades(pair='XBTZAR')
print(res)
time.sleep(0.5)

res = c.get_fee_info()
print(res)
time.sleep(0.5)

res = c.get_funding_address(asset='XBT')
print(res)
time.sleep(0.5)

res = c.list_withdrawals()
print(res)
time.sleep(0.5)

wid = ''
if res['withdrawals']:
wid = res['withdrawals'][0]['id']

if wid:
res = c.get_withdrawal(id=wid)
print(res)
time.sleep(0.5)
2 changes: 1 addition & 1 deletion luno_python/__init__.py
Original file line number Diff line number Diff line change
@@ -1 +1 @@
VERSION = '0.0.1'
VERSION = '0.0.1-alpha'
106 changes: 106 additions & 0 deletions luno_python/base_client.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,106 @@
import json
import platform
import requests
import six

try:
from json.decoder import JSONDecodeError
except ImportError:
JSONDecodeError = ValueError

from . import VERSION
from .error import APIError

DEFAULT_BASE_URL = 'https://api.mybitx.com'
DEFAULT_TIMEOUT = 10
PYTHON_VERSION = platform.python_version()
SYSTEM = platform.system()
ARCH = platform.machine()


class BaseClient:
def __init__(self, base_url='', timeout=0,
api_key_id='', api_key_secret=''):
"""
:type base_url: str
:type timeout: float
:type api_key_id: str
:type api_key_secret: str
"""
self.set_auth(api_key_id, api_key_secret)
self.set_base_url(base_url)
self.set_timeout(timeout)

self.session = requests.Session()

def set_auth(self, api_key_id, api_key_secret):
"""Provides the client with an API key and secret.

:type api_key_id: str
:type api_key_secret: str
"""
self.api_key_id = api_key_id
self.api_key_secret = api_key_secret

def set_base_url(self, base_url):
"""Overrides the default base URL. For internal use.

:type base_url: str
"""
if base_url == '':
base_url = DEFAULT_BASE_URL
self.base_url = base_url.rstrip('/')

def set_timeout(self, timeout):
"""Sets the timeout, in seconds, for requests made by the client.

:type timeout: float
"""
if timeout == 0:
timeout = DEFAULT_TIMEOUT
self.timeout = timeout

def do(self, method, path, req=None, auth=False):
"""Performs an API request and returns the response.

TODO: Handle 429s

:type method: str
:type path: str
:type req: object
:type auth: bool
"""
try:
params = json.loads(json.dumps(req))
except Exception:
params = None
headers = {'User-Agent': self.make_user_agent()}
args = dict(timeout=self.timeout, params=params, headers=headers)
if auth:
args['auth'] = (self.api_key_id, self.api_key_secret)
url = self.make_url(path, params)
res = self.session.request(method, url, **args)
try:
e = res.json()
if 'error' in e and 'error_code' in e:
raise APIError(e['error_code'], e['error'])
return e
except JSONDecodeError:
raise Exception('luno: unknown API error (%s)' % res.status_code)

def make_url(self, path, params):
"""
:type path: str
:rtype: str
"""
if params:
for k, v in six.iteritems(params):
path = path.replace('{' + k + '}', str(v))
return self.base_url + '/' + path.lstrip('/')

def make_user_agent(self):
"""
:rtype: str
"""
return "LunoPythonSDK/%s python/%s %s %s" % \
(VERSION, PYTHON_VERSION, SYSTEM, ARCH)
Loading