app.main.import_csv() - Code Metrics - Inspection of "27fb16249bd771d82f4188a7399aa48cb29e6ada" - whahn1983/pycashflow - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Branch — master (27fb16)

by William

created 2024-01-18 01:01 UTC

app.main.import_csv() A

↳ Parent: app.main

Complexity

Conditions

Size

Total Lines	12
Code Lines	9

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	9
nop	0
dl	0
loc	12
rs	9.95
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
from .cashflow import update_cash, plot_cash
from .auth import admin_required
from .files import export, upload


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

whahn1983 / pycashflow

Branch — master (27fb16)

app.main.import_csv() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like