app.main.update() - Code Metrics - Inspection of "added fix for end of the month dates getting shift..." - whahn1983/pycashflow - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( bfb52c...d38334 )

by William

created 2024-01-30 16:53 UTC

app.main.update() B

↳ Parent: app.main

Complexity

Conditions

Size

Total Lines	30
Code Lines	26

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	6
eloc	26
nop	0
dl	0
loc	30
rs	8.3226
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, select
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()

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

whahn1983 / pycashflow

Push — master ( bfb52c...d38334 )

app.main.update() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like