check db

app/__init__.py

@@ -18,6 +18,13 @@ jwt = JWTManager()
 mail = Mail()
 serializer = URLSafeTimedSerializer(Config.SECRET_KEY)
 
+def check_db_connection():
+    try:
+        db.session.execute(db.text('SELECT 1'))
+        return True, None
+    except Exception as e:
+        return False, str(e)
+
 def create_app():
     app = Flask(__name__, static_folder='../static', template_folder='templates')
     app.config.from_object(Config)
@@ -27,6 +34,13 @@ def create_app():
     jwt.init_app(app)
     mail.init_app(app)
 
+    # Check database connection
+    with app.app_context():
+        connection_ok, connection_error = check_db_connection()
+        if not connection_ok:
+            app.logger.error(f"Database connection failed: {connection_error}")
+            raise RuntimeError(f"Database connection failed: {connection_error}")
+
     if not os.path.exists('logs'):
         os.mkdir('logs')
 

app/routes/auth.py

 from flask import Blueprint, request, jsonify, make_response
 from flask_jwt_extended import create_access_token, create_refresh_token, jwt_required, get_jwt_identity, unset_jwt_cookies
 from app.models.user import User
-from app.utils.response import success_response, error_response
+from app.utils.response import success_response, error_response, check_db_connection
 from app import db, jwt
 import datetime
 from datetime import datetime as dtm
@@ -35,6 +35,11 @@ auth_bp = Blueprint('auth', __name__)
     }
 })
 def register():
+
+    connection_ok, connection_error = check_db_connection()
+    if not connection_ok:
+        return error_response(f"Database connection failed: {connection_error}", code=500)
+
     data = request.get_json()
     created_date = current_time
     first_name = data.get('first_name')
@@ -55,6 +60,11 @@ def register():
 
 @auth_bp.route('/login', methods=['POST'])
 def login():
+
+    connection_ok, connection_error = check_db_connection()
+    if not connection_ok:
+        return error_response(f"Database connection failed: {connection_error}", code=500)
+
     username = request.json.get('username', None)
     password = request.json.get('password', None)
     

app/routes/user.py

 from flask import Blueprint, jsonify, request
 from flask_jwt_extended import jwt_required, get_jwt_identity
 from app.models.user import User
-from app.utils.response import success_response, error_response
+from app.utils.response import success_response, error_response, check_db_connection
 from app import db, jwt
 from app.helper.Master import Master
 
@@ -10,6 +10,10 @@ user_bp = Blueprint('user', __name__)
 @user_bp.route('/profile', methods=['GET'])
 @jwt_required()
 def profile():
+    connection_ok, connection_error = check_db_connection()
+    if not connection_ok:
+        return error_response(f"Database connection failed: {connection_error}", code=500)
+
     current_user = get_jwt_identity()  # This should be a string (username)
     user = User.query.filter_by(username=current_user).first()
     
@@ -26,6 +30,10 @@ def profile():
 @user_bp.route('/all', methods=['GET'])
 @jwt_required()
 def get_users():
+    connection_ok, connection_error = check_db_connection()
+    if not connection_ok:
+        return error_response(f"Database connection failed: {connection_error}", code=500)
+
     users = User.query.all()
     users_data = [{"id": user.id, "username": user.username, "email": user.email, "role_id": [role.name for role in user.roles]} for user in users]
     return success_response(data=users_data)
@@ -33,6 +41,10 @@ def get_users():
 @user_bp.route('/<int:user_id>', methods=['GET'])
 @jwt_required()
 def get_user(user_id):
+    connection_ok, connection_error = check_db_connection()
+    if not connection_ok:
+        return error_response(f"Database connection failed: {connection_error}", code=500)
+
     user = User.query.get_or_404(user_id)
     user_data = {"id": user.id, "username": user.username, "email": user.email, "role_id": [role.name for role in user.roles]}
     return success_response(data=user_data)
@@ -40,6 +52,10 @@ def get_user(user_id):
 @user_bp.route('/create', methods=['POST'])
 @jwt_required()
 def create_user():
+    connection_ok, connection_error = check_db_connection()
+    if not connection_ok:
+        return error_response(f"Database connection failed: {connection_error}", code=500)
+
     data = request.json
     username = data.get('username')
     email = data.get('email')
@@ -64,6 +80,10 @@ def create_user():
 @user_bp.route('/roles', methods=['GET'])
 @jwt_required()
 def get_roles():
+    connection_ok, connection_error = check_db_connection()
+    if not connection_ok:
+        return error_response(f"Database connection failed: {connection_error}", code=500)
+
     roles = Role.query.all()
     roles_data = [{"id": role.id, "name": role.name, "privileges": [priv.name for priv in role.privileges]} for role in roles]
     return success_response(data=roles_data)
@@ -71,6 +91,10 @@ def get_roles():
 @user_bp.route('/roles', methods=['POST'])
 @jwt_required()
 def create_role():
+    connection_ok, connection_error = check_db_connection()
+    if not connection_ok:
+        return error_response(f"Database connection failed: {connection_error}", code=500)
+
     data = request.json
     role_name = data.get('name')
     privilege_names = data.get('privileges', [])
@@ -92,6 +116,11 @@ def create_role():
 @user_bp.route('/<int:user_id>', methods=['PUT'])
 @jwt_required()
 def update_user(user_id):
+
+    connection_ok, connection_error = check_db_connection()
+    if not connection_ok:
+        return error_response(f"Database connection failed: {connection_error}", code=500)
+
     user = User.query.get_or_404(user_id)
     data = request.json
     username = data.get('username')
@@ -118,6 +147,10 @@ def update_user(user_id):
 @user_bp.route('/roles/<int:role_id>', methods=['PUT'])
 @jwt_required()
 def update_role(role_id):
+    connection_ok, connection_error = check_db_connection()
+    if not connection_ok:
+        return error_response(f"Database connection failed: {connection_error}", code=500)
+
     role = Role.query.get_or_404(role_id)
     data = request.json
     role_name = data.get('name')

app/utils/response.py

 from flask import jsonify
+from app import db
+from sqlalchemy import text
 
 def success_response(data=None, message="Success", code=200):
     response = {
@@ -15,3 +17,11 @@ def error_response(message="Error", code=400):
         "data": None
     }
     return jsonify(response), code
+
+def check_db_connection():
+    try:
+        # A simple query to check the database connection
+        db.session.execute(text('SELECT 1'))
+        return True, None
+    except Exception as e:
+        return False, str(e)