87 lines
3.0 KiB
Python
87 lines
3.0 KiB
Python
"""Verify DB initialization results: enums, roles, admin user, pricing_settings."""
|
|
from __future__ import annotations
|
|
import logging
|
|
from sqlalchemy import create_engine, text
|
|
from config.database import DATABASE_URL
|
|
|
|
logging.basicConfig(level=logging.INFO)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
ENUMS = [
|
|
'miningoperationtype',
|
|
'scenariostatus',
|
|
'financialcategory',
|
|
'costbucket',
|
|
'distributiontype',
|
|
'stochasticvariable',
|
|
'resourcetype',
|
|
]
|
|
|
|
SQL_CHECK_ENUM = "SELECT typname FROM pg_type WHERE typname = ANY(:names)"
|
|
SQL_ROLES = "SELECT id, name, display_name FROM roles ORDER BY id"
|
|
SQL_ADMIN = "SELECT id, email, username, is_active, is_superuser FROM users WHERE id = 1"
|
|
SQL_USER_ROLES = "SELECT user_id, role_id, granted_by FROM user_roles WHERE user_id = 1"
|
|
SQL_PRICING = "SELECT id, slug, name, default_currency FROM pricing_settings WHERE slug = 'default'"
|
|
|
|
|
|
def run():
|
|
engine = create_engine(DATABASE_URL, future=True)
|
|
with engine.connect() as conn:
|
|
print('Using DATABASE_URL:', DATABASE_URL)
|
|
# enums
|
|
res = conn.execute(text(SQL_CHECK_ENUM), dict(names=ENUMS)).fetchall()
|
|
found = [r[0] for r in res]
|
|
print('\nEnums found:')
|
|
for name in ENUMS:
|
|
print(f' {name}:', 'YES' if name in found else 'NO')
|
|
|
|
# roles
|
|
try:
|
|
roles = conn.execute(text(SQL_ROLES)).fetchall()
|
|
print('\nRoles:')
|
|
if roles:
|
|
for r in roles:
|
|
print(f' id={r.id} name={r.name} display_name={r.display_name}')
|
|
else:
|
|
print(' (no roles found)')
|
|
except Exception as e:
|
|
print('\nRoles query failed:', e)
|
|
|
|
# admin user
|
|
try:
|
|
admin = conn.execute(text(SQL_ADMIN)).fetchone()
|
|
print('\nAdmin user:')
|
|
if admin:
|
|
print(f' id={admin.id} email={admin.email} username={admin.username} is_active={admin.is_active} is_superuser={admin.is_superuser}')
|
|
else:
|
|
print(' (admin user not found)')
|
|
except Exception as e:
|
|
print('\nAdmin query failed:', e)
|
|
|
|
# user_roles
|
|
try:
|
|
ur = conn.execute(text(SQL_USER_ROLES)).fetchall()
|
|
print('\nUser roles for user_id=1:')
|
|
if ur:
|
|
for row in ur:
|
|
print(f' user_id={row.user_id} role_id={row.role_id} granted_by={row.granted_by}')
|
|
else:
|
|
print(' (no user_roles rows for user_id=1)')
|
|
except Exception as e:
|
|
print('\nUser_roles query failed:', e)
|
|
|
|
# pricing settings
|
|
try:
|
|
p = conn.execute(text(SQL_PRICING)).fetchone()
|
|
print('\nPricing settings (slug=default):')
|
|
if p:
|
|
print(f' id={p.id} slug={p.slug} name={p.name} default_currency={p.default_currency}')
|
|
else:
|
|
print(' (default pricing settings not found)')
|
|
except Exception as e:
|
|
print('\nPricing query failed:', e)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
run()
|