114 lines
5.5 KiB
Python
114 lines
5.5 KiB
Python
# Importa tipos utilitários para anotações estáticas e sobrecarga de assinaturas.
|
|
from typing import Any, Mapping, List, Optional, Literal, Union, overload
|
|
|
|
# Função `text` para construir SQL parametrizado de forma segura.
|
|
from sqlalchemy import text
|
|
|
|
# Tipo de retorno de execução bruta de SQL (cursor/result set) do SQLAlchemy.
|
|
from sqlalchemy.engine import CursorResult
|
|
|
|
# Exceção base do SQLAlchemy para capturar erros de banco.
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
|
|
# Provedor do engine de conexão com o Firebird, centralizado no projeto.
|
|
from database.firebird import Firebird
|
|
|
|
|
|
# Define a classe base de repositórios, concentrando operações comuns de acesso a dados.
|
|
class BaseRepository:
|
|
# Sobrecarga 1: quando `fetch="all"`, o retorno é uma lista de mapeamentos (coluna->valor).
|
|
@overload
|
|
def _execute(
|
|
self, sql: str, params: Optional[dict[str, Any]], fetch: Literal["all"]
|
|
) -> List[Mapping[str, Any]]: ...
|
|
|
|
# Sobrecarga 2: quando `fetch="one"`, o retorno é um único mapeamento ou None.
|
|
@overload
|
|
def _execute(
|
|
self, sql: str, params: Optional[dict[str, Any]], fetch: Literal["one"]
|
|
) -> Optional[Mapping[str, Any]]: ...
|
|
|
|
# Sobrecarga 3: quando `fetch="none"`, não há retorno (operações DML sem leitura).
|
|
@overload
|
|
def _execute(
|
|
self, sql: str, params: Optional[dict[str, Any]], fetch: Literal["none"]
|
|
) -> None: ...
|
|
|
|
# Sobrecarga 4: quando `fetch="result"`, retorna o objeto `CursorResult` bruto do SQLAlchemy.
|
|
@overload
|
|
def _execute(
|
|
self, sql: str, params: Optional[dict[str, Any]], fetch: Literal["result"]
|
|
) -> CursorResult[Any]: ...
|
|
|
|
# Implementação concreta que atende às quatro sobrecargas por meio de um retorno em união.
|
|
def _execute(
|
|
self,
|
|
sql: str, # Comando SQL (SELECT/INSERT/UPDATE/DELETE) em texto.
|
|
params: Optional[dict[str, Any]] = None, # Parâmetros nomeados para o SQL.
|
|
fetch: Literal[
|
|
"all", "one", "none", "result"
|
|
] = "result", # Modo de leitura/retorno.
|
|
) -> Union[
|
|
List[Mapping[str, Any]], # Retorno quando `fetch="all"`.
|
|
Optional[Mapping[str, Any]], # Retorno quando `fetch="one"`.
|
|
None, # Retorno quando `fetch="none"`.
|
|
CursorResult[Any], # Retorno quando `fetch="result"`.
|
|
]:
|
|
"""Método interno que executa o SQL conforme o modo de fetch especificado."""
|
|
# Obtém o engine de conexão com o Firebird a partir do provider central.
|
|
engine = Firebird.get_engine()
|
|
try:
|
|
# Inicia um contexto transacional; `begin()` garante commit/rollback automáticos.
|
|
with engine.begin() as conn:
|
|
# Executa o SQL com parâmetros (usa dict vazio quando `params` é None).
|
|
result = conn.execute(text(sql), params or {})
|
|
# Quando for solicitado "all", converte o resultado em lista de mapeamentos (coluna->valor).
|
|
if fetch == "all":
|
|
# retorna Sequence[RowMapping], compatível com List[Mapping[str, Any]]
|
|
return list(result.mappings().all())
|
|
# Quando for solicitado "one", retorna apenas o primeiro registro (ou None).
|
|
elif fetch == "one":
|
|
return result.mappings().first()
|
|
# Quando for solicitado "none", não retorna dados (apenas executa o comando).
|
|
elif fetch == "none":
|
|
return None
|
|
# Caso padrão: retorna o objeto Result bruto para manipulações específicas.
|
|
return result
|
|
except SQLAlchemyError as e:
|
|
# Log simples para facilitar diagnóstico em ambiente de desenvolvimento/produção.
|
|
print(f"[ERRO SQL]: {e}")
|
|
# Propaga a exceção para camadas superiores tratarem (ex.: FastAPI Exception Handler).
|
|
raise
|
|
|
|
# Executa uma consulta e retorna o objeto `CursorResult` bruto (uso avançado ou stream).
|
|
def query(
|
|
self, sql: str, params: Optional[dict[str, Any]] = None
|
|
) -> CursorResult[Any]:
|
|
"""Executa uma consulta SQL e retorna o resultado como objeto ResultProxy."""
|
|
return self._execute(sql, params, fetch="result")
|
|
|
|
# Executa uma consulta e retorna todos os registros como lista de mapeamentos.
|
|
def fetch_all(
|
|
self, sql: str, params: Optional[dict[str, Any]] = None
|
|
) -> List[Mapping[str, Any]]:
|
|
"""Executa uma consulta SQL e retorna todos os registros com mapeamento de colunas."""
|
|
return self._execute(sql, params, fetch="all")
|
|
|
|
# Executa uma consulta e retorna apenas o primeiro registro ou None.
|
|
def fetch_one(
|
|
self, sql: str, params: Optional[dict[str, Any]] = None
|
|
) -> Optional[Mapping[str, Any]]:
|
|
"""Executa uma consulta SQL e retorna o primeiro registro com mapeamento de colunas."""
|
|
return self._execute(sql, params, fetch="one")
|
|
|
|
# Executa comandos DML (INSERT/UPDATE/DELETE) sem retorno de dados.
|
|
def run(self, sql: str, params: Optional[dict[str, Any]] = None) -> None:
|
|
"""Executa um SQL sem retorno de dados (ex: INSERT, UPDATE, DELETE)."""
|
|
self._execute(sql, params, fetch="none")
|
|
|
|
# Executa comandos com cláusula RETURNING e devolve o registro retornado (ou None).
|
|
def run_and_return(
|
|
self, sql: str, params: Optional[dict[str, Any]] = None
|
|
) -> Optional[Mapping[str, Any]]:
|
|
"""Executa SQL com RETURNING e retorna o primeiro registro como dict."""
|
|
return self._execute(sql, params, fetch="one")
|