Moved wait_for_db function to utils module

This commit is contained in:
Marcin-Ramotowski 2025-06-11 19:48:58 +00:00
parent 9e010ed389
commit d3d3c98f99
2 changed files with 20 additions and 20 deletions

View File

@ -4,29 +4,10 @@ from flask_jwt_extended import JWTManager
from jwt import ExpiredSignatureError from jwt import ExpiredSignatureError
from models import db, RevokedToken from models import db, RevokedToken
import os import os
from sqlalchemy import text from utils import init_db, wait_for_db
from sqlalchemy.exc import DatabaseError
import time
from utils import init_db
from views import user_bp from views import user_bp
from werkzeug.exceptions import HTTPException from werkzeug.exceptions import HTTPException
MAX_RETRIES = 100
def wait_for_db():
for retries in range(MAX_RETRIES):
try:
with db.engine.connect() as connection:
connection.execute(text("SELECT 1"))
print("Successfully connected with database.")
return
except DatabaseError:
print(f"Waiting for database... (retry {retries + 1})")
time.sleep(3)
print("Failed to connect to database.")
raise Exception("Database not ready after multiple retries.")
def create_app(config_name="default"): def create_app(config_name="default"):
"""Creates and returns a new instance of Flask app.""" """Creates and returns a new instance of Flask app."""
load_dotenv() load_dotenv()

View File

@ -2,6 +2,9 @@ from flask import abort
from flask_jwt_extended import get_jwt_identity from flask_jwt_extended import get_jwt_identity
from models import User, db from models import User, db
import os import os
from sqlalchemy import text
from sqlalchemy.exc import DatabaseError
import time
from werkzeug.security import generate_password_hash from werkzeug.security import generate_password_hash
@ -27,6 +30,22 @@ def get_user_or_404(user_id):
return user return user
MAX_RETRIES = 100
def wait_for_db():
for retries in range(MAX_RETRIES):
try:
with db.engine.connect() as connection:
connection.execute(text("SELECT 1"))
print("Successfully connected with database.")
return
except DatabaseError:
print(f"Waiting for database... (retry {retries + 1})")
time.sleep(3)
print("Failed to connect to database.")
raise Exception("Database not ready after multiple retries.")
def init_db(): def init_db():
"""Create default admin account if database is empty""" """Create default admin account if database is empty"""
with db.session.begin(): with db.session.begin():