46 lines
1.1 KiB
Python
46 lines
1.1 KiB
Python
import sys
|
|
import os
|
|
|
|
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
|
|
|
|
|
|
|
|
import os
|
|
import db
|
|
import logging
|
|
import sys
|
|
from migration import alembic_scipt as alembic
|
|
from sqlalchemy import inspect
|
|
|
|
session = db.session
|
|
engine = db.engine
|
|
|
|
def test_column(engine, tablename, row):
|
|
try:
|
|
inspector = inspect(engine)
|
|
columns = [col["name"] for col in inspector.get_columns(tablename)]
|
|
return row in columns
|
|
except Exception as e:
|
|
logging.error(e)
|
|
|
|
|
|
def test_table(table, columns):
|
|
for column in columns:
|
|
try:
|
|
exists = test_column(engine, table, column)
|
|
if exists == False:
|
|
logging.error(f"DB has not all columns. {column} is missing")
|
|
print(f"Column {column} is missing")
|
|
sys.exit()
|
|
os._exit(1)
|
|
return False
|
|
else:
|
|
logging.info("Column fine")
|
|
except Exception as e:
|
|
logging.error(e)
|
|
return True
|
|
|
|
def migration_check():
|
|
if alembic.check_db() == False:
|
|
print("migrating")
|
|
logging.info("DB is migrating") |