app.main.signups() - Code Metrics - Inspection of "Security fix to remove use of eval function." - whahn1983/pycashflow - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 9a9574...508c9f )

by William

created 2024-04-16 13:56 UTC

app.main.signups() B

↳ Parent: app.main

Complexity

Conditions

Size

Total Lines	30
Code Lines	21

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	5
eloc	21
nop	0
dl	0
loc	30
rs	8.9093
c	0
b	0
f	0

from flask import request, redirect, url_for, send_from_directory, flash, send_file, Response
from flask_login import login_required, current_user
from flask import Blueprint, render_template
from .models import Schedule, Balance, 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, check_password_hash
from .cashflow import update_cash, plot_cash
from .auth import admin_required
from .files import export, upload, version


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:
        float(balance.amount)
    except:
        balance = Balance(amount='0',
                          date=datetime.today())
        db.session.add(balance)
        db.session.commit()

    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():
    # get about info
    about = version()

    return render_template('settings.html', about=about)


@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(),
                            firstdate=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']
        startdate = request.form['startdate']
        if (datetime.strptime(startdate, format).date() != my_data.startdate and my_data.startdate.day !=
                datetime.strptime(startdate, format).day):
            my_data.firstdate = datetime.strptime(startdate, format).date()
        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()
    flash("Added Hold")

    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()
    flash("Added Skip")

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


@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()
        current = request.form['current']
        password = request.form['password']
        password2 = request.form['password2']
        if password == password2 and check_password_hash(my_user.password, current):
            my_user.password = generate_password_hash(password, method='scrypt')
            db.session.commit()
            flash('Password change successful')
        elif password != password2:
            flash('Passwords do not match')
        elif not check_password_hash(my_user.password, current):
            flash('Incorrect password')

        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:
            if request.form['signupvalue'] == "True":
                signupvalue = True
            else:
                signupvalue = False
            signupsettingname.value = 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
        if request.form['signupvalue'] == "True":
            signupvalue = True
        else:
            signupvalue = False
        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']
        if request.form['admin'] == "True":
            adminvalue = True
        else:
            adminvalue = False
        my_data.admin = adminvalue
        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']
        if request.form['admin'] == "True":
            adminvalue = True
        else:
            adminvalue = False
        password = generate_password_hash(request.form['password'], method='scrypt')
        user = User(name=name, email=email, admin=adminvalue, 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('/export', methods=('GET', 'POST'))
@login_required
@admin_required
def export_csv():

    csv_data = export()

    # Create a direct download response with the CSV data and appropriate headers
    response = Response(csv_data, content_type="text/csv")
    response.headers["Content-Disposition"] = "attachment; filename=schedule_export.csv"

    return response


@main.route('/import', methods=('GET', 'POST'))
@login_required
@admin_required
def import_csv():
    if request.method == 'POST':
        csv_file = request.files.get('file')

        upload(csv_file)

        flash("Import Successful")

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


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

whahn1983 / pycashflow

Push — master ( 9a9574...508c9f )

app.main.signups() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like