Uecko_ERP_FactuGES_sync/app/db/db_connection.py

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