49 lines
1.9 KiB
Python
49 lines
1.9 KiB
Python
from typing import Literal, Optional
|
|
|
|
from sqlalchemy import text
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
|
|
from database.firebird import Firebird
|
|
|
|
|
|
class BaseRepository:
|
|
def query(self, sql: str, params: Optional[dict] = None):
|
|
"""Executa uma consulta SQL e retorna o resultado como objeto ResultProxy"""
|
|
return self._execute(sql, params, fetch="result")
|
|
|
|
def fetch_all(self, sql: str, params: Optional[dict] = None):
|
|
"""Executa uma consulta SQL e retorna todos os registros com mapeamento de colunas"""
|
|
return self._execute(sql, params, fetch="all")
|
|
|
|
def fetch_one(self, sql: str, params: Optional[dict] = None):
|
|
"""Executa uma consulta SQL e retorna o primeiro registro com mapeamento de colunas"""
|
|
return self._execute(sql, params, fetch="one")
|
|
|
|
def run(self, sql: str, params: Optional[dict] = None):
|
|
"""Executa um SQL sem retorno de dados (ex: INSERT, UPDATE, DELETE)"""
|
|
return self._execute(sql, params, fetch="none")
|
|
|
|
def run_and_return(self, sql: str, params: Optional[dict] = None):
|
|
"""Executa SQL com RETURNING e retorna o primeiro registro como dict"""
|
|
return self._execute(sql, params, fetch="one")
|
|
|
|
def _execute(
|
|
self,
|
|
sql: str,
|
|
params: Optional[dict] = None,
|
|
fetch: Literal["all", "one", "result", "none"] = "result",
|
|
):
|
|
engine = Firebird.get_engine()
|
|
try:
|
|
with engine.begin() as conn:
|
|
result = conn.execute(text(sql), params or {})
|
|
if fetch == "all":
|
|
return result.mappings().all()
|
|
elif fetch == "one":
|
|
return result.mappings().first()
|
|
elif fetch == "none":
|
|
return None
|
|
return result # Result object
|
|
except SQLAlchemyError as e:
|
|
print(f"[ERRO SQL]: {e}")
|
|
raise
|