123 lines
4.0 KiB
Python
123 lines
4.0 KiB
Python
from flask import Flask, request, render_template, redirect, abort, send_file
|
|
from sqlalchemy import exc
|
|
from models import *
|
|
import csv
|
|
|
|
app = Flask(__name__)
|
|
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://assetadmin:1234@localhost/asset_test_db'
|
|
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
|
|
db.init_app(app)
|
|
|
|
|
|
with app.app_context():
|
|
db.create_all()
|
|
|
|
|
|
@app.route('/')
|
|
def index():
|
|
return redirect('/view')
|
|
|
|
|
|
@app.route('/create/', methods=['GET', 'POST'])
|
|
def create():
|
|
if request.method == 'GET':
|
|
return render_template('create.html')
|
|
|
|
if request.method == 'POST':
|
|
assettag = request.form['assettag']
|
|
hostname = request.form['hostname']
|
|
warrantyfrom = request.form['warrantyfrom']
|
|
status = request.form['status']
|
|
try:
|
|
staffnum = int(request.form['staffnum'])
|
|
except ValueError:
|
|
return render_template('create.html', exc='staffnum')
|
|
|
|
item = Asset(assettag=assettag, hostname=hostname, warrantyfrom=warrantyfrom, status=status, staffnum=staffnum)
|
|
|
|
try:
|
|
db.session.add(item)
|
|
db.session.commit()
|
|
except exc.IntegrityError:
|
|
return render_template('create.html', exc='integrity')
|
|
except exc.StatementError:
|
|
return render_template('create.html', exc='status')
|
|
|
|
return redirect('/view')
|
|
|
|
|
|
@app.route('/view/', methods=['GET', 'POST'])
|
|
def view_list():
|
|
if request.method == 'POST':
|
|
outfile = open('inventory_export.csv', 'w', newline='')
|
|
outcsv = csv.writer(outfile)
|
|
records = db.session.query(Asset).all()
|
|
outcsv.writerow([column.hostname for column in Asset.__mapper__.columns])
|
|
[outcsv.writerow([getattr(curr, column.hostname) for column in Asset.__mapper__.columns]) for curr in records]
|
|
outfile.close()
|
|
return send_file('inventory_export.csv', as_attachment=True)
|
|
|
|
items = Asset.query.all()
|
|
return render_template('viewList.html', items=items)
|
|
|
|
|
|
@app.route('/view/<string:assettag>/')
|
|
def view_item(assettag):
|
|
item = Asset.query.filter_by(assettag=assettag).first()
|
|
if item:
|
|
return render_template('viewitem.html', item=item)
|
|
return f"Asset {assettag} is not found"
|
|
|
|
|
|
@app.route('/update/<string:assettag>/', methods=['GET', 'POST'])
|
|
def update(assettag):
|
|
item = Asset.query.filter_by(assettag=assettag).first()
|
|
if request.method == 'POST':
|
|
if item:
|
|
assettag = request.form['assettag']
|
|
hostname = request.form['hostname']
|
|
warrantyfrom = request.form['warrantyfrom']
|
|
try:
|
|
status = float(request.form['status'])
|
|
except ValueError:
|
|
return render_template('update.html', item=item, exc='status')
|
|
try:
|
|
staffnum = int(request.form['staffnum'])
|
|
except ValueError:
|
|
return render_template('update.html', item=item, exc='staffnum')
|
|
|
|
try:
|
|
setattr(item, 'assettag', assettag)
|
|
setattr(item, 'hostname', hostname)
|
|
setattr(item, 'warrantyfrom', warrantyfrom)
|
|
setattr(item, 'status', status)
|
|
setattr(item, 'staffnum', staffnum)
|
|
|
|
db.session.commit()
|
|
except exc.IntegrityError:
|
|
return render_template('update.html', item=item, exc='integrity')
|
|
except (exc.StatementError, exc.InvalidRequestError) as e:
|
|
return render_template('update.html', item=item, exc='status')
|
|
|
|
return redirect(f'/view/')
|
|
return f"Asset {assettag} is not found"
|
|
|
|
return render_template('update.html', item=item)
|
|
|
|
|
|
@app.route('/delete/<string:assettag>/', methods=['GET', 'POST'])
|
|
def delete(assettag):
|
|
item = Asset.query.filter_by(assettag=assettag).first()
|
|
if request.method == 'POST':
|
|
if item:
|
|
db.session.delete(item)
|
|
db.session.commit()
|
|
return redirect('/view')
|
|
abort(404)
|
|
|
|
return render_template('delete.html')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
app.run(host='localhost', port=5000)
|