~trufas/ledgeroni

9c106be6e8bf0bca1a4358a6504e77917e920ec3 — Rafael Castillo 1 year, 9 months ago e03087c
Refactor queries to allow to filter by payee string
M conftest.py => conftest.py +64 -0
@@ 0,0 1,64 @@
from datetime import datetime
import pytest
import arrow
from ledgeroni.types import Transaction, Posting, Commodity

USD = Commodity(is_prefix=True, name='$')

def example_transactions():
    TRANS_1 = Transaction(
        date=arrow.get(datetime(2013, 2, 20)),
        description='Purchased reddit gold for the year')
    TRANS_1.add_posting(Posting(
        account=('Asset', 'Bitcoin Wallet'),
        amounts={None: -10.0}))
    TRANS_1.add_posting(Posting(
        account=('Expense', 'Web Services', 'Reddit'),
        amounts=None))


    TRANS_2 = Transaction(
        date=arrow.get(datetime(2013, 2, 20)),
        description='Spent some cool cash')
    TRANS_2.add_posting(Posting(
        account=('Bank', 'Paypal'),
        amounts={USD: -10.0}))
    TRANS_2.add_posting(Posting(
        account=('Cool', 'Thing'),
        amounts=None))


    TRANS_3 = Transaction(
        date=arrow.get(datetime(2013, 2, 20)),
        description='Purchased whatever digg sells for the year')
    TRANS_3.add_posting(Posting(
        account=('Asset', 'Bitcoin Wallet'),
        amounts={None: -10.0}))
    TRANS_3.add_posting(Posting(
        account=('Expense', 'Web Services', 'Digg'),
        amounts=None))


    TRANS_4 = Transaction(
        date=arrow.get(datetime(2013, 2, 20)),
        description='I owe Joe a favor')
    TRANS_4.add_posting(Posting(
        account=('Payable', 'Joe', 'Favor'),
        amounts={USD: -10.0}))
    TRANS_4.add_posting(Posting(
        account=('Expense', 'Favor'),
        amounts=None))


    TRANS_5 = Transaction(
        date=arrow.get(datetime(2013, 2, 20)),
        description='Traded some cards')
    TRANS_5.add_posting(Posting(
        account=('Asset', 'Pokemon Cards'),
        amounts={None: -10.0}))
    TRANS_5.add_posting(Posting(
        account=('Asset', 'MTG Cards'),
        amounts=None))

    return [TRANS_1, TRANS_2, TRANS_3, TRANS_4, TRANS_5]


M ledgeroni/aggregate.py => ledgeroni/aggregate.py +5 -4
@@ 6,7 6,7 @@ from dataclasses import dataclass, field
from typing import Dict, Tuple, Iterator
from collections import defaultdict, deque
from ledgeroni.types import Commodity, Transaction
from ledgeroni.query import Query
from ledgeroni.query import Query, MATCH_ALL
from ledgeroni.journal import Journal




@@ 17,6 17,7 @@ class AccountAggregate:
    subaccounts: Dict = field(
        default_factory=lambda: defaultdict(AccountAggregate))
    aggregates: Dict = field(default_factory=lambda: defaultdict(Fraction))
    query: Query = MATCH_ALL

    def add_commodity(self, account: Tuple[str], amount: Fraction,
                      commodity: Commodity):


@@ 32,10 33,10 @@ class AccountAggregate:
            self.own_balances[commodity] += amount
            self.aggregates[commodity] += amount

    def add_transaction(self, transaction: Transaction, query: Query = None):
    def add_transaction(self, transaction: Transaction):
        "Adds a given transactions postings to the aggregate"
        transaction = transaction.calc_totals()
        for posting in transaction.postings_matching(query):
        for posting in self.query.postings_matching(transaction):
            if posting.amounts is None:
                continue
            for commodity, amount in posting.amounts.items():


@@ 44,7 45,7 @@ class AccountAggregate:
    def add_from_journal(self, journal: Journal):
        "Adds all transactions in a journal to the aggregate"
        for transaction in journal.transactions:
            self.add_transaction(transaction, journal.query)
            self.add_transaction(transaction)

    def iter_aggregates(self) -> Iterator[Tuple[int, str, Dict]]:
        """Iterates through all aggregates in a depth first search, yielding

M ledgeroni/commands/print.py => ledgeroni/commands/print.py +3 -0
@@ 6,6 6,7 @@ import click
from ledgeroni.journal import Journal
from ledgeroni import expression


@click.command()
@click.argument('filter_strs', nargs=-1)
@click.pass_context


@@ 24,6 25,8 @@ def print_transactions(ctx, filter_strs):
    for filename in ctx.obj.get('LEDGER_FILES', []):
        journal.add_from_file(filename)

    print(journal)

    errors = journal.verify_transaction_balances()
    if errors:
        for error in errors:

M ledgeroni/commands/register.py => ledgeroni/commands/register.py +1 -0
@@ 2,6 2,7 @@
register.py: Defines the `register subcommand`
"""
import itertools
from os import sys
import click
from colorama import Fore, Style


M ledgeroni/journal.py => ledgeroni/journal.py +1 -1
@@ 25,7 25,7 @@ class Journal:

    def add_transaction(self, transaction: Transaction):
        "Adds and indexes a transaction."
        if not self.query or transaction.matches_query(self.query):
        if not self.query or self.query.execute(transaction):
            self.transactions.append(transaction)
            self.accounts.update(p.account for p in transaction.postings)
            self.commodities.update(c for p in transaction.postings

M ledgeroni/parser.py => ledgeroni/parser.py +2 -2
@@ 19,7 19,7 @@ import os
import arrow

from ledgeroni.types import (Transaction, Posting, Commodity, Price,
                             IgnoreSymbol, DefaultCommodity)
                             IgnoreSymbol, DefaultCommodity, INTEGER_COMMODITY)


def load_lines(filename: str) -> Iterator[str]:


@@ 103,7 103,7 @@ def read_amount(amtstr: str) -> Tuple[Fraction, Commodity]:

    multi = -1 if negation else 1
    amount = Fraction(amount)
    commodity = None
    commodity = INTEGER_COMMODITY
    if prefix:
        commodity = Commodity(is_prefix=True, name=prefix)
    if suffix:

M ledgeroni/query.py => ledgeroni/query.py +35 -11
@@ 3,22 3,30 @@ query.py Abstraction for filtering account names
"""
import re
from dataclasses import dataclass
from typing import Tuple, Iterable
from typing import Tuple, Iterable, Set, Iterator

from ledgeroni.types import Transaction, Posting

@dataclass(frozen=True)
class Query:
    "Base abstract query class"

    def execute(self, trans: Transaction) -> Set[Posting]:
        raise NotImplementedError

    def postings_matching(self, trans: Transaction) -> Iterator[Posting]:
        return (trans.postings[i] for i in self.execute(trans))


@dataclass(frozen=True)
class RegexQuery(Query):
    "Query to match a datum to a regular expression"
    regex: re.Pattern

    def execute(self, data: str) -> bool:
    def execute(self, trans: Transaction) -> Set[Posting]:
        "Runs the query on `data`"
        return self.regex.search(data) is not None
        return {i for i, posting in enumerate(trans.postings)
                if self.regex.search(posting.account_name) is not None}


@dataclass(frozen=True)


@@ 26,9 34,9 @@ class Or(Query):
    "Query to combine multiple subqueries with an OR operation"
    queries: Tuple[Query]

    def execute(self, data: str) -> bool:
    def execute(self, trans: Transaction) -> Set[Posting]:
        "Runs the query on data"
        return any(q.execute(data) for q in self.queries)
        return set.union(*(q.execute(trans) for q in self.queries))


@dataclass(frozen=True)


@@ 36,9 44,9 @@ class And(Query):
    "Query to combine multiple subqueries with an AND operation"
    queries: Tuple[Query]

    def execute(self, data: str) -> bool:
    def execute(self, trans: Transaction) -> Set[Posting]:
        "Runs the query on data"
        return all(q.execute(data) for q in self.queries)
        return set.intersection(*(q.execute(trans) for q in self.queries))


@dataclass(frozen=True)


@@ 46,18 54,34 @@ class Not(Query):
    "Query that wraps a query and returns the opposite result"
    query: Query

    def execute(self, data: str) -> bool:
    def execute(self, trans: Transaction) -> Set[Posting]:
        "Runs the query on data"
        return not self.query.execute(data)
        return set(range(len(trans.postings))) - self.query.execute(trans)


@dataclass(frozen=True)
class PayeeQuery(Query):
    """
    A query that matches a regex against the payee instead of the account names
    """
    query: RegexQuery

    def execute(self, data: str) -> bool:
    def execute(self, trans: Transaction) -> Set[Posting]:
        "Runs the query on data"
        return self.query.execute(data)
        regex = self.query.regex
        if regex.search(trans.description) is None:
            return set([])
        return set(range(len(trans.postings)))


@dataclass(frozen=True)
class AlwaysTrueQuery(Query):
    "A query that always returns all elements"
    def execute(self, trans: Transaction):
        return set(range(len(trans.postings)))


MATCH_ALL = AlwaysTrueQuery()


def build_simple_or_query(strs: Iterable[str]) -> Query:

M ledgeroni/types.py => ledgeroni/types.py +18 -18
@@ 8,7 8,6 @@ from typing import List, Tuple, Dict, Iterator
import copy
from fractions import Fraction
from arrow.arrow import Arrow
from ledgeroni.query import Query


@dataclass(frozen=True)


@@ 30,6 29,22 @@ class Commodity:
        return amt + self.name


class IntegerCommodity():
    name: str = ''
    is_prefix: bool = False

    @property
    def symbol(self) -> str:
        return ''

    def format_amount(self, amt: Fraction) -> str:
        "Formats the amount passed in with the commodity symbol"
        return '{:.2f}'.format(float(amt))


INTEGER_COMMODITY = IntegerCommodity()


@dataclass(frozen=True)
class Posting:
    "Represents a movement in a specific account"


@@ 41,10 56,6 @@ class Posting:
        "Full account name as shown in the original posting"
        return ':'.join(self.account)

    def matches_query(self, query: Query) -> bool:
        "Returns whether the posting matches the given query"
        return query.execute(self.account_name)

    def as_journal_format(self) -> str:
        "Returns the posting formatted in a ledger journal format"
        amount_str = ''


@@ 65,18 76,6 @@ class Transaction:
        "Adds a posting to this transaction"
        self.postings.append(posting)

    def matches_query(self, query: Query) -> bool:
        """
        Returns a boolean that indicates if any of the postings match the query
        """
        return any(p.matches_query(query) for p in self.postings)

    def postings_matching(self, query) -> Iterator:
        "Returns an iterator of postings that match query"
        if query is None:
            return self.postings
        return (p for p in self.postings if p.matches_query(query))

    @property
    def date_str(self):
        "Date of the transaction in journal format"


@@ 107,7 106,6 @@ class Transaction:

        return all(a == 0 for a in totals.values())


    def calc_totals(self) -> Transaction:
        """
        Returns a new transaction where postings with implicit amounts are


@@ 153,3 151,5 @@ class IgnoreSymbol:
class DefaultCommodity:
    "Specifies a commodity to be used as default"
    commodity: Commodity



M tests/test_aggregate.py => tests/test_aggregate.py +5 -3
@@ 1,8 1,10 @@
from datetime import datetime
import re
from fractions import Fraction
import arrow
from ledgeroni.aggregate import AccountAggregate
from ledgeroni.types import Commodity, Transaction, Posting
from ledgeroni.query import RegexQuery, Not
from ledgeroni.journal import Journal

USD = Commodity(is_prefix=True, name='$')


@@ 47,8 49,8 @@ def test_add_transaction():
    assert agg.aggregates == {USD: 0, BTC: 0}


def test_add_transaction():
    agg = AccountAggregate()
def test_query_filtering():
    agg = AccountAggregate(query=Not(RegexQuery(re.compile('Asset'))))

    trans = Transaction(
        date=arrow.get(datetime(2013, 2, 20)),


@@ 74,7 76,7 @@ def test_add_transaction():

    agg.add_transaction(trans)

    assert agg.aggregates == {USD: 0, BTC: 0}
    assert agg.aggregates == {USD: 0, BTC: Fraction(10)}


def test_add_from_journal():

M tests/test_parser.py => tests/test_parser.py +3 -3
@@ 5,7 5,7 @@ import arrow

from ledgeroni import parser
from ledgeroni.types import (Transaction, Posting, Commodity, Price,
                                   IgnoreSymbol, DefaultCommodity)
                             IgnoreSymbol, DefaultCommodity, INTEGER_COMMODITY)

USD = Commodity(is_prefix=True, name='$')
AU = Commodity(is_prefix=False, name='AU')


@@ 32,7 32,7 @@ class TestParserMethods(unittest.TestCase):
        posting_line = '	Asset:Bitcoin Wallet		-10'
        posting = Posting(
            account=('Asset', 'Bitcoin Wallet'),
            amounts={None: -10.0})
            amounts={INTEGER_COMMODITY: -10.0})
        result = parser.read_posting_line(posting_line)
        self.assertEqual(result, posting)



@@ 61,7 61,7 @@ class TestParserMethods(unittest.TestCase):
            description='Purchased reddit gold for the year')
        trans.add_posting(Posting(
            account=('Asset', 'Bitcoin Wallet'),
            amounts={None: -10.0}))
            amounts={INTEGER_COMMODITY: -10.0}))
        trans.add_posting(Posting(
            account=('Expense', 'Web Services', 'Reddit'),
            amounts=None))

M tests/test_query.py => tests/test_query.py +80 -17
@@ 1,36 1,99 @@
from ledgeroni.query import And, Or, Not, RegexQuery
from datetime import datetime
from fractions import Fraction
import arrow
from ledgeroni.query import And, Or, Not, RegexQuery, PayeeQuery
from ledgeroni.types import Transaction, Posting, Commodity
import re

USD = Commodity(is_prefix=True, name='$')

TRANS_1 = Transaction(
    date=arrow.get(datetime(2013, 2, 20)),
    description='Purchased reddit gold for the year')
TRANS_1.add_posting(Posting(
    account=('Asset', 'Bitcoin Wallet'),
    amounts={None: -10.0}))
TRANS_1.add_posting(Posting(
    account=('Expense', 'Web Services', 'Reddit'),
    amounts=None))


TRANS_2 = Transaction(
    date=arrow.get(datetime(2013, 2, 20)),
    description='Spent some cool cash')
TRANS_2.add_posting(Posting(
    account=('Bank', 'Paypal'),
    amounts={USD: -10.0}))
TRANS_2.add_posting(Posting(
    account=('Cool', 'Thing'),
    amounts=None))


TRANS_3 = Transaction(
    date=arrow.get(datetime(2013, 2, 20)),
    description='Purchased whatever digg sells for the year')
TRANS_3.add_posting(Posting(
    account=('Asset', 'Bitcoin Wallet'),
    amounts={None: -10.0}))
TRANS_3.add_posting(Posting(
    account=('Expense', 'Web Services', 'Digg'),
    amounts=None))


TRANS_4 = Transaction(
    date=arrow.get(datetime(2013, 2, 20)),
    description='I owe Joe a favor')
TRANS_4.add_posting(Posting(
    account=('Payable', 'Joe', 'Favor'),
    amounts={USD: -10.0}))
TRANS_4.add_posting(Posting(
    account=('Expense', 'Favor'),
    amounts=None))


TRANS_5 = Transaction(
    date=arrow.get(datetime(2013, 2, 20)),
    description='Traded some cards')
TRANS_5.add_posting(Posting(
    account=('Asset', 'Pokemon Cards'),
    amounts={None: -10.0}))
TRANS_5.add_posting(Posting(
    account=('Asset', 'MTG Cards'),
    amounts=None))


def test_regex_query():
    q = RegexQuery(re.compile('Expense'))

    assert q.execute('Expense:Web Services:Reddit') == True
    assert q.execute('Something:Expense:Web Services:Reddit') == True
    print(q.execute(TRANS_1))
    assert q.execute(TRANS_1)
    assert not q.execute(TRANS_2)


def test_and():
    q = And((RegexQuery(re.compile('Expense')),
             RegexQuery(re.compile('Reddit'))))

    assert q.execute('Expense:Web Services:Reddit') == True
    assert q.execute('Expense:Food') == False
    assert q.execute('We:Did:It:Reddit') == False
    assert q.execute('Bank:Paypal') == False
    assert q.execute(TRANS_1)
    assert not q.execute(TRANS_3)


def test_or():
    q = Or((RegexQuery(re.compile('Expense')),
             RegexQuery(re.compile('Reddit'))))
    q = Or((RegexQuery(re.compile('Digg')),
            RegexQuery(re.compile('Reddit'))))

    assert q.execute('Expense:Web Services:Reddit') == True
    assert q.execute('Expense:Food') == True
    assert q.execute('We:Did:It:Reddit') == True
    assert q.execute('Bank:Paypal') == False
    assert q.execute(TRANS_1)
    assert not q.execute(TRANS_2)
    assert q.execute(TRANS_3)


def test_not():
    q = Not(RegexQuery(re.compile('Expense')))
    q = Not(RegexQuery(re.compile('Asset')))

    assert q.execute(TRANS_1)
    assert not q.execute(TRANS_5)

def test_payee():
    q = PayeeQuery(RegexQuery(re.compile('reddit')))

    assert q.execute('Expense:Web Services:Reddit') == False
    assert q.execute('Bank:Paypal') == True
    assert q.execute(TRANS_1)
    assert not q.execute(TRANS_2)