Lizandro GuarnizoandClaude Sonnet 4.6
0a234f056c
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
116 lines
3.9 KiB
Python
116 lines
3.9 KiB
Python
import re
|
|
import fdb
|
|
import datetime
|
|
import decimal
|
|
from typing import Optional
|
|
|
|
|
|
def _safe(v):
|
|
if v is None:
|
|
return None
|
|
if isinstance(v, datetime.datetime):
|
|
return v.strftime("%Y-%m-%dT%H:%M:%S")
|
|
if isinstance(v, datetime.date):
|
|
return v.isoformat()
|
|
if isinstance(v, datetime.time):
|
|
return v.strftime("%H:%M:%S")
|
|
if isinstance(v, decimal.Decimal):
|
|
return float(v)
|
|
if isinstance(v, (bytes, bytearray, memoryview)):
|
|
return f"<BLOB {len(v)} bytes>"
|
|
if isinstance(v, (int, float, bool)):
|
|
return v
|
|
# strings: strip any spaces around colons (Firebird sometimes returns "12: 47: 37")
|
|
s = str(v)
|
|
return re.sub(r'\s*:\s*', ':', s) if ':' in s else s
|
|
|
|
|
|
class FirebirdService:
|
|
def __init__(self):
|
|
self.conn = None
|
|
|
|
def connect(self, host: str, port: int, database: str, user: str, password: str):
|
|
import platform, os
|
|
try:
|
|
if platform.system() == "Darwin":
|
|
mac_lib = "/Library/Frameworks/Firebird.framework/Versions/A/Resources/lib/libfbclient.dylib"
|
|
if os.path.exists(mac_lib):
|
|
os.environ.setdefault("DYLD_LIBRARY_PATH",
|
|
"/Library/Frameworks/Firebird.framework/Versions/A/Resources/lib")
|
|
try:
|
|
fdb.load_api(mac_lib)
|
|
except Exception:
|
|
pass
|
|
self.conn = fdb.connect(
|
|
host=host,
|
|
port=port,
|
|
database=database,
|
|
user=user,
|
|
password=password,
|
|
charset="UTF8",
|
|
)
|
|
return True, "Conexión exitosa"
|
|
except Exception as e:
|
|
return False, str(e)
|
|
|
|
def disconnect(self):
|
|
if self.conn:
|
|
self.conn.close()
|
|
self.conn = None
|
|
|
|
def is_connected(self) -> bool:
|
|
return self.conn is not None
|
|
|
|
def test_connection(self) -> tuple:
|
|
if not self.conn:
|
|
return False, "No hay conexión activa"
|
|
try:
|
|
cur = self.conn.cursor()
|
|
cur.execute("SELECT 1 FROM RDB$DATABASE")
|
|
cur.fetchone()
|
|
return True, "Conexión OK"
|
|
except Exception as e:
|
|
return False, str(e)
|
|
|
|
def execute_query(self, query: str, params: Optional[dict] = None) -> tuple:
|
|
if not self.conn:
|
|
return False, "No hay conexión activa", []
|
|
try:
|
|
cur = self.conn.cursor()
|
|
if params:
|
|
if isinstance(params, (list, tuple)):
|
|
# Parámetros posicionales directos (?)
|
|
cur.execute(query, params)
|
|
else:
|
|
# Parámetros nombrados (:name) → convertir a positional
|
|
param_names = re.findall(r':([a-zA-Z_][a-zA-Z0-9_]*)', query)
|
|
positional_sql = re.sub(r':[a-zA-Z_][a-zA-Z0-9_]*', '?', query)
|
|
positional_vals = [params[n] for n in param_names]
|
|
cur.execute(positional_sql, positional_vals)
|
|
else:
|
|
cur.execute(query)
|
|
columns = [desc[0] for desc in cur.description] if cur.description else []
|
|
rows = cur.fetchall()
|
|
return True, "", [{c: _safe(v) for c, v in zip(columns, row)} for row in rows]
|
|
except Exception as e:
|
|
return False, str(e), []
|
|
|
|
|
|
def _port(val, default=3050) -> int:
|
|
try:
|
|
return int(val)
|
|
except (TypeError, ValueError):
|
|
return default
|
|
|
|
|
|
def get_firebird_from_config(configs: dict) -> tuple:
|
|
fb = FirebirdService()
|
|
ok, msg = fb.connect(
|
|
configs.get("firebird_host", "localhost"),
|
|
_port(configs.get("firebird_port"), 3050),
|
|
configs.get("firebird_database", ""),
|
|
configs.get("firebird_user", "SYSDBA"),
|
|
configs.get("firebird_password", "masterkey"),
|
|
)
|
|
return fb, ok, msg
|