73 lines
2.5 KiB
Python
73 lines
2.5 KiB
Python
from typing import Optional
|
|
from sqlalchemy import create_engine, text
|
|
from sqlalchemy.engine import Engine
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
|
|
from actions.config.config import Config
|
|
|
|
class MySQL:
|
|
_engine: Optional[Engine] = None # Armazena a engine do SQLAlchemy para reutilização
|
|
|
|
@classmethod
|
|
def get_engine(cls) -> Engine:
|
|
"""
|
|
Retorna a engine do SQLAlchemy.
|
|
Cria a engine se ainda não existir e já testa a conexão.
|
|
"""
|
|
# Carrega configurações do JSON
|
|
database = Config.get('database/mysql.json')
|
|
|
|
if cls._engine is None:
|
|
# Monta DSN para o SQLAlchemy usando PyMySQL
|
|
dsn = (
|
|
f"mysql+pymysql://{database.user}:"
|
|
f"{database.password}@"
|
|
f"{database.host}:"
|
|
f"{database.port}/"
|
|
f"{database.name}"
|
|
)
|
|
|
|
# Cria a engine SQLAlchemy
|
|
cls._engine = create_engine(
|
|
dsn,
|
|
connect_args={"charset": database.charset},
|
|
pool_pre_ping=bool(database.pool.pre_ping), # Testa conexão antes de usar
|
|
pool_size=database.pool.size,
|
|
max_overflow=database.pool.max_overflow,
|
|
)
|
|
|
|
# Testa conexão imediatamente após criar a engine
|
|
try:
|
|
with cls._engine.connect() as conn:
|
|
conn.execute(text("SELECT 1")) # Consulta simples para verificar conexão
|
|
print("Conexão MySQL OK")
|
|
except SQLAlchemyError as e:
|
|
cls._engine = None # Se falhar, descarta a engine
|
|
print("Falha na conexão MySQL:", e)
|
|
raise # Repassa o erro para tratamento externo
|
|
|
|
return cls._engine
|
|
|
|
@classmethod
|
|
def dispose(cls):
|
|
"""
|
|
Fecha a engine e limpa o objeto para liberar recursos.
|
|
"""
|
|
if cls._engine:
|
|
cls._engine.dispose()
|
|
cls._engine = None
|
|
|
|
@classmethod
|
|
def test_connection(cls) -> bool:
|
|
"""
|
|
Testa a conexão com o banco de dados sem criar a engine permanente.
|
|
Retorna True se a conexão for bem-sucedida, False caso contrário.
|
|
"""
|
|
try:
|
|
engine = cls.get_engine()
|
|
with engine.connect() as conn:
|
|
conn.execute(text("SELECT 1")) # Consulta simples para testar conexão
|
|
return True
|
|
except SQLAlchemyError as e:
|
|
print("Erro ao testar conexão MySQL:", e)
|
|
return False
|