k4ever/client-barcode/freitagslib/network.py

153 lines
3.9 KiB
Python
Raw Normal View History

#! /usr/bin/env python
# -*- coding: utf-8 -*-
#
# Copyright (C) 2011 Sebastian Pipping <sebastian@pipping.org>
# Licensed under GPL v3 or later
from __future__ import print_function
import urllib
import urllib2
import simplejson as json
from decimal import Decimal
import os
import base64
import copy
if __name__ == '__main__':
import sys
sys.path.append('.')
from freitagslib.item import Item
from freitagslib.tools import require_type
DEPOSIT_CASH = 1
DEPOSIT_BANK = 2
2011-10-14 18:57:24 +02:00
_BASE_URL = 'https://k4ever.freitagsrunde.org:443/api2/'
_auth = base64.encodestring('%s:%s' % (os.environ['BARCODE_PLUGIN_USER'], os.environ['BARCODE_PLUGIN_PASS'])).rstrip()
_headers = {'Authorization':'Basic %s' % _auth}
ITEM_ONLY, DEPOSIT_ONLY, ITEM_AND_DEPOSIT = range(3)
if bool(os.environ.get('BARCODE_PLUGIN_DEBUG', False)):
_h = urllib2.HTTPHandler(debuglevel=1)
_opener = urllib2.build_opener(_h)
_urlopen = _opener.open
else:
_urlopen = urllib2.urlopen
def _request(api_rel_url, data, method, headers=None):
url = _BASE_URL + api_rel_url
if isinstance(data, dict):
data = urllib.urlencode(data)
# Add default headers
if headers is None:
final_headers = _headers
else:
final_headers = copy.copy(_headers)
final_headers.update(headers)
if method == 'GET':
url = '%s?%s' % (url, data)
req = urllib2.Request(url, headers=final_headers)
elif method == 'POST':
req = urllib2.Request(url, data=data, headers=final_headers)
else:
raise ValueError('Unsupported method "%s"' % method)
response = _urlopen(req)
return response.read()
def get_user_name_from_auth_blob(authblob):
content = _request('auth/user/', {'authblob':authblob}, method='GET')
d = json.loads(content)
return d['username']
def buy_item(item_id, what_about_it, user_name):
require_type(int, item_id)
assert(what_about_it in (ITEM_ONLY, DEPOSIT_ONLY, ITEM_AND_DEPOSIT))
content = _request('buyable/item/' + urllib.quote(str(item_id)),
{'user':user_name, 'deposit':what_about_it, 'amount':1}, method='POST')
if content != 'Created':
raise ValueError('Server says "%s"' % content)
def bulk_buy(buycommands, user_name):
item_ids = tuple(e.item_id() for e in buycommands if e.includes_commodity())
deposit_ids = tuple(e.item_id() for e in buycommands if e.includes_deposit())
if not item_ids and not deposit_ids:
return
request_dict = {
'user':user_name,
'items':item_ids,
'deposits':deposit_ids
}
data = json.dumps(request_dict, sort_keys=True, indent=4 * ' ')
headers = {
'Content-Type':'application/json',
'Content-Length':len(data)
}
content = _request('buyable/item/bulkbuy/', data, 'POST', headers=headers)
if content != 'Created':
raise ValueError('Server says "%s"' % content)
def get_item(barcode):
content = _request('buyable/item/',
{'barcode':barcode}, method='GET')
d = json.loads(content)
return Item(d)
def get_balance(user_name):
content = _request('account/balance/',
{'user':user_name}, method='GET')
d = json.loads(content)
return Decimal(d['balance'])
def deposit(amount, transaction_type, user_name):
require_type(Decimal, amount)
require_type(int, transaction_type)
content = _request('account/transactions/transact/',
{'user':user_name, 'amount':amount, 'type':transaction_type}, method='POST')
if content != 'OK':
raise ValueError('Server says "%s"' % content)
if __name__ == '__main__':
user_name = get_user_name_from_auth_blob(os.environ['BARCODE_PLUGIN_TEST_AUTH_BLOB'])
print('User name: ' + user_name)
buy_item(1, True, user_name)
balance = get_balance(user_name)
print('Balance is %f, type is "%s"' % (balance, type(balance).__name__))
deposit(Decimal('0.01'), DEPOSIT_CASH, user_name)
deposit(Decimal('0.01'), DEPOSIT_BANK, user_name)
def show(item):
print('Item "%s" is %f Euro each' % (item.name, item.price))
item5 = get_item(5)
item6 = get_item(6)
for item in (item5, item6):
show(item)
from freitagslib.commands import BuyCommand
buycommands = (BuyCommand(e) for e in (item5, item6))
bulk_buy(buycommands, user_name)