Spaces:
Sleeping
Sleeping
File size: 3,169 Bytes
a686b1b |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 |
"""
Script para executar migracao 003: Adicionar metadata JSONB.
Uso:
python scripts/run_migration_003.py
"""
import sys
import os
from pathlib import Path
# Adicionar src ao path
sys.path.insert(0, str(Path(__file__).parent.parent))
from src.database import DatabaseManager
from src.config import DATABASE_URL
def run_migration():
"""Executa migracao 003."""
print("Executando migracao 003: Adicionar metadata JSONB...")
# Conectar ao banco
db = DatabaseManager(DATABASE_URL)
# Ler arquivo de migracao
migration_file = Path(__file__).parent.parent / "db" / "migrations" / "003_add_metadata.sql"
if not migration_file.exists():
print(f"Erro: Arquivo de migracao nao encontrado: {migration_file}")
return False
with open(migration_file, 'r', encoding='utf-8') as f:
migration_sql = f.read()
# Executar migracao
try:
with db.get_connection() as conn:
with conn.cursor() as cur:
# Executar cada statement
statements = [s.strip() for s in migration_sql.split(';') if s.strip() and not s.strip().startswith('--')]
for i, statement in enumerate(statements, 1):
if statement:
print(f" Executando statement {i}/{len(statements)}...")
cur.execute(statement)
conn.commit()
print("Migracao 003 executada com sucesso!")
return True
except Exception as e:
print(f"Erro ao executar migracao: {e}")
return False
def verify_migration():
"""Verifica se migracao foi aplicada."""
print("\nVerificando migracao...")
db = DatabaseManager(DATABASE_URL)
try:
with db.get_connection() as conn:
with conn.cursor() as cur:
# Verificar se coluna existe
cur.execute("""
SELECT column_name
FROM information_schema.columns
WHERE table_name = 'documents'
AND column_name = 'metadata'
""")
if cur.fetchone():
print(" Coluna 'metadata' existe")
# Verificar indices
cur.execute("""
SELECT indexname
FROM pg_indexes
WHERE tablename = 'documents'
AND indexname LIKE '%metadata%'
""")
indices = [row[0] for row in cur.fetchall()]
print(f" Indices criados: {len(indices)}")
for idx in indices:
print(f" - {idx}")
print("\nMigracao verificada com sucesso!")
return True
else:
print(" Erro: Coluna 'metadata' nao encontrada")
return False
except Exception as e:
print(f"Erro ao verificar migracao: {e}")
return False
if __name__ == "__main__":
success = run_migration()
if success:
verify_migration()
else:
sys.exit(1)
|