100 lines
4 KiB
Python
100 lines
4 KiB
Python
from typing import Optional, Literal
|
|
from sqlalchemy import text
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
from database.mysql import MySQL # Importa a classe MySQL que gerencia a engine
|
|
|
|
class BaseRepository:
|
|
"""
|
|
Classe base para todos os repositórios.
|
|
Contém métodos genéricos para executar SQL no MySQL usando SQLAlchemy.
|
|
"""
|
|
|
|
def query(self, sql: str, params: Optional[dict] = None):
|
|
"""
|
|
Executa uma query e retorna o ResultProxy bruto.
|
|
Útil para operações customizadas que precisam do objeto SQLAlchemy diretamente.
|
|
"""
|
|
return self._execute(sql, params, fetch="result")
|
|
|
|
def fetch_all(self, sql: str, params: Optional[dict] = None):
|
|
"""
|
|
Executa uma query SQL e retorna todos os registros como lista de dicionários.
|
|
Retorna lista vazia se não encontrar nada.
|
|
"""
|
|
return self._execute(sql, params, fetch="all")
|
|
|
|
def fetch_one(self, sql: str, params: Optional[dict] = None):
|
|
"""
|
|
Executa uma query SQL e retorna o primeiro registro como dicionário.
|
|
Retorna None se não encontrar nenhum registro.
|
|
"""
|
|
return self._execute(sql, params, fetch="one")
|
|
|
|
def run(self, sql: str, params: Optional[dict] = None):
|
|
"""
|
|
Executa um SQL sem retorno (ex: INSERT, UPDATE, DELETE).
|
|
Não retorna nenhum dado.
|
|
"""
|
|
return self._execute(sql, params, fetch="none")
|
|
|
|
def run_and_return(self, sql: str, params: Optional[dict] = None):
|
|
"""
|
|
Executa um INSERT e retorna o último ID gerado no MySQL.
|
|
Se for um SELECT, retorna o primeiro registro normalmente.
|
|
"""
|
|
engine = MySQL.get_engine() # Obtém a engine do MySQL
|
|
try:
|
|
with engine.begin() as conn: # Inicia uma transação automática
|
|
result = conn.execute(text(sql), params or {}) # Executa o SQL
|
|
|
|
# Se for INSERT, retorna o último ID inserido
|
|
if sql.strip().upper().startswith("INSERT"):
|
|
last_id = conn.execute(text("SELECT LAST_INSERT_ID() AS id")).mappings().first()
|
|
return last_id
|
|
|
|
# Se não for INSERT, retorna o primeiro registro
|
|
return result.mappings().first()
|
|
|
|
except SQLAlchemyError as e:
|
|
print(f"[ERRO SQL]: {e}") # Imprime o erro para debug
|
|
raise
|
|
|
|
def _execute(
|
|
self,
|
|
sql: str,
|
|
params: Optional[dict] = None,
|
|
fetch: Literal["all", "one", "result", "none"] = "result",
|
|
):
|
|
"""
|
|
Método interno que executa o SQL no MySQL.
|
|
Suporta diferentes tipos de retorno:
|
|
- all -> todos os registros como lista de dicionários
|
|
- one -> primeiro registro como dicionário
|
|
- result -> ResultProxy bruto
|
|
- none -> não retorna nada
|
|
"""
|
|
engine = MySQL.get_engine() # Pega a engine do MySQL
|
|
|
|
try:
|
|
with engine.connect() as conn: # Abre a conexão com o banco
|
|
result = conn.execute(text(sql), params or {}) # Executa o SQL com parâmetros
|
|
|
|
# Commit explícito para operações DML
|
|
# Se não for um SELECT, faça o commit para persistir a alteração.
|
|
sql_upper = sql.strip().upper()
|
|
if sql_upper.startswith(("INSERT", "UPDATE", "DELETE")):
|
|
conn.commit()
|
|
|
|
# Retorno baseado no tipo solicitado
|
|
if fetch == "all":
|
|
return result.mappings().all() # Todos os registros
|
|
elif fetch == "one":
|
|
return result.mappings().first() # Apenas o primeiro registro
|
|
elif fetch == "result":
|
|
return result # Retorno bruto do SQLAlchemy
|
|
elif fetch == "none":
|
|
return result.rowcount # Retorna o número de linhas afetadas
|
|
|
|
except SQLAlchemyError as e:
|
|
print(f"[ERRO SQL]: {e}") # Log de erro para debug
|
|
raise
|