Files
@ f0a5db5a2907
Branch filter:
Location: NPO-Accounting/experimental-accounting-api/accounting/client.py
f0a5db5a2907
4.2 KiB
text/x-python
Refactored accounting.client
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 | import sys
import argparse
import json
import logging
from datetime import datetime
from decimal import Decimal
import requests
from accounting.models import Transaction, Posting, Amount
from accounting.transport import AccountingDecoder, AccountingEncoder
_log = logging.getLogger(__name__)
class Client:
def __init__(self, host=None, json_encoder=None,
json_decoder=None):
self.host = host or 'http://localhost:5000'
self.json_encoder = json_encoder or AccountingEncoder
self.json_decoder = json_decoder or AccountingDecoder
def get_balance(self):
balance = self.get('/balance')
return balance['balance_report']
def get(self, path):
response = requests.get(self.host + path)
return self._decode_response(response)
def _decode_response(self, response):
response_data = response.json(cls=self.json_decoder)
_log.debug('response_data: %s', response_data)
return response_data
def post(self, path, payload, **kw):
kw.update({'headers': {'Content-Type': 'application/json'}})
kw.update({'data': json.dumps(payload, cls=self.json_encoder)})
return self._decode_response(requests.post(self.host + path, **kw))
def simple_transaction(self, from_acc, to_acc, amount):
t = Transaction(
date=datetime.today(),
payee='PayPal donation',
postings=[
Posting(account=from_acc,
amount=Amount(symbol='$', amount=-amount)),
Posting(account=to_acc,
amount=Amount(symbol='$', amount=amount))
]
)
return self.post('/transaction', {'transactions': [t]})
def get_register(self):
register = self.get('/register')
return register['register_report']
def print_transactions(transactions):
for transaction in transactions:
print('{date} {t.payee:.<69}'.format(
date=transaction.date.strftime('%Y-%m-%d'),
t=transaction))
for posting in transaction.postings:
print(' ' + posting.account +
' ' * (80 - len(posting.account) - len(posting.amount.symbol) -
len(str(posting.amount.amount)) - 1 - 1) +
posting.amount.symbol + ' ' + str(posting.amount.amount))
def print_balance_accounts(accounts, level=0):
for account in accounts:
print(' ' * level + ' + {account.name}'.format(account=account) +
' ' + '-' * (80 - len(str(account.name)) - level))
for amount in account.amounts:
print(' ' * level + ' {amount.symbol} {amount.amount}'.format(
amount=amount))
print_balance_accounts(account.accounts, level+1)
def main(argv=None, prog=None):
global HOST
if argv is None:
prog = sys.argv[0]
argv = sys.argv[1:]
parser = argparse.ArgumentParser(prog=prog)
actions = parser.add_subparsers(title='Actions', dest='action')
insert = actions.add_parser('insert',
aliases=['in'])
insert.add_argument('from_account')
insert.add_argument('to_account')
insert.add_argument('amount', type=Decimal)
balance = actions.add_parser('balance', aliases=['bal'])
register = actions.add_parser('register', aliases=['reg'])
parser.add_argument('-v', '--verbosity',
default='WARNING',
help=('Filter logging output. Possible values:' +
' CRITICAL, ERROR, WARNING, INFO, DEBUG'))
parser.add_argument('--host', default='http://localhost:5000')
args = parser.parse_args(argv)
logging.basicConfig(level=getattr(logging, args.verbosity))
client = Client(args.host)
if args.action in ['insert', 'in']:
print(client.simple_transaction(args.from_account, args.to_account,
args.amount))
elif args.action in ['balance', 'bal']:
print_balance_accounts(client.get_balance())
elif args.action in ['register', 'reg']:
print_transactions(client.get_register())
else:
parser.print_help()
if __name__ == '__main__':
sys.exit(main())
|