You cannot select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
orquestrador/app/db/bootstrap.py

84 lines
2.3 KiB
Python

"""
Rotina dedicada de bootstrap de banco de dados.
Cria tabelas e executa seed inicial de forma explicita, fora do startup do app.
"""
from app.core.settings import settings
from app.db.database import Base, engine
from app.db.mock_database import MockBase, mock_engine
from app.db.models import Tool
from app.db.mock_models import (
ConversationTurn,
Customer,
IntegrationDelivery,
IntegrationRoute,
Order,
RentalContract,
RentalFine,
RentalPayment,
RentalVehicle,
ReviewSchedule,
Vehicle,
)
from app.db.mock_seed import seed_mock_data
from app.db.tool_seed import seed_tools
def bootstrap_databases(
*,
run_tools_seed: bool | None = None,
run_mock_seed: bool | None = None,
) -> None:
"""Cria tabelas e executa seed inicial em ambos os bancos."""
print("Inicializando bancos...")
failures: list[str] = []
should_seed_tools = settings.auto_seed_tools if run_tools_seed is None else bool(run_tools_seed)
should_seed_mock = (
settings.auto_seed_mock and settings.mock_seed_enabled
if run_mock_seed is None
else bool(run_mock_seed)
)
try:
print("Criando tabelas MySQL (tools)...")
Base.metadata.create_all(bind=engine)
if should_seed_tools:
print("Populando tools iniciais...")
seed_tools()
else:
print("Seed de tools desabilitada por configuracao.")
print("MySQL tools OK.")
except Exception as exc:
print(f"Aviso: falha no MySQL (tools): {exc}")
failures.append(f"tools={exc}")
try:
print("Criando tabelas MySQL (dados ficticios)...")
MockBase.metadata.create_all(bind=mock_engine)
if should_seed_mock:
print("Populando dados ficticios iniciais...")
seed_mock_data()
else:
print("Seed mock desabilitada por configuracao.")
print("MySQL mock OK.")
except Exception as exc:
print(f"Aviso: falha no MySQL mock: {exc}")
failures.append(f"mock={exc}")
if failures:
raise RuntimeError(
"Falha ao inicializar bancos do orquestrador: " + " | ".join(failures)
)
print("Bancos inicializados com sucesso!")
def main() -> None:
"""Executa o bootstrap dedicado quando chamado via modulo."""
bootstrap_databases()
if __name__ == "__main__":
main()