# DrCr: Web-based double-entry bookkeeping framework # Copyright (C) 2022–2023 Lee Yingtong Li (RunasSudo) # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU Affero General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. # # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . from flask import Flask, g import toml app = Flask(__name__) app.config.from_file('config.toml', load=toml.load) from flask_sqlalchemy.record_queries import get_recorded_queries from .database import db from .models import Transaction from .plugins import init_plugins, transaction_providers from .statements.models import StatementLine import time app.config['SQLALCHEMY_RECORD_QUERIES'] = app.debug db.init_app(app) def all_transactions(join_postings=True): # All Transactions in database if join_postings: transactions = db.session.scalars(db.select(Transaction).options(db.selectinload(Transaction.postings))).all() else: transactions = db.session.scalars(db.select(Transaction)).all() # Unreconciled StatementLines transactions.extend(line.into_transaction() for line in StatementLine.query.filter(StatementLine.reconciliation == None)) # Plugins for transaction_provider in transaction_providers: transactions.extend(transaction_provider()) return transactions from . import views from .journal import views from .statements import views init_plugins() @app.cli.command('initdb') def initdb(): db.create_all() if app.debug: @app.before_request def before_request(): g.start = time.time() @app.after_request def after_request(response): diff = time.time() - g.start if response.response and response.status_code == 200 and response.content_type.startswith('text/html'): response.set_data(response.get_data().replace(b'__EXECUTION_TIME__', bytes(format(diff * 1000, '.1f'), 'utf-8'))) return response @app.context_processor def add_dbtime(): def dbtime(): queries = get_recorded_queries() total_duration = sum(q.duration for q in queries) return format(total_duration * 1000, '.1f') return dict(dbtime=dbtime)