app.main.transactions() - Code Metrics - Inspection of "Added ability to skip future transactions from the..." - whahn1983/pycashflow - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 4e81b4...441e87 )

by William

created 2023-11-12 20:21 UTC

app.main.transactions() A

↳ Parent: app.main

Complexity

Conditions

Size

Total Lines	6
Code Lines	5

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	5
nop	0
dl	0
loc	6
rs	10
c	0
b	0
f	0

from flask import request, redirect, url_for, send_from_directory, flash, send_file
from flask_login import login_required, current_user
from flask import Blueprint, render_template
from .models import Schedule, Balance, Total, Running, User, Settings, Transactions, Email, Hold, Skip
from app import db
from datetime import datetime
import os
from sqlalchemy import desc, extract, asc
from werkzeug.security import generate_password_hash
from .cashflow import calc_schedule, calc_transactions, plot_cash


main = Blueprint('main', __name__)


@main.route('/', methods=('GET', 'POST'))
@login_required
def index():
    # get today's date
    todaydate = datetime.today().strftime('%A, %B %d, %Y')

    # query the latest balance information
    balance = Balance.query.order_by(desc(Balance.date), desc(Balance.id)).first()

    try:
        if balance.amount:
            db.session.query(Balance).delete()
            balance = Balance(amount=balance.amount, date=datetime.today())
            db.session.add(balance)
            db.session.commit()
    except:
        balance = Balance(amount='0',
                          date=datetime.today())
        db.session.add(balance)
        db.session.commit()

    # empty the tables to create fresh data from the schedule
    db.session.query(Total).delete()
    db.session.query(Running).delete()
    db.session.query(Transactions).delete()
    db.session.commit()

    # retrieve the number of years for the cash flow plot
    yearamount = request.form.get('yearamount')

    # calculate total events for the year amount
    calc_schedule(yearamount)

    # calculate sum of running transactions
    calc_transactions(balance)

    # plot cash flow results
    minbalance, graphJSON = plot_cash()

    return render_template('index.html', title='Index', todaydate=todaydate, balance=balance.amount,
                           minbalance=minbalance, graphJSON=graphJSON)


@main.route('/profile')
@login_required
def profile():

    return render_template('profile.html')


@main.route('/schedule')
@login_required
def schedule():
    schedule = Schedule.query.order_by(asc(extract('day', Schedule.startdate)))

    return render_template('schedule_table.html', title='Schedule Table', schedule=schedule)


@main.route('/holds')
@login_required
def holds():
    hold = Hold.query
    skip = Skip.query

    return render_template('holds_table.html', title='Holds Table', hold=hold, skip=skip)


@main.route('/create', methods=('GET', 'POST'))
@login_required
def create():
    # create a new schedule item
    format = '%Y-%m-%d'
    if request.method == 'POST':
        name = request.form['name']
        amount = request.form['amount']
        frequency = request.form['frequency']
        startdate = request.form['startdate']
        type = request.form['type']
        schedule = Schedule(name=name,
                          type=type,
                          amount=amount,
                          frequency=frequency,
                          startdate=datetime.strptime(startdate, format).date())
        existing = Schedule.query.filter_by(name=name).first()
        if existing:
            flash("Schedule already exists")
            return redirect(url_for('main.schedule'))
        db.session.add(schedule)
        db.session.commit()
        flash("Added Successfully")

        return redirect(url_for('main.schedule'))

    return redirect(url_for('main.schedule'))


@main.route('/update', methods=['GET', 'POST'])
@login_required
def update():
    # update an existing schedule item
    format = '%Y-%m-%d'

    if request.method == 'POST':
        existing = Schedule.query.filter_by(name=request.form['name']).first()
        if existing:
            flash("Schedule already exists")
            return redirect(url_for('main.schedule'))
        my_data = Schedule.query.get(request.form.get('id'))
        my_data.name = request.form['name']
        my_data.amount = request.form['amount']
        my_data.type = request.form['type']
        my_data.frequency = request.form['frequency']
        my_data.startdate = request.form['startdate']
        startdate = request.form['startdate']
        my_data.startdate = datetime.strptime(startdate, format).date()
        db.session.commit()
        flash("Updated Successfully")

        return redirect(url_for('main.schedule'))

    return redirect(url_for('main.schedule'))


@main.route('/addhold/<id>')
@login_required
def addhold(id):
    # add a hold item from the schedule
    schedule = Schedule.query.filter_by(id=id).first()
    hold = Hold(name=schedule.name, type=schedule.type, amount=schedule.amount)
    db.session.add(hold)
    db.session.commit()

    return redirect(url_for('main.schedule'))


@main.route('/addskip/<id>')
@login_required
def addskip(id):
    # add a skip item from the schedule
    transaction = Transactions.query.filter_by(id=id).first()
    trans_type = ""
    if transaction.type == "Expense":
        trans_type = "Income"
    elif transaction.type == "Income":
        trans_type = "Expense"
    skip = Skip(name=transaction.name + " (SKIP)", type=trans_type, amount=transaction.amount, date=transaction.date)
    db.session.add(skip)
    db.session.commit()

    return redirect(url_for('main.transactions'))


@main.route('/deletehold/<id>')
@login_required
def holds_delete(id):
    # delete a hold item
    hold = Hold.query.filter_by(id=id).first()

    if hold:
        db.session.delete(hold)
        db.session.commit()
        flash("Deleted Successfully")

    return redirect(url_for('main.holds'))


@main.route('/deleteskip/<id>')
@login_required
def skips_delete(id):
    # delete a skip item
    skip = Skip.query.filter_by(id=id).first()

    if skip:
        db.session.delete(skip)
        db.session.commit()
        flash("Deleted Successfully")

    return redirect(url_for('main.holds'))


@main.route('/clearholds')
@login_required
def clear_holds():
    # clear holds

    db.session.query(Hold).delete()
    db.session.commit()

    return redirect(url_for('main.index'))


@main.route('/clearskips')
@login_required
def clear_skips():
    # clear skips

    db.session.query(Skip).delete()
    db.session.commit()

    return redirect(url_for('main.index'))


@main.route('/delete/<id>')
@login_required
def schedule_delete(id):
    # delete a schedule item
    schedule = Schedule.query.filter_by(id=id).first()

    if schedule:
        db.session.delete(schedule)
        db.session.commit()
        flash("Deleted Successfully")

    return redirect(url_for('main.schedule'))


@main.route('/favicon')
def favicon():
    return send_from_directory(os.path.join(main.root_path, 'static'),
                               'favicon.ico', mimetype='image/vnd.microsoft.icon')


@main.route('/appleicon')
def appleicon():
    return send_from_directory(os.path.join(main.root_path, 'static'),
                               'apple-touch-icon.png', mimetype='image/png')


@main.route('/balance', methods=('GET', 'POST'))
@login_required
def balance():
    # manually update the balance from the balance button
    format = '%Y-%m-%d'
    if request.method == 'POST':
        amount = request.form['amount']
        dateentry = request.form['date']
        balance = Balance(amount=amount,
                          date=datetime.strptime(dateentry, format).date())
        db.session.add(balance)
        db.session.commit()

        return redirect(url_for('main.index'))


@main.route('/changepw', methods=('GET', 'POST'))
@login_required
def changepw():
    # change the users password from the profile page
    if request.method == 'POST':
        curr_user = current_user.id
        my_user = User.query.filter_by(id=curr_user).first()
        password = request.form['password']
        my_user.password = generate_password_hash(password, method='scrypt')
        db.session.commit()

        return redirect(url_for('main.profile'))

    return redirect(url_for('main.profile'))


@main.route('/settings', methods=('GET', 'POST'))
@login_required
def settings():
    # set the settings options, in this case disable signups, from the profile page
    if request.method == 'POST':
        signupsettingname = Settings.query.filter_by(name='signup').first()

        if signupsettingname:
            signupvalue = request.form['signupvalue']
            signupsettingname.value = eval(signupvalue)
            db.session.commit()

            return redirect(url_for('main.profile'))

        # store the signup option value in the database to check when the user clicks signup
        signupvalue = request.form['signupvalue']
        signupvalue = eval(signupvalue)
        settings = Settings(name="signup",
                          value=signupvalue)
        db.session.add(settings)
        db.session.commit()

        return redirect(url_for('main.profile'))

    return redirect(url_for('main.profile'))


@main.route('/transactions')
@login_required
def transactions():
    total = Transactions.query

    return render_template('transactions_table.html', total=total)


@main.route('/email', methods=('GET', 'POST'))
@login_required
def email():
    # set the users email address, password, and server for the auto email balance update
    if request.method == 'POST':
        emailsettings = Email.query.filter_by(id=1).first()

        if emailsettings:
            email = request.form['email']
            password = request.form['password']
            server = request.form['server']
            subjectstr = request.form['subject_str']
            startstr = request.form['start_str']
            endstr = request.form['end_str']
            emailsettings.email = email
            emailsettings.password = password
            emailsettings.server = server
            emailsettings.subjectstr = subjectstr
            emailsettings.startstr = startstr
            emailsettings.endstr = endstr
            db.session.commit()

            return redirect(url_for('main.profile'))

        email = request.form['email']
        password = request.form['password']
        server = request.form['server']
        subjectstr = request.form['subject_str']
        startstr = request.form['start_str']
        endstr = request.form['end_str']
        emailentry = Email(email=email, password=password, server=server, subjectstr=subjectstr, startstr=startstr,
                           endstr=endstr)
        db.session.add(emailentry)
        db.session.commit()

        return redirect(url_for('main.profile'))

    return redirect(url_for('main.profile'))


@main.route('/manifest.json')
def serve_manifest():
    return send_file('manifest.json', mimetype='application/manifest+json')


@main.route('/sw.js')
def serve_sw():
    return send_file('sw.js', mimetype='application/javascript')


1			from flask import request, redirect, url_for, send_from_directory, flash, send_file
2			from flask_login import login_required, current_user
3			from flask import Blueprint, render_template
4			from .models import Schedule, Balance, Total, Running, User, Settings, Transactions, Email, Hold, Skip
5			from app import db
6			from datetime import datetime
7			import os
8			from sqlalchemy import desc, extract, asc
9			from werkzeug.security import generate_password_hash
10			from .cashflow import calc_schedule, calc_transactions, plot_cash
11
12
13			main = Blueprint('main', __name__)
14
15
16			@main.route('/', methods=('GET', 'POST'))
17			@login_required
18			def index():
19			# get today's date
20			todaydate = datetime.today().strftime('%A, %B %d, %Y')
21
22			# query the latest balance information
23			balance = Balance.query.order_by(desc(Balance.date), desc(Balance.id)).first()
24
25			try:
26			if balance.amount:
27			db.session.query(Balance).delete()
28			balance = Balance(amount=balance.amount, date=datetime.today())
29			db.session.add(balance)
30			db.session.commit()
31			except:
32			balance = Balance(amount='0',
33			date=datetime.today())
34			db.session.add(balance)
35			db.session.commit()
36
37			# empty the tables to create fresh data from the schedule
38			db.session.query(Total).delete()
39			db.session.query(Running).delete()
40			db.session.query(Transactions).delete()
41			db.session.commit()
42
43			# retrieve the number of years for the cash flow plot
44			yearamount = request.form.get('yearamount')
45
46			# calculate total events for the year amount
47			calc_schedule(yearamount)
48
49			# calculate sum of running transactions
50			calc_transactions(balance)
51
52			# plot cash flow results
53			minbalance, graphJSON = plot_cash()
54
55			return render_template('index.html', title='Index', todaydate=todaydate, balance=balance.amount,
56			minbalance=minbalance, graphJSON=graphJSON)
57
58
59			@main.route('/profile')
60			@login_required
61			def profile():
62
63			return render_template('profile.html')
64
65
66			@main.route('/schedule')
67			@login_required
68			def schedule():
69			schedule = Schedule.query.order_by(asc(extract('day', Schedule.startdate)))
70
71			return render_template('schedule_table.html', title='Schedule Table', schedule=schedule)
72
73
74			@main.route('/holds')
75			@login_required
76			def holds():
77			hold = Hold.query
78			skip = Skip.query
79
80			return render_template('holds_table.html', title='Holds Table', hold=hold, skip=skip)
81
82
83			@main.route('/create', methods=('GET', 'POST'))
84			@login_required
85			def create():
86			# create a new schedule item
87			format = '%Y-%m-%d'
88			if request.method == 'POST':
89			name = request.form['name']
90			amount = request.form['amount']
91			frequency = request.form['frequency']
92			startdate = request.form['startdate']
93			type = request.form['type']
94			schedule = Schedule(name=name,
95			type=type,
96			amount=amount,
97			frequency=frequency,
98			startdate=datetime.strptime(startdate, format).date())
99			existing = Schedule.query.filter_by(name=name).first()
100			if existing:
101			flash("Schedule already exists")
102			return redirect(url_for('main.schedule'))
103			db.session.add(schedule)
104			db.session.commit()
105			flash("Added Successfully")
106
107			return redirect(url_for('main.schedule'))
108
109			return redirect(url_for('main.schedule'))
110
111
112			@main.route('/update', methods=['GET', 'POST'])
113			@login_required
114			def update():
115			# update an existing schedule item
116			format = '%Y-%m-%d'
117
118			if request.method == 'POST':
119			existing = Schedule.query.filter_by(name=request.form['name']).first()
120			if existing:
121			flash("Schedule already exists")
122			return redirect(url_for('main.schedule'))
123			my_data = Schedule.query.get(request.form.get('id'))
124			my_data.name = request.form['name']
125			my_data.amount = request.form['amount']
126			my_data.type = request.form['type']
127			my_data.frequency = request.form['frequency']
128			my_data.startdate = request.form['startdate']
129			startdate = request.form['startdate']
130			my_data.startdate = datetime.strptime(startdate, format).date()
131			db.session.commit()
132			flash("Updated Successfully")
133
134			return redirect(url_for('main.schedule'))
135
136			return redirect(url_for('main.schedule'))
137
138
139			@main.route('/addhold/<id>')
140			@login_required
141			def addhold(id):
142			# add a hold item from the schedule
143			schedule = Schedule.query.filter_by(id=id).first()
144			hold = Hold(name=schedule.name, type=schedule.type, amount=schedule.amount)
145			db.session.add(hold)
146			db.session.commit()
147
148			return redirect(url_for('main.schedule'))
149
150
151			@main.route('/addskip/<id>')
152			@login_required
153			def addskip(id):
154			# add a skip item from the schedule
155			transaction = Transactions.query.filter_by(id=id).first()
156			trans_type = ""
157			if transaction.type == "Expense":
158			trans_type = "Income"
159			elif transaction.type == "Income":
160			trans_type = "Expense"
161			skip = Skip(name=transaction.name + " (SKIP)", type=trans_type, amount=transaction.amount, date=transaction.date)
162			db.session.add(skip)
163			db.session.commit()
164
165			return redirect(url_for('main.transactions'))
166
167
168			@main.route('/deletehold/<id>')
169			@login_required
170			def holds_delete(id):
171			# delete a hold item
172			hold = Hold.query.filter_by(id=id).first()
173
174			if hold:
175			db.session.delete(hold)
176			db.session.commit()
177			flash("Deleted Successfully")
178
179			return redirect(url_for('main.holds'))
180
181
182			@main.route('/deleteskip/<id>')
183			@login_required
184			def skips_delete(id):
185			# delete a skip item
186			skip = Skip.query.filter_by(id=id).first()
187
188			if skip:
189			db.session.delete(skip)
190			db.session.commit()
191			flash("Deleted Successfully")
192
193			return redirect(url_for('main.holds'))
194
195
196			@main.route('/clearholds')
197			@login_required
198			def clear_holds():
199			# clear holds
200
201			db.session.query(Hold).delete()
202			db.session.commit()
203
204			return redirect(url_for('main.index'))
205
206
207			@main.route('/clearskips')
208			@login_required
209			def clear_skips():
210			# clear skips
211
212			db.session.query(Skip).delete()
213			db.session.commit()
214
215			return redirect(url_for('main.index'))
216
217
218			@main.route('/delete/<id>')
219			@login_required
220			def schedule_delete(id):
221			# delete a schedule item
222			schedule = Schedule.query.filter_by(id=id).first()
223
224			if schedule:
225			db.session.delete(schedule)
226			db.session.commit()
227			flash("Deleted Successfully")
228
229			return redirect(url_for('main.schedule'))
230
231
232			@main.route('/favicon')
233			def favicon():
234			return send_from_directory(os.path.join(main.root_path, 'static'),
235			'favicon.ico', mimetype='image/vnd.microsoft.icon')
236
237
238			@main.route('/appleicon')
239			def appleicon():
240			return send_from_directory(os.path.join(main.root_path, 'static'),
241			'apple-touch-icon.png', mimetype='image/png')
242
243
244			@main.route('/balance', methods=('GET', 'POST'))
245			@login_required
246			def balance():
247			# manually update the balance from the balance button
248			format = '%Y-%m-%d'
249			if request.method == 'POST':
250			amount = request.form['amount']
251			dateentry = request.form['date']
252			balance = Balance(amount=amount,
253			date=datetime.strptime(dateentry, format).date())
254			db.session.add(balance)
255			db.session.commit()
256
257			return redirect(url_for('main.index'))
258
259
260			@main.route('/changepw', methods=('GET', 'POST'))
261			@login_required
262			def changepw():
263			# change the users password from the profile page
264			if request.method == 'POST':
265			curr_user = current_user.id
266			my_user = User.query.filter_by(id=curr_user).first()
267			password = request.form['password']
268			my_user.password = generate_password_hash(password, method='scrypt')
269			db.session.commit()
270
271			return redirect(url_for('main.profile'))
272
273			return redirect(url_for('main.profile'))
274
275
276			@main.route('/settings', methods=('GET', 'POST'))
277			@login_required
278			def settings():
279			# set the settings options, in this case disable signups, from the profile page
280			if request.method == 'POST':
281			signupsettingname = Settings.query.filter_by(name='signup').first()
282
283			if signupsettingname:
284			signupvalue = request.form['signupvalue']
285			signupsettingname.value = eval(signupvalue)
286			db.session.commit()
287
288			return redirect(url_for('main.profile'))
289
290			# store the signup option value in the database to check when the user clicks signup
291			signupvalue = request.form['signupvalue']
292			signupvalue = eval(signupvalue)
293			settings = Settings(name="signup",
294			value=signupvalue)
295			db.session.add(settings)
296			db.session.commit()
297
298			return redirect(url_for('main.profile'))
299
300			return redirect(url_for('main.profile'))
301
302
303			@main.route('/transactions')
304			@login_required
305			def transactions():
306			total = Transactions.query
307
308			return render_template('transactions_table.html', total=total)
309
310
311			@main.route('/email', methods=('GET', 'POST'))
312			@login_required
313			def email():
314			# set the users email address, password, and server for the auto email balance update
315			if request.method == 'POST':
316			emailsettings = Email.query.filter_by(id=1).first()
317
318			if emailsettings:
319			email = request.form['email']
320			password = request.form['password']
321			server = request.form['server']
322			subjectstr = request.form['subject_str']
323			startstr = request.form['start_str']
324			endstr = request.form['end_str']
325			emailsettings.email = email
326			emailsettings.password = password
327			emailsettings.server = server
328			emailsettings.subjectstr = subjectstr
329			emailsettings.startstr = startstr
330			emailsettings.endstr = endstr
331			db.session.commit()
332
333			return redirect(url_for('main.profile'))
334
335			email = request.form['email']
336			password = request.form['password']
337			server = request.form['server']
338			subjectstr = request.form['subject_str']
339			startstr = request.form['start_str']
340			endstr = request.form['end_str']
341			emailentry = Email(email=email, password=password, server=server, subjectstr=subjectstr, startstr=startstr,
342			endstr=endstr)
343			db.session.add(emailentry)
344			db.session.commit()
345
346			return redirect(url_for('main.profile'))
347
348			return redirect(url_for('main.profile'))
349
350
351			@main.route('/manifest.json')
352			def serve_manifest():
353			return send_file('manifest.json', mimetype='application/manifest+json')
354
355
356			@main.route('/sw.js')
357			def serve_sw():
358			return send_file('sw.js', mimetype='application/javascript')
359

whahn1983 / pycashflow

Push — master ( 4e81b4...441e87 )

app.main.transactions() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like