app.main.refresh() - Code Metrics - Inspection of "added auto refresh for first load of the day as we..." - whahn1983/pycashflow - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 8e059e...02a255 )

by William

created 2024-01-02 17:42 UTC

app.main.refresh() A

↳ Parent: app.main

Complexity

Conditions

Size

Total Lines	9
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	7
nop	0
dl	0
loc	9
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 update_cash, plot_cash
from .auth import admin_required


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()

    refresh=0

    update_cash(balance, refresh)

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

    if current_user.admin:
        return render_template('index.html', title='Index', todaydate=todaydate, balance=balance.amount,
                           minbalance=minbalance, graphJSON=graphJSON)
    else:
        return render_template('index_guest.html', title='Index', todaydate=todaydate, balance=balance.amount,
                           minbalance=minbalance, graphJSON=graphJSON)


@main.route('/refresh')
@login_required
def refresh():
    balance = Balance.query.order_by(desc(Balance.date), desc(Balance.id)).first()
    refresh=1

    update_cash(balance, refresh)

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


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

    if current_user.admin:
        return render_template('profile.html')
    else:
        return render_template('profile_guest.html')


@main.route('/settings')
@login_required
@admin_required
def settings():

    return render_template('settings.html')


@main.route('/schedule')
@login_required
@admin_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
@admin_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
@admin_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
@admin_required
def update():
    # update an existing schedule item
    format = '%Y-%m-%d'

    if request.method == 'POST':
        current = Schedule.query.filter_by(id=request.form['id']).first()
        existing = Schedule.query.filter_by(name=request.form['name']).first()
        if existing:
            if current.name != request.form['name']:
                flash("Schedule name 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
@admin_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
@admin_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
@admin_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
@admin_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
@admin_required
def clear_holds():
    # clear holds
    db.session.query(Hold).delete()
    db.session.commit()

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


@main.route('/clearskips')
@login_required
@admin_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
@admin_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
@admin_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('/signups', methods=('GET', 'POST'))
@login_required
@admin_required
def signups():
    # 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.settings'))

        # 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.settings'))

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


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

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


@main.route('/email', methods=('GET', 'POST'))
@login_required
@admin_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.settings'))

        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.settings'))

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


@main.route('/users_table')
@login_required
@admin_required
def users():
    users = User.query

    return render_template('users_table.html', title='Users Table', users=users)


@main.route('/update_user', methods=['GET', 'POST'])
@login_required
@admin_required
def update_user():
    # update an existing user
    if request.method == 'POST':
        current = User.query.filter_by(id=request.form['id']).first()
        existing = User.query.filter_by(email=request.form['email']).first()
        if existing:
            if current.email != request.form['email']:
                flash("Email already exists")
                return redirect(url_for('main.users'))
        my_data = User.query.get(request.form.get('id'))
        my_data.name = request.form['name']
        my_data.email = request.form['email']
        my_data.admin = eval(request.form['admin'])
        db.session.commit()
        flash("Updated Successfully")

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

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


@main.route('/delete_user/<id>')
@login_required
@admin_required
def delete_user(id):
    # delete a user
    user = User.query.filter_by(id=id).first()

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

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


@main.route('/create_user', methods=('GET', 'POST'))
@login_required
@admin_required
def create_user():
    # create a new user
    if request.method == 'POST':
        name = request.form['name']
        email = request.form['email']
        admin = eval(request.form['admin'])
        password = generate_password_hash(request.form['password'], method='scrypt')
        user = User(name=name, email=email, admin=admin, password=password)
        existing = User.query.filter_by(email=email).first()
        if existing:
            flash("User already exists")
            return redirect(url_for('main.users'))
        db.session.add(user)
        db.session.commit()
        flash("Added Successfully")

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

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


@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 update_cash, plot_cash
11			from .auth import admin_required
12
13
14			main = Blueprint('main', __name__)
15
16
17			@main.route('/', methods=('GET', 'POST'))
18			@login_required
19			def index():
20			# get today's date
21			todaydate = datetime.today().strftime('%A, %B %d, %Y')
22
23			# query the latest balance information
24			balance = Balance.query.order_by(desc(Balance.date), desc(Balance.id)).first()
25
26			refresh=0
27
28			update_cash(balance, refresh)
29
30			# plot cash flow results
31			minbalance, graphJSON = plot_cash()
32
33			if current_user.admin:
34			return render_template('index.html', title='Index', todaydate=todaydate, balance=balance.amount,
35			minbalance=minbalance, graphJSON=graphJSON)
36			else:
37			return render_template('index_guest.html', title='Index', todaydate=todaydate, balance=balance.amount,
38			minbalance=minbalance, graphJSON=graphJSON)
39
40
41			@main.route('/refresh')
42			@login_required
43			def refresh():
44			balance = Balance.query.order_by(desc(Balance.date), desc(Balance.id)).first()
45			refresh=1
46
47			update_cash(balance, refresh)
48
49			return redirect(url_for('main.index'))
50
51
52			@main.route('/profile')
53			@login_required
54			def profile():
55
56			if current_user.admin:
57			return render_template('profile.html')
58			else:
59			return render_template('profile_guest.html')
60
61
62			@main.route('/settings')
63			@login_required
64			@admin_required
65			def settings():
66
67			return render_template('settings.html')
68
69
70			@main.route('/schedule')
71			@login_required
72			@admin_required
73			def schedule():
74			schedule = Schedule.query.order_by(asc(extract('day', Schedule.startdate)))
75
76			return render_template('schedule_table.html', title='Schedule Table', schedule=schedule)
77
78
79			@main.route('/holds')
80			@login_required
81			@admin_required
82			def holds():
83			hold = Hold.query
84			skip = Skip.query
85
86			return render_template('holds_table.html', title='Holds Table', hold=hold, skip=skip)
87
88
89			@main.route('/create', methods=('GET', 'POST'))
90			@login_required
91			@admin_required
92			def create():
93			# create a new schedule item
94			format = '%Y-%m-%d'
95			if request.method == 'POST':
96			name = request.form['name']
97			amount = request.form['amount']
98			frequency = request.form['frequency']
99			startdate = request.form['startdate']
100			type = request.form['type']
101			schedule = Schedule(name=name,
102			type=type,
103			amount=amount,
104			frequency=frequency,
105			startdate=datetime.strptime(startdate, format).date())
106			existing = Schedule.query.filter_by(name=name).first()
107			if existing:
108			flash("Schedule already exists")
109			return redirect(url_for('main.schedule'))
110			db.session.add(schedule)
111			db.session.commit()
112			flash("Added Successfully")
113
114			return redirect(url_for('main.schedule'))
115
116			return redirect(url_for('main.schedule'))
117
118
119			@main.route('/update', methods=['GET', 'POST'])
120			@login_required
121			@admin_required
122			def update():
123			# update an existing schedule item
124			format = '%Y-%m-%d'
125
126			if request.method == 'POST':
127			current = Schedule.query.filter_by(id=request.form['id']).first()
128			existing = Schedule.query.filter_by(name=request.form['name']).first()
129			if existing:
130			if current.name != request.form['name']:
131			flash("Schedule name already exists")
132			return redirect(url_for('main.schedule'))
133			my_data = Schedule.query.get(request.form.get('id'))
134			my_data.name = request.form['name']
135			my_data.amount = request.form['amount']
136			my_data.type = request.form['type']
137			my_data.frequency = request.form['frequency']
138			my_data.startdate = request.form['startdate']
139			startdate = request.form['startdate']
140			my_data.startdate = datetime.strptime(startdate, format).date()
141			db.session.commit()
142			flash("Updated Successfully")
143
144			return redirect(url_for('main.schedule'))
145
146			return redirect(url_for('main.schedule'))
147
148
149			@main.route('/addhold/<id>')
150			@login_required
151			@admin_required
152			def addhold(id):
153			# add a hold item from the schedule
154			schedule = Schedule.query.filter_by(id=id).first()
155			hold = Hold(name=schedule.name, type=schedule.type, amount=schedule.amount)
156			db.session.add(hold)
157			db.session.commit()
158
159			return redirect(url_for('main.schedule'))
160
161
162			@main.route('/addskip/<id>')
163			@login_required
164			@admin_required
165			def addskip(id):
166			# add a skip item from the schedule
167			transaction = Transactions.query.filter_by(id=id).first()
168			trans_type = ""
169			if transaction.type == "Expense":
170			trans_type = "Income"
171			elif transaction.type == "Income":
172			trans_type = "Expense"
173			skip = Skip(name=transaction.name + " (SKIP)", type=trans_type, amount=transaction.amount, date=transaction.date)
174			db.session.add(skip)
175			db.session.commit()
176
177			return redirect(url_for('main.transactions'))
178
179
180			@main.route('/deletehold/<id>')
181			@login_required
182			@admin_required
183			def holds_delete(id):
184			# delete a hold item
185			hold = Hold.query.filter_by(id=id).first()
186
187			if hold:
188			db.session.delete(hold)
189			db.session.commit()
190			flash("Deleted Successfully")
191
192			return redirect(url_for('main.holds'))
193
194
195			@main.route('/deleteskip/<id>')
196			@login_required
197			@admin_required
198			def skips_delete(id):
199			# delete a skip item
200			skip = Skip.query.filter_by(id=id).first()
201
202			if skip:
203			db.session.delete(skip)
204			db.session.commit()
205			flash("Deleted Successfully")
206
207			return redirect(url_for('main.holds'))
208
209
210			@main.route('/clearholds')
211			@login_required
212			@admin_required
213			def clear_holds():
214			# clear holds
215			db.session.query(Hold).delete()
216			db.session.commit()
217
218			return redirect(url_for('main.holds'))
219
220
221			@main.route('/clearskips')
222			@login_required
223			@admin_required
224			def clear_skips():
225			# clear skips
226			db.session.query(Skip).delete()
227			db.session.commit()
228
229			return redirect(url_for('main.index'))
230
231
232			@main.route('/delete/<id>')
233			@login_required
234			@admin_required
235			def schedule_delete(id):
236			# delete a schedule item
237			schedule = Schedule.query.filter_by(id=id).first()
238
239			if schedule:
240			db.session.delete(schedule)
241			db.session.commit()
242			flash("Deleted Successfully")
243
244			return redirect(url_for('main.schedule'))
245
246
247			@main.route('/favicon')
248			def favicon():
249			return send_from_directory(os.path.join(main.root_path, 'static'),
250			'favicon.ico', mimetype='image/vnd.microsoft.icon')
251
252
253			@main.route('/appleicon')
254			def appleicon():
255			return send_from_directory(os.path.join(main.root_path, 'static'),
256			'apple-touch-icon.png', mimetype='image/png')
257
258
259			@main.route('/balance', methods=('GET', 'POST'))
260			@login_required
261			@admin_required
262			def balance():
263			# manually update the balance from the balance button
264			format = '%Y-%m-%d'
265			if request.method == 'POST':
266			amount = request.form['amount']
267			dateentry = request.form['date']
268			balance = Balance(amount=amount,
269			date=datetime.strptime(dateentry, format).date())
270			db.session.add(balance)
271			db.session.commit()
272
273			return redirect(url_for('main.index'))
274
275
276			@main.route('/changepw', methods=('GET', 'POST'))
277			@login_required
278			def changepw():
279			# change the users password from the profile page
280			if request.method == 'POST':
281			curr_user = current_user.id
282			my_user = User.query.filter_by(id=curr_user).first()
283			password = request.form['password']
284			my_user.password = generate_password_hash(password, method='scrypt')
285			db.session.commit()
286
287			return redirect(url_for('main.profile'))
288
289			return redirect(url_for('main.profile'))
290
291
292			@main.route('/signups', methods=('GET', 'POST'))
293			@login_required
294			@admin_required
295			def signups():
296			# set the settings options, in this case disable signups, from the profile page
297			if request.method == 'POST':
298			signupsettingname = Settings.query.filter_by(name='signup').first()
299
300			if signupsettingname:
301			signupvalue = request.form['signupvalue']
302			signupsettingname.value = eval(signupvalue)
303			db.session.commit()
304
305			return redirect(url_for('main.settings'))
306
307			# store the signup option value in the database to check when the user clicks signup
308			signupvalue = request.form['signupvalue']
309			signupvalue = eval(signupvalue)
310			settings = Settings(name="signup",
311			value=signupvalue)
312			db.session.add(settings)
313			db.session.commit()
314
315			return redirect(url_for('main.settings'))
316
317			return redirect(url_for('main.settings'))
318
319
320			@main.route('/transactions')
321			@login_required
322			@admin_required
323			def transactions():
324			total = Transactions.query
325
326			return render_template('transactions_table.html', total=total)
327
328
329			@main.route('/email', methods=('GET', 'POST'))
330			@login_required
331			@admin_required
332			def email():
333			# set the users email address, password, and server for the auto email balance update
334			if request.method == 'POST':
335			emailsettings = Email.query.filter_by(id=1).first()
336
337			if emailsettings:
338			email = request.form['email']
339			password = request.form['password']
340			server = request.form['server']
341			subjectstr = request.form['subject_str']
342			startstr = request.form['start_str']
343			endstr = request.form['end_str']
344			emailsettings.email = email
345			emailsettings.password = password
346			emailsettings.server = server
347			emailsettings.subjectstr = subjectstr
348			emailsettings.startstr = startstr
349			emailsettings.endstr = endstr
350			db.session.commit()
351
352			return redirect(url_for('main.settings'))
353
354			email = request.form['email']
355			password = request.form['password']
356			server = request.form['server']
357			subjectstr = request.form['subject_str']
358			startstr = request.form['start_str']
359			endstr = request.form['end_str']
360			emailentry = Email(email=email, password=password, server=server, subjectstr=subjectstr, startstr=startstr,
361			endstr=endstr)
362			db.session.add(emailentry)
363			db.session.commit()
364
365			return redirect(url_for('main.settings'))
366
367			return redirect(url_for('main.settings'))
368
369
370			@main.route('/users_table')
371			@login_required
372			@admin_required
373			def users():
374			users = User.query
375
376			return render_template('users_table.html', title='Users Table', users=users)
377
378
379			@main.route('/update_user', methods=['GET', 'POST'])
380			@login_required
381			@admin_required
382			def update_user():
383			# update an existing user
384			if request.method == 'POST':
385			current = User.query.filter_by(id=request.form['id']).first()
386			existing = User.query.filter_by(email=request.form['email']).first()
387			if existing:
388			if current.email != request.form['email']:
389			flash("Email already exists")
390			return redirect(url_for('main.users'))
391			my_data = User.query.get(request.form.get('id'))
392			my_data.name = request.form['name']
393			my_data.email = request.form['email']
394			my_data.admin = eval(request.form['admin'])
395			db.session.commit()
396			flash("Updated Successfully")
397
398			return redirect(url_for('main.users'))
399
400			return redirect(url_for('main.users'))
401
402
403			@main.route('/delete_user/<id>')
404			@login_required
405			@admin_required
406			def delete_user(id):
407			# delete a user
408			user = User.query.filter_by(id=id).first()
409
410			if user:
411			db.session.delete(user)
412			db.session.commit()
413			flash("Deleted Successfully")
414
415			return redirect(url_for('main.users'))
416
417
418			@main.route('/create_user', methods=('GET', 'POST'))
419			@login_required
420			@admin_required
421			def create_user():
422			# create a new user
423			if request.method == 'POST':
424			name = request.form['name']
425			email = request.form['email']
426			admin = eval(request.form['admin'])
427			password = generate_password_hash(request.form['password'], method='scrypt')
428			user = User(name=name, email=email, admin=admin, password=password)
429			existing = User.query.filter_by(email=email).first()
430			if existing:
431			flash("User already exists")
432			return redirect(url_for('main.users'))
433			db.session.add(user)
434			db.session.commit()
435			flash("Added Successfully")
436
437			return redirect(url_for('main.users'))
438
439			return redirect(url_for('main.users'))
440
441
442			@main.route('/manifest.json')
443			def serve_manifest():
444			return send_file('manifest.json', mimetype='application/manifest+json')
445
446
447			@main.route('/sw.js')
448			def serve_sw():
449			return send_file('sw.js', mimetype='application/javascript')
450

whahn1983 / pycashflow

Push — master ( 8e059e...02a255 )

app.main.refresh() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like