Uecko_ERP_FactuGES_sync/app/db/db_connection.py
2025-11-30 22:31:09 +01:00

70 lines
2.3 KiB
Python

from typing import Any, Mapping
import fdb
import mysql.connector
from app.config import logger
def get_factuges_connection(config) -> fdb.Connection:
"""
Crea y devuelve una conexión a FactuGES (Firebird).
"""
try:
conn = fdb.connect(
host=config['FACTUGES_HOST'],
port=int(config['FACTUGES_PORT']),
database=config['FACTUGES_DATABASE'],
user=config['FACTUGES_USER'],
password=config['FACTUGES_PASSWORD'],
charset='UTF8',
)
logger.info(
"Conexión a la base de datos FactuGES establecida: %s with database:%s - using user:%s",
config['FACTUGES_HOST'],
config['FACTUGES_DATABASE'],
config['FACTUGES_USER'],
)
return conn
except Exception as e:
logger.error("Error al conectar a la base de datos FactuGES.")
logger.error(
"(ERROR) Failed to establish connection to: %s with database:%s - using user:%s",
config['FACTUGES_HOST'],
config['FACTUGES_DATABASE'],
config['FACTUGES_USER'],
)
logger.error(str(e))
raise e
def get_mysql_connection(config: Mapping[str, Any]):
"""
Crea y devuelve una conexión a MySQL.
"""
try:
conn = mysql.connector.connect(
host=config["FWEB_MYSQL_HOST"],
port=config["FWEB_MYSQL_PORT"],
database=config["FWEB_MYSQL_DATABASE"],
user=config["FWEB_MYSQL_USER"],
password=config["FWEB_MYSQL_PASSWORD"],
)
logger.info(
"Conexión a la base de datos MySQL establecida a: %s with database:%s - using user:%s",
config['FWEB_MYSQL_HOST'],
config['FWEB_MYSQL_DATABASE'],
config['FWEB_MYSQL_USER'],
)
return conn
except Exception as e:
logger.error("Error al conectar a la base de datos MySQL.")
logger.error(
"(ERROR) Failed to establish connection to: %s with database:%s - using user:%s",
config['FWEB_MYSQL_HOST'],
config['FWEB_MYSQL_DATABASE'],
config['FWEB_MYSQL_USER'],
)
logger.error(str(e))
raise e