File size: 3,875 Bytes
b31dde2 |
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 109 110 111 112 113 114 115 116 117 118 119 120 |
# -*- coding: utf-8 -*-
"""
init_db.py — Inicializa o banco ATUAL (roteado pelo db_router/banco).
- Cria/atualiza o schema (Base.metadata.create_all) no banco selecionado.
- Realiza seed idempotente dos usuários padrão com hash via utils_seguranca.
- Não usa caminhos fixos; respeita o banco escolhido na UI (prod/test/treinamento).
Uso:
- Execute este arquivo diretamente, ou
- Importe e chame run() a partir do app, após selecionar o banco.
"""
from __future__ import annotations
from datetime import datetime
from sqlalchemy import text
from sqlalchemy.exc import SQLAlchemyError
# 👇 Usa as fábricas dinâmicas do módulo banco (que delegam ao db_router)
from banco import get_engine, init_schema, SessionLocal, db_info
# Importa seus modelos para garantir que o metadata conheça todas as tabelas
import models # noqa: F401 (garante o registro das classes no metadata)
# Utilitário de hash (bcrypt) já existente no seu projeto
from utils_seguranca import gerar_hash_senha
def _print_header():
try:
info = db_info()
print("===== Application Startup — init_db.run() =====")
print(f"📦 Router ativo? {info.get('using_router')}")
label = info.get("label") or info.get("choice") or "(desconhecido)"
print(f"🏷️ Banco selecionado: {label}")
print(f"🔗 URL: {info.get('url')}")
except Exception as e:
print(f"ℹ️ db_info indisponível: {e}")
def _seed_usuarios(db):
"""
Insere usuários padrão se não existirem.
Usa gerar_hash_senha(senha_plana) -> hash (bcrypt).
"""
from models import Usuario # import interno para evitar custos em startup
usuarios_padrao = [
("admin", "admin123", "admin"),
("usuario", "usuario123", "usuario"),
("consulta", "consulta123", "consulta"),
]
for nome, senha_plana, perfil in usuarios_padrao:
try:
existe = db.query(Usuario).filter(Usuario.usuario == nome).first()
except Exception:
# Se a tabela ainda não estiver criada (latência em provedor), reforça schema
init_schema()
existe = db.query(Usuario).filter(Usuario.usuario == nome).first()
if not existe:
try:
senha_hash = gerar_hash_senha(senha_plana)
except Exception as e:
print(f"❌ Falha ao gerar hash para '{nome}': {e}")
continue
novo = Usuario(
usuario=nome,
senha=senha_hash, # ⚠️ Armazena HASH, nunca em texto puro
perfil=perfil,
ativo=True,
data_criacao=datetime.utcnow(),
nome=nome.capitalize(),
email=None, # ajuste se quiser e-mail padrão
)
db.add(novo)
print(f"✅ Usuário '{nome}' criado")
else:
print(f"ℹ️ Usuário '{nome}' já existe")
def run():
"""
Executa a inicialização do banco atual:
- Cria/atualiza schema
- Insere usuários padrão (idempotente)
"""
_print_header()
# 1) Cria/atualiza schema no banco ATUAL
try:
init_schema()
except Exception as e:
print(f"❌ Falha ao criar/atualizar schema: {e}")
# ainda assim tenta seguir, pois alguns provedores aplicam lazy
pass
# 2) Seed de usuários
db = SessionLocal()
try:
_seed_usuarios(db)
db.commit()
print("✅ Banco inicializado com sucesso!")
except SQLAlchemyError as e:
db.rollback()
print(f"❌ Erro SQL ao inicializar banco: {e}")
except Exception as e:
db.rollback()
print(f"❌ Erro ao inicializar banco: {e}")
finally:
try:
db.close()
except Exception:
pass
if __name__ == "__main__":
run() |