Files
@ 124bd1706d79
Branch filter:
Location: NPO-Accounting/experimental-accounting-api/accounting/storage/sql/__init__.py - annotation
124bd1706d79
2.0 KiB
text/x-python
Added SQL storage backend
- Added Storage ABC
- Moved Ledger class to storage/ledgercli.py
- Added SQL requirements to requirements.txt
- Added Storage ABC
- Moved Ledger class to storage/ledgercli.py
- Added SQL requirements to requirements.txt
124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 124bd1706d79 | import logging
import json
from flask.ext.sqlalchemy import SQLAlchemy
from accounting.storage import Storage
from accounting.models import Transaction, Posting, Amount
_log = logging.getLogger(__name__)
db = None
class SQLStorage(Storage):
def __init__(self, app):
global db
self.app = app
db = self.db = SQLAlchemy(app)
from .models import Transaction as SQLTransaction, \
Posting as SQLPosting, Amount as SQLAmount
db.create_all()
self.Transaction = SQLTransaction
self.Posting = SQLPosting
self.Amount = SQLAmount
def get_transactions(self, *args, **kw):
transactions = []
for transaction in self.Transaction.query.all():
dict_transaction = transaction.as_dict()
dict_postings = dict_transaction.pop('postings')
postings = []
for dict_posting in dict_postings:
dict_amount = dict_posting.pop('amount')
posting = Posting(**dict_posting)
posting.amount = Amount(**dict_amount)
postings.append(posting)
dict_transaction.update({'postings': postings})
transactions.append(Transaction(**dict_transaction))
return transactions
def add_transaction(self, transaction):
if transaction.id is None:
transaction.generate_id()
_t = self.Transaction()
_t.uuid = str(transaction.id)
_t.date = transaction.date
_t.payee = transaction.payee
_t.meta = json.dumps(transaction.metadata)
self.db.session.add(_t)
for posting in transaction.postings:
_p = self.Posting()
_p.transaction_uuid = str(transaction.id)
_p.account = posting.account
_p.meta = json.dumps(posting.metadata)
_p.amount = self.Amount(symbol=posting.amount.symbol,
amount=posting.amount.amount)
self.db.session.add(_p)
self.db.session.commit()
|