summary history files

web/penny/resources/data/transactions/controllers.py
from penny import models
from flask import (
    current_app as app,
    Blueprint,
    g,
    jsonify,
    request,
    make_response,
    abort,
)
from flask_security.decorators import auth_required
from sqlalchemy.exc import NoResultFound
from sqlalchemy import or_
from sqlalchemy.sql import func
from datetime import datetime

data_transactions = Blueprint(
    "data_transactions", __name__, url_prefix="/data/transactions"
)


@data_transactions.route("/")
@auth_required()
def transactions():
    """Return data on all transactions."""

    data = {"rows": []}

    search = request.args.get("search")
    offset = request.args.get("offset", 1)
    limit = request.args.get("limit", 25)

    total = models.db.session.query(func.count(models.Transaction.id)).filter(
        models.Transaction.user == g.user, models.Transaction.is_deleted == 0
    )

    transactions = (
        models.db.session.query(models.Transaction)
        .filter(models.Transaction.user == g.user, models.Transaction.is_deleted == 0)
        .order_by(models.Transaction.date.desc())
    )

    if search:
        total = total.filter(models.Transaction.memo.like("%{0}%".format(search)))
        transactions = transactions.filter(
            or_(
                models.Transaction.memo.like("%{0}%".format(search)),
                models.Transaction.credit.like("{0}%".format(search)),
                models.Transaction.debit.like("-{0}%".format(search)),
            )
        )

    data["total"] = total.one()[0]

    for transaction in transactions.offset(offset).limit(limit).all():
        data["rows"].append(transaction.dump())

    app.logger.info(data)
    return jsonify(data)


@data_transactions.route(
    "/account", defaults={"start_date": None, "end_date": None, "id": None}
)
@data_transactions.route(
    "/account/<int:id>", defaults={"start_date": None, "end_date": None}
)
@data_transactions.route("/account/<int:id>/<string:start_date>/<string:end_date>")
@auth_required()
def account(id, start_date, end_date):
    """Return data on all transactions for an account."""

    data = {"rows": []}

    search = request.args.get("search")
    offset = request.args.get("offset", 1)
    limit = request.args.get("limit", 25)

    if start_date and end_date:
        start_date = datetime.strptime(start_date, "%Y%m%d")
        end_date = datetime.strptime(end_date, "%Y%m%d")

    if start_date and end_date:
        total = models.db.session.query(func.count(models.Transaction.id)).filter(
            models.Transaction.account_id == id,
            models.Transaction.user_id == g.user.id,
            models.Transaction.date >= start_date,
            models.Transaction.date <= end_date,
            models.Transaction.is_deleted == 0,
        )
    else:
        total = models.db.session.query(func.count(models.Transaction.id)).filter(
            models.Transaction.account_id == id,
            models.Transaction.user == g.user,
            models.Transaction.is_deleted == 0,
        )

    transactions = (
        models.db.session.query(models.Transaction)
        .filter(
            models.Transaction.account_id == id,
            models.Transaction.user == g.user,
            models.Transaction.is_deleted == 0,
        )
        .order_by(models.Transaction.date.desc())
    )

    if start_date and end_date:
        transactions = transactions.filter(
            models.Transaction.date >= start_date, models.Transaction.date <= end_date
        )

    if search:
        total = total.filter(models.Transaction.memo.like("%{0}%".format(search)))
        transactions = transactions.filter(
            or_(
                models.Transaction.memo.like("%{0}%".format(search)),
                models.Transaction.credit.like("{0}%".format(search)),
                models.Transaction.debit.like("-{0}%".format(search)),
            )
        )

    data["total"] = total.one()[0]

    for transaction in transactions.offset(offset).limit(limit).all():
        data["rows"].append(transaction.dump())

    return jsonify(data)


@data_transactions.route(
    "/bankaccount/<int:id>", defaults={"start_date": None, "end_date": None}
)
@data_transactions.route(
    "/bankaccount/<int:id>/" "<string:start_date>/<string:end_date>"
)
@auth_required()
def bankaccount(id, start_date, end_date):
    """Return data on all transactions for an bankaccount."""

    data = {"rows": []}

    search = request.args.get("search")
    offset = request.args.get("offset", 1)
    limit = request.args.get("limit", 25)

    if start_date and end_date:
        start_date = datetime.strptime(start_date, "%Y%m%d")
        end_date = datetime.strptime(end_date, "%Y%m%d")

    total = models.db.session.query(func.count(models.Transaction.id)).filter(
        models.Transaction.bankaccount_id == id,
        models.Transaction.user_id == g.user.id,
        models.Transaction.is_deleted == 0,
    )

    if start_date and end_date:
        total = total.filter(
            models.Transaction.date >= start_date, models.Transaction.date <= end_date
        )

    transactions = (
        models.db.session.query(models.Transaction)
        .filter(
            models.Transaction.bankaccount_id == id,
            models.Transaction.user == g.user,
            models.Transaction.is_deleted == 0,
        )
        .order_by(models.Transaction.date.desc())
    )

    if start_date and end_date:
        transactions = transactions.filter(
            models.Transaction.date >= start_date, models.Transaction.date <= end_date
        )

    if search:
        total = total.filter(models.Transaction.memo.like("%{0}%".format(search)))
        transactions = transactions.filter(
            or_(
                models.Transaction.memo.like("%{0}%".format(search)),
                models.Transaction.credit.like("{0}%".format(search)),
                models.Transaction.debit.like("-{0}%".format(search)),
            )
        )

    data["total"] = total.one()[0]

    for transaction in transactions.offset(offset).limit(limit).all():
        data["rows"].append(transaction.dump())

    return jsonify(data)


@data_transactions.route(
    "/accounttype/<string:accounttype>", defaults={"start_date": None, "end_date": None}
)
@data_transactions.route(
    "/accounttype/<string:accounttype>/<string:start_date>/<string:end_date>"
)
@auth_required()
def accounttype(accounttype, start_date, end_date):
    """Return data on all transactions for an account type."""

    data = {"rows": []}

    search = request.args.get("search")
    offset = request.args.get("offset", 1)
    limit = request.args.get("limit", 25)

    if accounttype.lower() == "revenue":
        at = (
            models.db.session.query(models.AccountType)
            .filter(
                models.AccountType.name == "Revenue", models.AccountType.parent == None
            )
            .one()
        )  # noqa[E711]
    elif accounttype.lower() == "expenses":
        at = (
            models.db.session.query(models.AccountType)
            .filter(
                models.AccountType.name == "Expenses", models.AccountType.parent == None
            )
            .one()
        )  # noqa[E711]
    elif accounttype.lower() == "liabilities":
        at = (
            models.db.session.query(models.AccountType)
            .filter(
                models.AccountType.name == "Liabilities",
                models.AccountType.parent == None,
            )
            .one()
        )  # noqa[E711]
    elif accounttype.lower() == "assets":
        at = (
            models.db.session.query(models.AccountType)
            .filter(
                models.AccountType.name == "Assets", models.AccountType.parent == None
            )
            .one()
        )  # noqa[E711]
    else:
        raise Exception("Unknown accounttype")

    if start_date and end_date:
        start_date = datetime.strptime(start_date, "%Y%m%d")
        end_date = datetime.strptime(end_date, "%Y%m%d")

    results = (
        models.db.session.query(models.Transaction, models.Account, models.AccountType)
        .filter(
            models.Transaction.user_id == g.user.id,
            models.Transaction.is_deleted == 0,
        )
        .filter(models.Transaction.account_id == models.Account.id)
        .filter(models.Account.accounttype_id == models.AccountType.id)
        .filter(models.AccountType.parent == at)
        .order_by(models.Transaction.date.desc())
    )

    if start_date:
        results = results.filter(models.Transaction.date >= start_date)

    if end_date:
        results = results.filter(models.Transaction.date <= end_date)

    if search:
        results = results.filter(
            or_(
                models.Transaction.memo.like("%{0}%".format(search)),
                models.Transaction.credit.like("{0}%".format(search)),
                models.Transaction.debit.like("-{0}%".format(search)),
            )
        )

    data["total"] = results.with_entities(func.count()).scalar()

    for results in results.offset(offset).limit(limit).all():
        transaction = results[0]
        data["rows"].append(transaction.dump())

    return jsonify(data)


@data_transactions.route(
    "/tag/<int:id>", defaults={"start_date": None, "end_date": None}
)
@data_transactions.route("/tag/<int:id>/<string:start_date>/<string:end_date>")
@auth_required()
def tag(id, **kwargs):
    """Return data on all transactions for a tag."""

    data = {"rows": []}

    search = request.args.get("search")
    offset = request.args.get("offset", 0)
    limit = request.args.get("limit", 25)

    try:
        tag = (
            models.db.session.query(models.Tag)
            .filter(models.Tag.user == g.user, models.Tag.id == id)
            .one()
        )
    except NoResultFound:
        return abort(404)

    results = tag.transactions.filter(models.Transaction.is_deleted == 0).order_by(
        models.Transaction.date.desc()
    )

    if kwargs.get("start_date") and kwargs.get("end_date"):
        try:
            start_date: datetime = datetime.strptime(kwargs.get("start_date"), "%Y%m%d")
            end_date: datetime = datetime.strptime(kwargs.get("end_date"), "%Y%m%d")
        except ValueError:
            return abort(400)
        else:
            results = results.filter(
                models.Transaction.date >= start_date,
                models.Transaction.date <= end_date,
            )

    if search:
        results = results.filter(
            or_(
                models.Transaction.memo.like("%{0}%".format(search)),
                models.Transaction.credit.like(  # pyright: ignore[reportGeneralTypeIssues]
                    "{0}%".format(search)
                ),
                models.Transaction.debit.like(  # pyright: ignore[reportGeneralTypeIssues]
                    "-{0}%".format(search)
                ),
            )
        )

    data["total"] = len(list(results))  # pyright: ignore[reportGeneralTypeIssues]
    for transaction in results.offset(offset).limit(limit).all():
        data["rows"].append(transaction.dump())

    return jsonify(data)