92 lines
3.4 KiB
Python
92 lines
3.4 KiB
Python
import os
|
|
import sys
|
|
import django
|
|
from django.core.management import call_command
|
|
from django.db import connection, OperationalError
|
|
from django.apps import apps
|
|
from io import StringIO
|
|
import psycopg2
|
|
|
|
# Set up Django environment
|
|
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "leadsdb.settings")
|
|
django.setup()
|
|
|
|
def get_model_fields(model):
|
|
return {f.name: f.db_column or f.name for f in model._meta.fields}
|
|
|
|
def get_db_fields(table_name):
|
|
try:
|
|
with connection.cursor() as cursor:
|
|
cursor.execute(f"SELECT column_name, data_type FROM information_schema.columns WHERE table_name = '{table_name}'")
|
|
return {row[0]: row[1] for row in cursor.fetchall()}
|
|
except OperationalError as e:
|
|
print(f"Error connecting to the database: {e}")
|
|
print("Please check if the PostgreSQL server is running and the connection details are correct.")
|
|
sys.exit(1)
|
|
|
|
def check_migrations():
|
|
out = StringIO()
|
|
call_command('showmigrations', stdout=out)
|
|
return out.getvalue()
|
|
|
|
def check_postgres_connection():
|
|
try:
|
|
connection.ensure_connection()
|
|
print("Successfully connected to the PostgreSQL database on barad-dur.")
|
|
except OperationalError as e:
|
|
print(f"Error connecting to the PostgreSQL database on barad-dur: {e}")
|
|
print("Please check if the PostgreSQL server is running on barad-dur.")
|
|
print("You can check the status with the following command:")
|
|
print("ssh barad-dur.lan.mgk.one systemctl --user status leads-pgsql-db")
|
|
sys.exit(1)
|
|
|
|
def main():
|
|
print("Checking PostgreSQL connection to barad-dur...")
|
|
check_postgres_connection()
|
|
|
|
print("\nChecking database schema against models...")
|
|
|
|
# Check migrations
|
|
migrations = check_migrations()
|
|
if '[X]' not in migrations:
|
|
print("WARNING: There are unapplied migrations. Run 'python manage.py migrate' to apply them.")
|
|
print(migrations)
|
|
else:
|
|
print("All migrations are applied.")
|
|
|
|
inconsistencies = []
|
|
|
|
for model in apps.get_models():
|
|
if model._meta.app_label != 'leads':
|
|
continue # Skip non-leads models
|
|
|
|
print(f"\nChecking model: {model.__name__}")
|
|
table_name = model._meta.db_table
|
|
model_fields = get_model_fields(model)
|
|
db_fields = get_db_fields(table_name)
|
|
|
|
print(f"Model fields: {model_fields}")
|
|
print(f"Database fields: {db_fields}")
|
|
|
|
# Check for fields in model but not in DB
|
|
for field_name, db_column in model_fields.items():
|
|
if db_column not in db_fields:
|
|
inconsistencies.append(f"Field '{field_name}' (DB column: '{db_column}') in model '{model.__name__}' is not in the database table '{table_name}'")
|
|
else:
|
|
print(f"Field '{field_name}' (DB column: '{db_column}') exists in both model and database.")
|
|
|
|
# Check for fields in DB but not in model
|
|
for db_field, data_type in db_fields.items():
|
|
if db_field not in model_fields.values():
|
|
inconsistencies.append(f"Column '{db_field}' ({data_type}) in table '{table_name}' is not in the model '{model.__name__}'")
|
|
|
|
if inconsistencies:
|
|
print("\nThe following inconsistencies were found:")
|
|
for inconsistency in inconsistencies:
|
|
print(f"- {inconsistency}")
|
|
else:
|
|
print("\nNo inconsistencies found. The database schema matches the models.")
|
|
|
|
if __name__ == "__main__":
|
|
main()
|