feat: Add new scripts and update project structure for database management and user authentication
This commit introduces several new scripts for managing database operations, including user creation, permission grants, and data migrations. It also adds new documentation files to guide users through the setup and configuration processes. Additionally, the project structure is updated to enhance organization and maintainability, ensuring a smoother development experience for contributors. These changes support the ongoing transition to a web-based architecture and improve overall project functionality.
This commit is contained in:
Tanya
Executable
+78
@@ -0,0 +1,78 @@
|
||||
import sqlite3
|
||||
import sys
|
||||
import os
|
||||
|
||||
|
||||
def drop_all_tables(db_path: str) -> None:
|
||||
if not os.path.exists(db_path):
|
||||
print(f"Database not found: {db_path}")
|
||||
return
|
||||
|
||||
conn = sqlite3.connect(db_path)
|
||||
try:
|
||||
conn.isolation_level = None # autocommit mode for DDL
|
||||
cur = conn.cursor()
|
||||
|
||||
# Disable foreign key enforcement to allow dropping in any order
|
||||
cur.execute("PRAGMA foreign_keys = OFF;")
|
||||
|
||||
# Collect tables and views
|
||||
cur.execute("SELECT name, type FROM sqlite_master WHERE type IN ('table','view') AND name NOT LIKE 'sqlite_%';")
|
||||
objects = cur.fetchall()
|
||||
print(f"DB: {db_path}")
|
||||
if not objects:
|
||||
print("No user tables or views found.")
|
||||
return
|
||||
|
||||
# Drop views first, then tables
|
||||
views = [name for name, t in objects if t == 'view']
|
||||
tables = [name for name, t in objects if t == 'table']
|
||||
|
||||
print(f"Found {len(tables)} tables and {len(views)} views.")
|
||||
for v in views:
|
||||
print(f"Dropping view: {v}")
|
||||
cur.execute(f"DROP VIEW IF EXISTS \"{v}\";")
|
||||
|
||||
for t in tables:
|
||||
print(f"Dropping table: {t}")
|
||||
cur.execute(f"DROP TABLE IF EXISTS \"{t}\";")
|
||||
|
||||
# Vacuum to clean up
|
||||
cur.execute("VACUUM;")
|
||||
print("Done.")
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def list_tables(db_path: str) -> None:
|
||||
if not os.path.exists(db_path):
|
||||
print(f"Database not found: {db_path}")
|
||||
return
|
||||
conn = sqlite3.connect(db_path)
|
||||
try:
|
||||
cur = conn.cursor()
|
||||
cur.execute("SELECT name, type FROM sqlite_master WHERE type IN ('table','view') AND name NOT LIKE 'sqlite_%' ORDER BY type, name;")
|
||||
objects = cur.fetchall()
|
||||
print(f"DB: {db_path}")
|
||||
if not objects:
|
||||
print("No user tables or views found.")
|
||||
return
|
||||
for name, t in objects:
|
||||
print(f"- {t}: {name}")
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
# Usage: python drop_all_tables.py <db1> [<db2> ...]
|
||||
paths = sys.argv[1:]
|
||||
if not paths:
|
||||
base = os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))
|
||||
paths = [os.path.join(base, 'photos.db'), os.path.join(base, 'data', 'photos.db')]
|
||||
|
||||
for p in paths:
|
||||
list_tables(p)
|
||||
for p in paths:
|
||||
drop_all_tables(p)
|
||||
for p in paths:
|
||||
list_tables(p)
|
||||
@@ -0,0 +1,59 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Drop all tables from the web database to start fresh."""
|
||||
|
||||
import sys
|
||||
import os
|
||||
|
||||
# Add project root to path
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
from sqlalchemy import inspect
|
||||
from backend.db.session import engine, get_database_url
|
||||
from backend.db.models import Base
|
||||
|
||||
# Ordered list ensures foreign-key dependents drop first
|
||||
TARGET_TABLES = [
|
||||
"photo_favorites",
|
||||
"phototaglinkage",
|
||||
"person_encodings",
|
||||
"faces",
|
||||
"tags",
|
||||
"photos",
|
||||
"people",
|
||||
]
|
||||
|
||||
|
||||
def drop_all_tables():
|
||||
"""Drop all tables from the database."""
|
||||
db_url = get_database_url()
|
||||
print(f"Connecting to database: {db_url}")
|
||||
|
||||
inspector = inspect(engine)
|
||||
existing_tables = set(inspector.get_table_names())
|
||||
|
||||
print("\nDropping selected tables...")
|
||||
for table_name in TARGET_TABLES:
|
||||
if table_name not in Base.metadata.tables:
|
||||
print(f" ⚠️ Table '{table_name}' not found in metadata, skipping.")
|
||||
continue
|
||||
if table_name not in existing_tables:
|
||||
print(f" ℹ️ Table '{table_name}' does not exist in database, skipping.")
|
||||
continue
|
||||
table = Base.metadata.tables[table_name]
|
||||
print(f" 🗑️ Dropping '{table_name}'...")
|
||||
table.drop(bind=engine, checkfirst=True)
|
||||
|
||||
print("✅ Selected tables dropped successfully!")
|
||||
print("\nYou can now recreate tables using:")
|
||||
print(" python scripts/recreate_tables_web.py")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
drop_all_tables()
|
||||
except Exception as e:
|
||||
print(f"❌ Error dropping tables: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
sys.exit(1)
|
||||
|
||||
Executable
+115
@@ -0,0 +1,115 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Grant DELETE permission on auth database users table.
|
||||
|
||||
This script grants DELETE permission to the database user specified in DATABASE_URL_AUTH.
|
||||
It requires superuser access (postgres user) to grant permissions.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from urllib.parse import urlparse
|
||||
|
||||
from dotenv import load_dotenv
|
||||
from sqlalchemy import create_engine, text
|
||||
|
||||
# Load environment variables
|
||||
env_path = Path(__file__).parent.parent.parent / ".env"
|
||||
load_dotenv(dotenv_path=env_path)
|
||||
|
||||
|
||||
def parse_database_url(db_url: str) -> dict:
|
||||
"""Parse database URL into components."""
|
||||
# Handle postgresql+psycopg2:// format
|
||||
if db_url.startswith("postgresql+psycopg2://"):
|
||||
db_url = db_url.replace("postgresql+psycopg2://", "postgresql://")
|
||||
|
||||
parsed = urlparse(db_url)
|
||||
return {
|
||||
"user": parsed.username,
|
||||
"password": parsed.password,
|
||||
"host": parsed.hostname or "localhost",
|
||||
"port": parsed.port or 5432,
|
||||
"database": parsed.path.lstrip("/"),
|
||||
}
|
||||
|
||||
|
||||
def grant_delete_permission() -> None:
|
||||
"""Grant DELETE permission on users and pending_photos tables in auth database."""
|
||||
auth_db_url = os.getenv("DATABASE_URL_AUTH")
|
||||
if not auth_db_url:
|
||||
print("❌ Error: DATABASE_URL_AUTH environment variable not set")
|
||||
sys.exit(1)
|
||||
|
||||
if not auth_db_url.startswith("postgresql"):
|
||||
print("ℹ️ Auth database is not PostgreSQL. No permissions to grant.")
|
||||
return
|
||||
|
||||
db_info = parse_database_url(auth_db_url)
|
||||
db_user = db_info["user"]
|
||||
db_name = db_info["database"]
|
||||
|
||||
print(f"📋 Granting DELETE permission on auth database tables...")
|
||||
print(f" Database: {db_name}")
|
||||
print(f" User: {db_user}")
|
||||
|
||||
# Tables that need DELETE permission
|
||||
tables = ["users", "pending_photos", "pending_identifications", "inappropriate_photo_reports"]
|
||||
|
||||
# Connect as postgres superuser to grant permissions
|
||||
# Try to connect as postgres user (superuser)
|
||||
try:
|
||||
# Try to get postgres password from environment or use peer authentication
|
||||
postgres_url = f"postgresql://postgres@{db_info['host']}:{db_info['port']}/{db_name}"
|
||||
engine = create_engine(postgres_url)
|
||||
|
||||
with engine.connect() as conn:
|
||||
for table in tables:
|
||||
try:
|
||||
# Grant DELETE permission
|
||||
conn.execute(text(f"""
|
||||
GRANT DELETE ON TABLE {table} TO {db_user}
|
||||
"""))
|
||||
print(f" ✅ Granted DELETE on {table}")
|
||||
except Exception as e:
|
||||
# Table might not exist, skip it
|
||||
print(f" ⚠️ Could not grant DELETE on {table}: {e}")
|
||||
conn.commit()
|
||||
|
||||
print(f"✅ Successfully granted DELETE permissions to user '{db_user}'")
|
||||
return
|
||||
except Exception as e:
|
||||
# If connecting as postgres fails, try with the same user (might have grant privileges)
|
||||
print(f"⚠️ Could not connect as postgres user: {e}")
|
||||
print(f" Trying with current database user...")
|
||||
|
||||
try:
|
||||
engine = create_engine(auth_db_url)
|
||||
with engine.connect() as conn:
|
||||
for table in tables:
|
||||
try:
|
||||
# Try to grant permission
|
||||
conn.execute(text(f"""
|
||||
GRANT DELETE ON TABLE {table} TO {db_user}
|
||||
"""))
|
||||
print(f" ✅ Granted DELETE on {table}")
|
||||
except Exception as e2:
|
||||
print(f" ⚠️ Could not grant DELETE on {table}: {e2}")
|
||||
conn.commit()
|
||||
|
||||
print(f"✅ Successfully granted DELETE permissions to user '{db_user}'")
|
||||
return
|
||||
except Exception as e2:
|
||||
print(f"❌ Failed to grant permission: {e2}")
|
||||
print(f"\n💡 To grant permission manually, run as postgres superuser:")
|
||||
for table in tables:
|
||||
print(f" sudo -u postgres psql -d {db_name} -c \"GRANT DELETE ON TABLE {table} TO {db_user};\"")
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
grant_delete_permission()
|
||||
|
||||
|
||||
@@ -0,0 +1,264 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Migrate data from SQLite to PostgreSQL database.
|
||||
|
||||
This script:
|
||||
1. Creates PostgreSQL databases if they don't exist
|
||||
2. Creates all tables in PostgreSQL
|
||||
3. Migrates all data from SQLite to PostgreSQL
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
# Add project root to path
|
||||
project_root = Path(__file__).parent.parent
|
||||
sys.path.insert(0, str(project_root))
|
||||
|
||||
from sqlalchemy import create_engine, inspect, text
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
from backend.db.models import Base
|
||||
from backend.db.session import get_database_url
|
||||
import sqlite3
|
||||
|
||||
def create_postgresql_databases():
|
||||
"""Create PostgreSQL databases if they don't exist."""
|
||||
from urllib.parse import urlparse
|
||||
|
||||
# Get database URLs from environment
|
||||
db_url = os.getenv("DATABASE_URL", "postgresql+psycopg2://punimtag:punimtag_password@localhost:5432/punimtag")
|
||||
auth_db_url = os.getenv("DATABASE_URL_AUTH", "postgresql://punimtag:punimtag_password@localhost:5432/punimtag_auth")
|
||||
|
||||
# Parse URLs
|
||||
main_parsed = urlparse(db_url.replace("postgresql+psycopg2://", "postgresql://"))
|
||||
auth_parsed = urlparse(auth_db_url.replace("postgresql+psycopg2://", "postgresql://"))
|
||||
|
||||
main_db_name = main_parsed.path.lstrip("/")
|
||||
auth_db_name = auth_parsed.path.lstrip("/")
|
||||
|
||||
# Connect to postgres database to create other databases
|
||||
postgres_url = f"postgresql://{main_parsed.username}:{main_parsed.password}@{main_parsed.hostname}:{main_parsed.port or 5432}/postgres"
|
||||
|
||||
try:
|
||||
engine = create_engine(postgres_url)
|
||||
with engine.connect() as conn:
|
||||
# Check if databases exist
|
||||
result = conn.execute(text("SELECT 1 FROM pg_database WHERE datname = :name"), {"name": main_db_name})
|
||||
if not result.fetchone():
|
||||
conn.execute(text("COMMIT")) # End any transaction
|
||||
conn.execute(text(f'CREATE DATABASE "{main_db_name}"'))
|
||||
print(f"✅ Created database: {main_db_name}")
|
||||
else:
|
||||
print(f"✅ Database already exists: {main_db_name}")
|
||||
|
||||
result = conn.execute(text("SELECT 1 FROM pg_database WHERE datname = :name"), {"name": auth_db_name})
|
||||
if not result.fetchone():
|
||||
conn.execute(text("COMMIT"))
|
||||
conn.execute(text(f'CREATE DATABASE "{auth_db_name}"'))
|
||||
print(f"✅ Created database: {auth_db_name}")
|
||||
else:
|
||||
print(f"✅ Database already exists: {auth_db_name}")
|
||||
except Exception as e:
|
||||
print(f"⚠️ Error creating databases: {e}")
|
||||
print(" Make sure PostgreSQL is running and credentials are correct")
|
||||
|
||||
def migrate_data():
|
||||
"""Migrate data from SQLite to PostgreSQL."""
|
||||
print("=" * 80)
|
||||
print("MIGRATING DATA FROM SQLITE TO POSTGRESQL")
|
||||
print("=" * 80)
|
||||
|
||||
# Get database URLs
|
||||
sqlite_url = "sqlite:///data/punimtag.db"
|
||||
postgres_url = os.getenv("DATABASE_URL", "postgresql+psycopg2://punimtag:punimtag_password@localhost:5432/punimtag")
|
||||
|
||||
if not postgres_url.startswith("postgresql"):
|
||||
print("❌ DATABASE_URL is not set to PostgreSQL")
|
||||
print(" Set DATABASE_URL in .env file to PostgreSQL connection string")
|
||||
return False
|
||||
|
||||
# Connect to both databases
|
||||
sqlite_engine = create_engine(sqlite_url)
|
||||
postgres_engine = create_engine(postgres_url)
|
||||
|
||||
# Create tables in PostgreSQL
|
||||
print("\n📋 Creating tables in PostgreSQL...")
|
||||
Base.metadata.create_all(bind=postgres_engine)
|
||||
print("✅ Tables created")
|
||||
|
||||
# Get table names
|
||||
inspector = inspect(sqlite_engine)
|
||||
all_tables = inspector.get_table_names()
|
||||
|
||||
# Exclude system tables
|
||||
all_tables = [t for t in all_tables if not t.startswith("sqlite_")]
|
||||
|
||||
# Define migration order (respecting foreign key constraints)
|
||||
# Tables with no dependencies first, then dependent tables
|
||||
migration_order = [
|
||||
"alembic_version", # Migration tracking (optional)
|
||||
"photos", # Base table
|
||||
"people", # Base table
|
||||
"tags", # Base table
|
||||
"users", # Base table
|
||||
"faces", # Depends on photos, people, users
|
||||
"person_encodings", # Depends on people, faces
|
||||
"phototaglinkage", # Depends on photos, tags
|
||||
"photo_favorites", # Depends on photos
|
||||
"photo_person_linkage", # Depends on photos, people, users
|
||||
"role_permissions", # Base table
|
||||
]
|
||||
|
||||
# Filter to only tables that exist
|
||||
tables = [t for t in migration_order if t in all_tables]
|
||||
# Add any remaining tables not in the order list
|
||||
for t in all_tables:
|
||||
if t not in tables:
|
||||
tables.append(t)
|
||||
|
||||
print(f"\n📊 Found {len(tables)} tables to migrate: {', '.join(tables)}")
|
||||
|
||||
# Boolean columns mapping (SQLite stores as integer, PostgreSQL needs boolean)
|
||||
boolean_columns = {
|
||||
"photos": ["processed"],
|
||||
"faces": ["is_primary_encoding", "excluded"],
|
||||
"users": ["is_active", "is_admin", "password_change_required"],
|
||||
"role_permissions": ["allowed"],
|
||||
}
|
||||
|
||||
# Columns that might be missing in SQLite but required in PostgreSQL
|
||||
# Map: table_name -> {column: default_value}
|
||||
default_values = {
|
||||
"photos": {"file_hash": "migrated"}, # file_hash might be missing in old SQLite
|
||||
}
|
||||
|
||||
# Migrate each table
|
||||
with sqlite_engine.connect() as sqlite_conn, postgres_engine.connect() as postgres_conn:
|
||||
for table in tables:
|
||||
print(f"\n🔄 Migrating table: {table}")
|
||||
|
||||
# Get row count
|
||||
count_result = sqlite_conn.execute(text(f"SELECT COUNT(*) FROM {table}"))
|
||||
row_count = count_result.scalar()
|
||||
|
||||
if row_count == 0:
|
||||
print(f" ⏭️ Table is empty, skipping")
|
||||
continue
|
||||
|
||||
print(f" 📦 {row_count} rows to migrate")
|
||||
|
||||
# Check if table already has data in PostgreSQL
|
||||
try:
|
||||
pg_count_result = postgres_conn.execute(text(f'SELECT COUNT(*) FROM "{table}"'))
|
||||
pg_count = pg_count_result.scalar()
|
||||
if pg_count > 0:
|
||||
print(f" ⚠️ Table already has {pg_count} rows in PostgreSQL")
|
||||
# Auto-truncate for non-interactive mode, or ask in interactive
|
||||
print(f" 🗑️ Truncating existing data...")
|
||||
postgres_conn.execute(text(f'TRUNCATE TABLE "{table}" CASCADE'))
|
||||
postgres_conn.commit()
|
||||
except Exception as e:
|
||||
# Table might not exist yet, that's OK
|
||||
pass
|
||||
|
||||
# Get column names and types from SQLite
|
||||
columns_result = sqlite_conn.execute(text(f"PRAGMA table_info({table})"))
|
||||
column_info = columns_result.fetchall()
|
||||
sqlite_columns = [row[1] for row in column_info]
|
||||
|
||||
# Get PostgreSQL column names
|
||||
pg_inspector = inspect(postgres_engine)
|
||||
pg_columns_info = pg_inspector.get_columns(table)
|
||||
pg_columns = [col['name'] for col in pg_columns_info]
|
||||
|
||||
# Use PostgreSQL columns (they're the source of truth)
|
||||
columns = pg_columns
|
||||
|
||||
# Get boolean columns for this table
|
||||
table_bool_cols = boolean_columns.get(table, [])
|
||||
|
||||
# Get default values for missing columns
|
||||
table_defaults = default_values.get(table, {})
|
||||
|
||||
# Build SELECT statement for SQLite (only select columns that exist)
|
||||
select_cols = [col for col in columns if col in sqlite_columns]
|
||||
select_sql = f"SELECT {', '.join(select_cols)} FROM {table}"
|
||||
|
||||
# Fetch all data
|
||||
data_result = sqlite_conn.execute(text(select_sql))
|
||||
rows = data_result.fetchall()
|
||||
|
||||
# Insert into PostgreSQL
|
||||
inserted = 0
|
||||
for row in rows:
|
||||
try:
|
||||
# Build insert statement with boolean conversion
|
||||
values = {}
|
||||
for i, col in enumerate(select_cols):
|
||||
val = row[i]
|
||||
# Convert integer booleans to Python booleans for PostgreSQL
|
||||
if col in table_bool_cols:
|
||||
val = bool(val) if val is not None else None
|
||||
values[col] = val
|
||||
|
||||
# Add default values for missing columns
|
||||
for col, default_val in table_defaults.items():
|
||||
if col not in values and col in columns:
|
||||
values[col] = default_val
|
||||
|
||||
# Only insert columns we have values for (that exist in PostgreSQL)
|
||||
insert_cols = [col for col in columns if col in values]
|
||||
cols_str = ', '.join([f'"{c}"' for c in insert_cols])
|
||||
placeholders = ', '.join([f':{c}' for c in insert_cols])
|
||||
insert_sql = f'INSERT INTO "{table}" ({cols_str}) VALUES ({placeholders})'
|
||||
|
||||
postgres_conn.execute(text(insert_sql), values)
|
||||
inserted += 1
|
||||
|
||||
if inserted % 100 == 0:
|
||||
postgres_conn.commit()
|
||||
print(f" ✅ Inserted {inserted}/{row_count} rows...", end='\r')
|
||||
|
||||
except Exception as e:
|
||||
print(f"\n ❌ Error inserting row: {e}")
|
||||
print(f" Row data: {dict(zip(columns, row))}")
|
||||
postgres_conn.rollback()
|
||||
break
|
||||
|
||||
postgres_conn.commit()
|
||||
print(f" ✅ Migrated {inserted}/{row_count} rows from {table}")
|
||||
|
||||
print("\n" + "=" * 80)
|
||||
print("✅ MIGRATION COMPLETE")
|
||||
print("=" * 80)
|
||||
print("\nNext steps:")
|
||||
print("1. Update .env file to use PostgreSQL:")
|
||||
print(" DATABASE_URL=postgresql+psycopg2://punimtag:punimtag_password@localhost:5432/punimtag")
|
||||
print("2. Restart the backend API")
|
||||
print("3. Restart the viewer frontend")
|
||||
print("4. Verify data in viewer frontend")
|
||||
|
||||
return True
|
||||
|
||||
if __name__ == "__main__":
|
||||
print("🔧 SQLite to PostgreSQL Migration Tool\n")
|
||||
|
||||
# Check if SQLite database exists
|
||||
sqlite_path = project_root / "data" / "punimtag.db"
|
||||
if not sqlite_path.exists():
|
||||
print(f"❌ SQLite database not found: {sqlite_path}")
|
||||
sys.exit(1)
|
||||
|
||||
print(f"✅ Found SQLite database: {sqlite_path}")
|
||||
|
||||
# Create PostgreSQL databases
|
||||
print("\n📦 Creating PostgreSQL databases...")
|
||||
create_postgresql_databases()
|
||||
|
||||
# Migrate data
|
||||
print("\n")
|
||||
migrate_data()
|
||||
|
||||
@@ -0,0 +1,35 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Recreate all tables from models (fresh start)."""
|
||||
|
||||
import sys
|
||||
import os
|
||||
|
||||
# Add project root to path
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
from backend.db.models import Base
|
||||
from backend.db.session import engine, get_database_url
|
||||
|
||||
|
||||
def recreate_tables():
|
||||
"""Recreate all tables from models."""
|
||||
db_url = get_database_url()
|
||||
print(f"Connecting to database: {db_url}")
|
||||
|
||||
# Create all tables from models
|
||||
print("\nCreating all tables from models...")
|
||||
Base.metadata.create_all(bind=engine)
|
||||
|
||||
print("✅ All tables created successfully!")
|
||||
print("✅ Database is now fresh and ready to use!")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
recreate_tables()
|
||||
except Exception as e:
|
||||
print(f"❌ Error recreating tables: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
sys.exit(1)
|
||||
|
||||
@@ -0,0 +1,129 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Show all tables and their structures in the database."""
|
||||
|
||||
import sys
|
||||
import os
|
||||
|
||||
# Add project root to path
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
from sqlalchemy import inspect, text
|
||||
from backend.db.session import engine, get_database_url
|
||||
from backend.db.models import Base
|
||||
|
||||
|
||||
def show_table_structure(table_name: str, inspector):
|
||||
"""Show the structure of a table."""
|
||||
print(f"\n{'='*80}")
|
||||
print(f"Table: {table_name}")
|
||||
print(f"{'='*80}")
|
||||
|
||||
# Get columns
|
||||
columns = inspector.get_columns(table_name)
|
||||
print("\nColumns:")
|
||||
print(f"{'Name':<30} {'Type':<25} {'Nullable':<10} {'Primary Key':<12} {'Default'}")
|
||||
print("-" * 100)
|
||||
|
||||
for col in columns:
|
||||
col_type = str(col['type'])
|
||||
nullable = "Yes" if col['nullable'] else "No"
|
||||
primary_key = "Yes" if col.get('primary_key', False) else "No"
|
||||
default = str(col.get('default', ''))[:30] if col.get('default') else ''
|
||||
print(f"{col['name']:<30} {col_type:<25} {nullable:<10} {primary_key:<12} {default}")
|
||||
|
||||
# Get indexes
|
||||
indexes = inspector.get_indexes(table_name)
|
||||
if indexes:
|
||||
print("\nIndexes:")
|
||||
for idx in indexes:
|
||||
unique = "UNIQUE" if idx.get('unique', False) else ""
|
||||
columns_str = ", ".join(idx['column_names'])
|
||||
print(f" {idx['name']}: {columns_str} {unique}")
|
||||
|
||||
# Get foreign keys
|
||||
foreign_keys = inspector.get_foreign_keys(table_name)
|
||||
if foreign_keys:
|
||||
print("\nForeign Keys:")
|
||||
for fk in foreign_keys:
|
||||
constrained_cols = ", ".join(fk['constrained_columns'])
|
||||
referred_table = fk['referred_table']
|
||||
referred_cols = ", ".join(fk['referred_columns'])
|
||||
print(f" {constrained_cols} -> {referred_table}({referred_cols})")
|
||||
|
||||
|
||||
def show_all_tables():
|
||||
"""Show all tables and their structures."""
|
||||
db_url = get_database_url()
|
||||
print(f"Database: {db_url}")
|
||||
print(f"\n{'='*80}")
|
||||
|
||||
# Create inspector
|
||||
inspector = inspect(engine)
|
||||
|
||||
# Get all table names
|
||||
table_names = inspector.get_table_names()
|
||||
|
||||
if not table_names:
|
||||
print("No tables found in database.")
|
||||
print("\nTables should be created on web app startup.")
|
||||
print("\nHere are the table structures from models:")
|
||||
|
||||
# Show from models instead
|
||||
from backend.db.models import Photo, Person, Face, PersonEmbedding, Tag, PhotoTag
|
||||
|
||||
models = [
|
||||
("photos", Photo),
|
||||
("people", Person),
|
||||
("faces", Face),
|
||||
("person_embeddings", PersonEmbedding),
|
||||
("tags", Tag),
|
||||
("photo_tags", PhotoTag),
|
||||
]
|
||||
|
||||
for table_name, model in models:
|
||||
print(f"\n{'='*80}")
|
||||
print(f"Table: {table_name}")
|
||||
print(f"{'='*80}")
|
||||
print("\nColumns:")
|
||||
for col in model.__table__.columns:
|
||||
nullable = "Yes" if col.nullable else "No"
|
||||
primary_key = "Yes" if col.primary_key else "No"
|
||||
default = str(col.default) if col.default else ''
|
||||
print(f" {col.name:<30} {col.type!s:<25} Nullable: {nullable:<10} PK: {primary_key:<12} Default: {default}")
|
||||
|
||||
# Show indexes
|
||||
indexes = model.__table__.indexes
|
||||
if indexes:
|
||||
print("\nIndexes:")
|
||||
for idx in indexes:
|
||||
unique = "UNIQUE" if idx.unique else ""
|
||||
cols = ", ".join([c.name for c in idx.columns])
|
||||
print(f" {idx.name}: {cols} {unique}")
|
||||
|
||||
# Show foreign keys
|
||||
fks = [fk for fk in model.__table__.foreign_keys]
|
||||
if fks:
|
||||
print("\nForeign Keys:")
|
||||
for fk in fks:
|
||||
print(f" {fk.parent.name} -> {fk.column.table.name}({fk.column.name})")
|
||||
|
||||
return
|
||||
|
||||
print(f"\nFound {len(table_names)} table(s):")
|
||||
for table_name in sorted(table_names):
|
||||
print(f" - {table_name}")
|
||||
|
||||
# Show structure for each table
|
||||
for table_name in sorted(table_names):
|
||||
show_table_structure(table_name, inspector)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
show_all_tables()
|
||||
except Exception as e:
|
||||
print(f"❌ Error showing tables: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
sys.exit(1)
|
||||
|
||||
Reference in New Issue
Block a user