You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

271 lines
8.4 KiB

2 years ago
2 years ago
2 years ago
2 years ago
2 years ago
2 years ago
2 years ago
2 years ago
2 years ago
2 years ago
2 years ago
2 years ago
2 years ago
2 years ago
2 years ago
2 years ago
2 years ago
2 years ago
2 years ago
2 years ago
2 years ago
2 years ago
2 years ago
2 years ago
2 years ago
2 years ago
2 years ago
2 years ago
2 years ago
  1. #!/usr/bin/env python3
  2. from flask import Flask, render_template, session, request, abort, redirect, url_for, jsonify
  3. from flask_sqlalchemy import SQLAlchemy
  4. from sqlalchemy import inspect, or_, and_
  5. from flask_wtf import FlaskForm
  6. import bcrypt
  7. from wtforms_alchemy import model_form_factory
  8. db: SQLAlchemy = SQLAlchemy()
  9. app = Flask(__name__)
  10. app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///project.db"
  11. app.secret_key = '98d31240f9fbe14c8083586db49c19c3a8d3f726'
  12. db.init_app(app)
  13. BaseModelForm = model_form_factory(FlaskForm)
  14. class ModelForm(BaseModelForm):
  15. @classmethod
  16. def get_session(self):
  17. return db.session
  18. class Admin(db.Model):
  19. id = db.Column(db.Integer, primary_key=True)
  20. username = db.Column(db.String, unique=True, nullable=False)
  21. password = db.Column(db.String, nullable=False)
  22. query: db.Query
  23. @classmethod
  24. def generate_password(cls, pw: str):
  25. return bcrypt.hashpw(pw, bcrypt.gensalt(12))
  26. @classmethod
  27. def authenticate(cls, username: str, pw: str):
  28. user = Admin.query.filter_by(username=username).one_or_none()
  29. if user and bcrypt.checkpw(pw, user.password):
  30. session['admin'] = user.username
  31. return user
  32. else:
  33. return None
  34. @classmethod
  35. def exists(cls):
  36. user = Admin.query.one_or_none()
  37. return True if user else False
  38. @classmethod
  39. def authorize(cls):
  40. if not session.get('admin'):
  41. return redirect(url_for("admin_login"))
  42. def object_as_dict(obj):
  43. return {c.key: getattr(obj, c.key)
  44. for c in inspect(obj).mapper.column_attrs}
  45. class Chemical(db.Model):
  46. query: db.Query
  47. id = db.Column(db.Integer, primary_key=True)
  48. # all fields after here are included in the database
  49. chemical_db_id = db.Column(db.String)
  50. library = db.Column(db.String)
  51. # important fields
  52. name = db.Column(db.String, nullable=False)
  53. formula = db.Column(db.String, nullable=False)
  54. mass = db.Column(db.Float, nullable=False)
  55. pubchem_cid = db.Column(db.Integer)
  56. pubmed_refcount = db.Column(db.Integer)
  57. standard_class = db.Column(db.String)
  58. inchikey = db.Column(db.String)
  59. inchikey14 = db.Column(db.String)
  60. final_mz = db.Column(db.Float, nullable=False)
  61. final_rt = db.Column(db.Float, nullable=False)
  62. final_adduct = db.Column(db.String)
  63. adduct = db.Column(db.String)
  64. detected_adducts = db.Column(db.String)
  65. adduct_calc_mz = db.Column(db.String)
  66. msms_detected = db.Column(db.Boolean)
  67. msms_purity = db.Column(db.Float)
  68. class ChemicalForm(ModelForm):
  69. class Meta:
  70. csrf = False
  71. model = Chemical
  72. # Error Handlers
  73. @app.errorhandler(404)
  74. def handler_404(msg):
  75. return render_template("errors/404.html")
  76. @app.errorhandler(403)
  77. def handler_403(msg):
  78. return render_template("errors/403.html")
  79. # Admin routes
  80. @app.route('/admin')
  81. def admin_root():
  82. if login := Admin.authorize():
  83. return login
  84. return render_template("admin.html", user=session.get("admin"))
  85. @app.route('/admin/create', methods=['GET', 'POST'])
  86. def admin_create():
  87. if Admin.exists():
  88. if login := Admin.authorize():
  89. return login
  90. if request.method == "GET":
  91. return render_template("register.html")
  92. else:
  93. username, pw = request.form.get('username'), request.form.get('password')
  94. if username is None or pw is None:
  95. return render_template("register.html", fail="Invalid Input.")
  96. elif db.session.execute(db.select(Admin).filter_by(username=username)).fetchone():
  97. return render_template("register.html", fail="Username already exists.")
  98. else:
  99. db.session.add(Admin(username=username, password=Admin.generate_password(pw)))
  100. db.session.commit()
  101. return render_template("register.html", success=True)
  102. @app.route('/admin/login', methods=['GET', 'POST'])
  103. def admin_login():
  104. if request.method == "POST":
  105. username, pw = request.form.get('username', ''), request.form.get('password', '')
  106. if Admin.authenticate(username, pw):
  107. return render_template("login.html", success=True)
  108. else:
  109. return render_template("login.html", fail="Could not authenticate.")
  110. else:
  111. return render_template("login.html")
  112. @app.route('/admin/logout', methods=['GET'])
  113. def admin_logout():
  114. session.pop('admin')
  115. return redirect(url_for('home'))
  116. @app.route("/")
  117. def home():
  118. if Admin.exists():
  119. return render_template("index.html")
  120. else:
  121. return redirect(url_for("admin_create"))
  122. # Routes for CRUD operations on chemicals
  123. @app.route("/chemical/create", methods=['GET', 'POST'])
  124. def chemical_create():
  125. if not session.get('admin'):
  126. abort(403)
  127. if request.method == "POST":
  128. form = ChemicalForm(**request.form)
  129. if form.validate():
  130. new_chemical = Chemical(**form.data)
  131. db.session.add(new_chemical)
  132. db.session.commit()
  133. return render_template("create_chemical.html", form=ChemicalForm(), success=True)
  134. else:
  135. return render_template("create_chemical.html", form=form, invalid=True)
  136. else:
  137. form = ChemicalForm()
  138. return render_template("create_chemical.html", form=form)
  139. @app.route("/chemical/<int:id>/update", methods=['GET', 'POST'])
  140. def chemical_update(id: int):
  141. if not session.get('admin'):
  142. abort(403)
  143. current_chemical:Chemical = Chemical.query.filter_by(id=id).one_or_404()
  144. dct = object_as_dict(current_chemical)
  145. if request.method == "POST":
  146. form = ChemicalForm(**request.form)
  147. if form.validate():
  148. # take the row with id and update it.
  149. for k in form.data:
  150. setattr(current_chemical, k, form.data[k])
  151. db.session.commit()
  152. return render_template("create_chemical.html", form=form, success=True, id=id)
  153. else:
  154. form = ChemicalForm(**dct)
  155. return render_template("create_chemical.html", form=form, invalid=True, id=id)
  156. else:
  157. form = ChemicalForm(**dct)
  158. return render_template("create_chemical.html", form=form, id=id)
  159. @app.route("/chemical/<int:id>/delete")
  160. def chemical_delete(id: int):
  161. if not session.get('admin'):
  162. abort(403)
  163. current_chemical:Chemical = Chemical.query.filter_by(id=id).one_or_404()
  164. db.session.delete(current_chemical)
  165. db.session.commit()
  166. return render_template("delete_chemical.html", id=id)
  167. @app.route("/chemical/<int:id>/view")
  168. def chemical_view(id: int):
  169. current_chemical:Chemical = Chemical.query.filter_by(id=id).one_or_404()
  170. dct = object_as_dict(current_chemical)
  171. return render_template("view_chemical.html", id=id, chemical=dct)
  172. @app.route("/chemical/all")
  173. def chemical_all():
  174. if not session.get('admin'):
  175. abort(403)
  176. result = Chemical.query.all()
  177. data = []
  178. for x in result:
  179. data.append({"url": url_for("chemical_view", id=x.id), "name": x.name, "mz": x.final_mz, "rt": x.final_rt})
  180. return jsonify(data)
  181. @app.route("/chemical/search")
  182. def search_api():
  183. mz_min, mz_max = request.args.get('mz_min'), request.args.get('mz_max')
  184. rt_min, rt_max = request.args.get('rt_min'), request.args.get('rt_max')
  185. if (mz_min is None and mz_max is None) or (rt_min is None and rt_max is None):
  186. abort(400)
  187. try:
  188. if mz_min is not None and mz_max is None:
  189. mz_max = float(mz_min) + 3
  190. elif mz_max is not None and mz_min is None:
  191. mz_min = float(mz_max) - 3
  192. if rt_min is not None and rt_max is None:
  193. rt_max = float(rt_min) + 3
  194. elif rt_max is not None and rt_min is None:
  195. rt_min = float(rt_max) - 3
  196. mz_min, mz_max = float(mz_min), float(mz_max)
  197. rt_min, rt_max = float(rt_min), float(rt_max)
  198. except ValueError:
  199. abort(400)
  200. mz_filter = and_(mz_max > Chemical.final_mz, Chemical.final_mz > mz_min)
  201. rt_filter = and_(rt_max > Chemical.final_rt, Chemical.final_rt > rt_min)
  202. result = Chemical.query.filter(
  203. or_(mz_filter, rt_filter)
  204. ).limit(20).all()
  205. data = []
  206. for x in result:
  207. data.append({"url": url_for("chemical_view", id=x.id), "name": x.name, "mz": x.final_mz, "rt": x.final_rt})
  208. return jsonify(data)
  209. @app.route("/search")
  210. def search():
  211. return render_template("search.html")
  212. if __name__ == "__main__":
  213. with app.app_context():
  214. db.create_all()
  215. app.run(debug=True)