Source code for personalcapital.query

import json
import structlog
from datetime import datetime, timedelta

from personalcapital import APIConnector
from personalcapital.database import Database

log = structlog.getLogger()


def get_transactions(start_date, end_date):
    with APIConnector.connect() as session:
        transactions_response = session.fetch('/transaction/getUserTransactions', {
            'sort_cols': 'transactionTime',
            'sort_rev': 'true',
            'page': '0',
            'rows_per_page': '100',
            'startDate': start_date,
            'endDate': end_date,
            'component': 'DATAGRID'
        }).json()
    transaction_data = transactions_response['spData']
    log.msg('Retrieved {} transactions between {} and {}.'.format(
        len(transaction_data['transactions']), transaction_data['startDate'], transaction_data['endDate']))
    return transaction_data['transactions']


def get_accounts(session):
    accounts_response = session.fetch('/newaccount/getAccounts')
    with open('accounts_response.json', 'w') as fp:
        json.dump(accounts_response.json(), fp, indent=4)
    accounts = accounts_response.json()['spData']
    log.msg('Networth: {0}'.format(accounts['networth']))
    return accounts


def fetch_and_dump(session, endpoint, fn):
    response = session.fetch(endpoint)
    with open(fn, 'w') as fp:
        json.dump(response.json(), fp, indent=4)


def update_transaction_categories():
    with APIConnector.connect() as session, Database() as db:
        categories = session.fetch('/transactioncategory/getCategories').json()['spData']
        db.banking.transaction_categories.insert_many(categories)


def get_transaction_categories():
    with Database() as db:
        categories = db.banking.transaction_categories.find({})
        return {
            category["transactionCategoryId"]: category
            for category in categories
        }