Spaces:
Runtime error
Runtime error
File size: 3,886 Bytes
53fdb6d |
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 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 |
"""
Router de testes de database
Endpoints para verificar conexão e estatísticas do banco
"""
from fastapi import APIRouter, HTTPException
from api.models.responses import DatabaseStatsResponse, MessageResponse
from api.utils.logger import setup_logger
router = APIRouter()
logger = setup_logger(__name__)
@router.get("/health")
async def database_health():
"""
**Verifica saúde da conexão com banco de dados.**
Retorna status da conexão e informações básicas.
"""
try:
from database.db_manager import get_db_manager
db = get_db_manager()
is_healthy = db.health_check()
if is_healthy:
return {
"status": "healthy",
"message": "Database connection OK",
"database_url": db.get_connection_info()
}
else:
return {
"status": "unhealthy",
"message": "Database connection failed"
}
except Exception as e:
logger.error(f"Database health check error: {e}")
return {
"status": "error",
"message": str(e)
}
@router.get("/stats", response_model=DatabaseStatsResponse)
async def database_stats():
"""
**Estatísticas do banco de dados.**
Retorna contagem de registros nas principais tabelas.
"""
try:
from database.db_manager import get_db_manager
db = get_db_manager()
stats = {
"total_acordaos": db.count_records("acordaos"),
"total_tribunais": db.count_records("tribunais"),
"total_usuarios": db.count_records("usuarios"),
"total_decisoes": db.count_records("decisoes"),
}
# Tentar pegar tamanho do banco
try:
size_mb = db.get_database_size_mb()
stats["database_size_mb"] = size_mb
except:
stats["database_size_mb"] = None
return DatabaseStatsResponse(**stats)
except Exception as e:
logger.error(f"Error getting database stats: {e}")
raise HTTPException(status_code=500, detail=str(e))
@router.get("/tables")
async def list_tables():
"""
**Lista todas as tabelas do banco.**
Útil para verificar se DDL foi executado corretamente.
"""
try:
from database.db_manager import get_db_manager
db = get_db_manager()
tables = db.list_tables()
return {
"tables": tables,
"total": len(tables)
}
except Exception as e:
logger.error(f"Error listing tables: {e}")
raise HTTPException(status_code=500, detail=str(e))
@router.post("/test-insert")
async def test_insert():
"""
**Testa inserção no banco de dados.**
Insere um registro de teste e depois remove.
Útil para verificar permissões de escrita.
"""
try:
from database.db_manager import get_db_manager
import uuid
db = get_db_manager()
# Tentar inserir um tribunal de teste
test_id = str(uuid.uuid4())
success = db.execute_query(
"""
INSERT INTO tribunais (id, nome, sigla, uf, tipo, created_at, updated_at)
VALUES (%s, %s, %s, %s, %s, NOW(), NOW())
""",
(test_id, "Tribunal Teste", "TEST", "DF", "Teste")
)
if success:
# Remover registro de teste
db.execute_query(
"DELETE FROM tribunais WHERE id = %s",
(test_id,)
)
return MessageResponse(
message="Test insert/delete successful",
details={"test_id": test_id}
)
else:
raise Exception("Insert failed")
except Exception as e:
logger.error(f"Test insert error: {e}")
raise HTTPException(status_code=500, detail=str(e))
|