insert error handlers for SQlalchemy

Backend/application/engine_controller.py

@@ -8,6 +8,7 @@
 from time import gmtime, strftime
 import datetime
 import socket
+from sqlalchemy import exc
 
 
 def find_extension_by_lang(lang):
@@ -20,25 +21,30 @@ def find_extension_by_lang(lang):
  return lang
 
 def list_table_cache():
- Engine.to_dict = Engine.to_dict
- elements = Engine.query.all()
- Cache_Array = []
+ try:
+ Engine.to_dict = Engine.to_dict
+ elements = Engine.query.all()
+ Cache_Array = []
 
- for item in elements:
+  for item in elements:
  line={}
  line["rule_id"]=str(item.rule_id)
  line["title"]=str(item.title)
  line["path"]=str(item.path)
  line["lines"]=str(item.lines)
  line["lang"]=str(item.lang)
  Cache_Array.append(line)
- return jsonify(Cache_Array)
+ return jsonify(Cache_Array)
+ except exc.SQLAlchemyError as e:
+ print(e)
+ return "Error"
 
 def clear_cache_all():
  try:
  total = db.session.query(Engine).delete()
  db.session.commit()
- except:
+ except exc.SQLAlchemyError as e:
+ print(e)
  total=0
  db.session.rollback()
  return jsonify(total)
@@ -77,8 +83,9 @@ def search_sinks(directory, extension,sink):
  if extension and name.lower().endswith(extension):
  current_path=os.path.join(dirpath, name)
  Rules.to_dict = Rules.to_dict
- elements = Rules.query.filter_by(lang=lang_db) 
- for item in elements:
+ try:
+ elements = Rules.query.filter_by(lang=lang_db) 
+ for item in elements:
  if sink == 0:
  regex1=item.match1
  regex2=item.match2
@@ -103,6 +110,9 @@ def search_sinks(directory, extension,sink):
  db.session.commit()
  total+=1
  lines=0
+ except exc.SQLAlchemyError as e:
+ print(e)
+ return "Error"
  return total
 
 def getsinks():

Backend/application/rule_controller.py

@@ -6,22 +6,26 @@
 from time import gmtime, strftime
 import datetime
 import socket
+from sqlalchemy import exc
 
 def List_table_rules():
- Rules.to_dict = Rules.to_dict
- elements = Rules.query.all()
- Rules_Array = []
-
- for item in elements:
- line={}
- line["id"]=str(item.id)
- line["lang"]=str(item.lang)
- line["title"]=str(item.title)
- line["level"]=str(item.level)
- line["created_at"]=str(item.created_at)
- Rules_Array.append(line)
- return jsonify(Rules_Array)
-
+ try:
+ Rules.to_dict = Rules.to_dict
+ elements = Rules.query.all()
+ Rules_Array = []
+
+ for item in elements:
+ line={}
+ line["id"]=str(item.id)
+ line["lang"]=str(item.lang)
+ line["title"]=str(item.title)
+ line["level"]=str(item.level)
+ line["created_at"]=str(item.created_at)
+ Rules_Array.append(line)
+ return jsonify(Rules_Array)
+ except exc.SQLAlchemyError as e:
+ print(e)
+ return("Error")
 
 
 def insert_rule():
@@ -36,39 +40,51 @@ def insert_rule():
  # TODO Improve validation
  if d['lang'] is None:
  abort(400)
- url = Rules(**d)
- db.session.add(url)
- db.session.commit()
- return ("True")
+ try:
+ url = Rules(**d)
+ db.session.add(url)
+ db.session.commit()
+ return ("True")
+ except exc.SQLAlchemyError as e:
+ print(e)
+ return("False")
 
 
 def return_rule(Rules_id):
  input=str(Rules_id)
- item = Rules.query.filter_by(id=input).first()
- #Rules.append(Rules.toDict()) 
- Rules_Array=[]
- line={}
- line['id']=str(item.id)
- line['lang']=str(item.lang)
- line['title']=str(item.title)
- line['description']=str(item.description)
- line['level']=str(item.level)
- line['match1']=str(item.match1)
- line['match2']=str(item.match2)
- line['update_at']=str(item.update_at)
- line['created_t']=str(item.created_at)
- Rules_Array.append(line)
- return jsonify(Rules_Array[0])
+ try:
+ item = Rules.query.filter_by(id=input).first()
+ #Rules.append(Rules.toDict()) 
+ Rules_Array=[]
+ line={}
+ line['id']=str(item.id)
+ line['lang']=str(item.lang)
+ line['title']=str(item.title)
+ line['description']=str(item.description)
+ line['level']=str(item.level)
+ line['match1']=str(item.match1)
+ line['match2']=str(item.match2)
+ line['update_at']=str(item.update_at)
+ line['created_t']=str(item.created_at)
+ Rules_Array.append(line)
+ return jsonify(Rules_Array[0])
+ except exc.SQLAlchemyError as e:
+ print(e)
+ return("False")
 
 
 def delete_rule():
  input=request.json.get('id')
  if input is None:
- abort(400) 
- url = Rules.query.filter_by(id=input).first()
- db.session.delete(url)
- db.session.commit()
- return "True"
+ abort(400)
+ try:
+ url = Rules.query.filter_by(id=input).first()
+ db.session.delete(url)
+ db.session.commit()
+ return "True"
+ except exc.SQLAlchemyError as e:
+ print(e)
+ return("False")
 
 def update_rule():
  d={}
@@ -93,10 +109,14 @@ def update_rule():
 
  d['update_at']=datetime.datetime.now()
  url=Rules()
- db.session.query(Rules).filter(Rules.id == id).update(d) 
- db.session.flush()
- result= db.session.commit()
- return "True"
+ try:
+ db.session.query(Rules).filter(Rules.id == id).update(d) 
+ db.session.flush()
+ result= db.session.commit()
+ return "True"
+ except exc.SQLAlchemyError as e:
+ print(e)
+ return "False"