Refactor inicialización modelo

app/apis/FirebaseAuth.py

@@ -1,12 +1,11 @@
 import firebase_admin.auth
 from firebase_admin.auth import *
-from fastapi import Request, Header
+from fastapi import Request
 from fastapi.responses import JSONResponse
 from utils.Validadores import validar_txt_token
 from utils.Fechas import convertir_datetime_str
 from utils.Diccionario import ver_si_existe_clave
 from apis.Firestore import obtener_roles_usuarios, obtener_rol_usuario
-from constants import TEXTOS
 import asyncio
 
 
@@ -34,7 +33,7 @@ def validar_token(
 
 
 async def verificar_token(
-    peticion: Request, firebase_app, call_next, authorization: str, idioma="es"
+    peticion: Request, firebase_app, call_next, authorization: str, textos: dict[str, str], idioma: str ="es"
 ) -> JSONResponse:
     """
     Verifica el token de Firebase en la solicitud.
@@ -58,31 +57,32 @@ async def verificar_token(
                 return await call_next(peticion)
             case 0:
                 return JSONResponse(
-                    {"error": TEXTOS[idioma]["errTokenInvalido"]},
+                    {"error": textos[idioma]["errTokenInvalido"]},
                     status_code=403,
                     media_type="application/json",
                 )
             case _:
                 return JSONResponse(
-                    {"error": TEXTOS[idioma]["errValidarToken"]},
+                    {"error": textos[idioma]["errValidarToken"]},
                     status_code=400,
                     media_type="application/json",
                 )
     except Exception as e:
         return JSONResponse(
-            {"error": f"{TEXTOS[idioma]['errTry']} {e}"},
+            {"error": f"{textos[idioma]['errTry']} {e}"},
             status_code=500,
             media_type="application/json",
         )
 
 
-def ver_datos_token(token: str, firebase_app, idioma: str) -> tuple[int, dict]:
+def ver_datos_token(token: str, firebase_app, idioma: str, textos: dict[str, str]) -> tuple[int, dict]:
     """
     Obtiene los datos del token de Firebase.
     Args:
         token (str): El token de autorización de la solicitud.
         firebase_app: La instancia de la aplicación Firebase.
         idioma (str): El idioma para los mensajes de error.
+        textos (dict[str, str]): El diccionario de textos para los mensajes de error.
     Returns:
         tuple: (True, datos) si el token es válido, (False, error) si hay un error.
     """
@@ -91,28 +91,29 @@ def ver_datos_token(token: str, firebase_app, idioma: str) -> tuple[int, dict]:
         reg_validacion = validar_txt_token(token)
 
         if not reg_validacion:
-            return (0, {"error": f"{TEXTOS[idioma]['errTokenInvalido']}"})
+            return (0, {"error": f"{textos[idioma]['errTokenInvalido']}"})
 
         res_validacion = validar_token(token, firebase_app, True)
 
         match res_validacion[0]:
             case 1:
                 return res_validacion
             case 0:
-                return (0, {"error": f"{TEXTOS[idioma]['errTokenInvalido']}"})
+                return (0, {"error": f"{textos[idioma]['errTokenInvalido']}"})
             case _:
-                return (-1, {"error": f"{TEXTOS[idioma]['errValidarToken']}"})
+                return (-1, {"error": f"{textos[idioma]['errValidarToken']}"})
 
     except Exception as e:
-        return (-1, {"error": f"{TEXTOS[idioma]['errProcesarToken']}: {e}."})
+        return (-1, {"error": f"{textos[idioma]['errProcesarToken']}: {e}."})
 
 
-async def ver_datos_usuarios(firebase_app, idioma: str) -> JSONResponse:
+async def ver_datos_usuarios(firebase_app, idioma: str, textos: dict[str, str]) -> JSONResponse:
     """
     Obtiene los datos de los usuarios registrados en Firebase.
     Args:
         firebase_app: La instancia de la aplicación Firebase.
         idioma (str): El idioma para los mensajes de error.
+        textos (dict[str, str]): El diccionario de textos para los mensajes de error.
     Returns:
         JSONResponse: Los datos de los usuarios, o un error si ocurre un problema.
     """
@@ -155,19 +156,20 @@ async def ver_datos_usuarios(firebase_app, idioma: str) -> JSONResponse:
         )
     except Exception as e:
         return JSONResponse(
-            {"error": f"{TEXTOS[idioma]['errObtenerDatosUsuarios']}: {e}"},
+            {"error": f"{textos[idioma]['errObtenerDatosUsuarios']}: {e}"},
             status_code=400,
             media_type="application/json",
         )
 
 
-async def ver_datos_usuario(firebase_app, uid: str, idioma: str) -> JSONResponse:
+async def ver_datos_usuario(firebase_app, uid: str, idioma: str, textos: dict[str, str]) -> JSONResponse:
     """
     Obtiene los datos de un usuario específico usando el UID.
     Args:
         firebase_app: La instancia de la aplicación Firebase.
         uid (str): El UID del usuario a buscar.
         idioma (str): El idioma para los mensajes de error.
+        textos (dict[str, str]): El diccionario de textos para los mensajes de error.
     Returns:
         JSONResponse: Los datos del usuario, o un error si ocurre un problema.
     """
@@ -177,7 +179,7 @@ async def ver_datos_usuario(firebase_app, uid: str, idioma: str) -> JSONResponse
         ROL = await roles_task
 
         if ROL == -1:
-            raise UserNotFoundError(f"{TEXTOS[idioma]['errUsuarioNoEncontrado']}")
+            raise UserNotFoundError(f"{textos[idioma]['errUsuarioNoEncontrado']}")
 
         RES = {
             "correo": usuario.email,
@@ -200,13 +202,13 @@ async def ver_datos_usuario(firebase_app, uid: str, idioma: str) -> JSONResponse
         )
     except UserNotFoundError:
         return JSONResponse(
-            {"error": f"{TEXTOS[idioma]['errUsuarioNoEncontrado']}"},
+            {"error": f"{textos[idioma]['errUsuarioNoEncontrado']}"},
             status_code=404,
             media_type="application/json",
         )
     except Exception as e:
         return JSONResponse(
-            {"error": f"{TEXTOS[idioma]['errObtenerDatosUsuarios']}: {e}"},
+            {"error": f"{textos[idioma]['errObtenerDatosUsuarios']}: {e}"},
             status_code=400,
             media_type="application/json",
         )
@@ -230,7 +232,7 @@ def ver_usuario_firebase(firebase_app, uid: str) -> tuple[int, UserRecord | None
 
 
 def actualizar_estado_usuario(
-    firebase_app, uid: str, estado: bool, idioma: str
+    firebase_app, uid: str, estado: bool, idioma: str, textos: dict[str, str]
 ) -> JSONResponse:
     """
     Actualiza el estado (activado/desactivado) de un usuario específico.
@@ -239,26 +241,27 @@ def actualizar_estado_usuario(
         uid (str): El UID del usuario a actualizar.
         estado (bool): El nuevo estado del usuario (True para desactivado, False para activado).
         idioma (str): El idioma para los mensajes de error.
+        textos (dict[str, str]): El diccionario de textos para los mensajes de error.
     Returns:
         JSONResponse | Response: Mensaje de éxito o error.
     """
     try:
         firebase_admin.auth.update_user(uid=uid, disabled=estado, app=firebase_app)
 
         return JSONResponse(
-            {"mensaje": f"{TEXTOS[idioma]['msgUsuarioActualizado']}"},
+            {"mensaje": f"{textos[idioma]['msgUsuarioActualizado']}"},
             status_code=200,
             media_type="application/json",
         )
     except ValueError:
         return JSONResponse(
-            {"error": f"{TEXTOS[idioma]['errEstadoInvalido']}"},
+            {"error": f"{textos[idioma]['errEstadoInvalido']}"},
             status_code=401,
             media_type="application/json",
         )
     except Exception as e:
         return JSONResponse(
-            {"error": f"{TEXTOS[idioma]['errTry']} {str(e)}"},
+            {"error": f"{textos[idioma]['errTry']} {str(e)}"},
             status_code=500,
             media_type="application/json",
         )

app/apis/Recaptcha.py

@@ -1,35 +1,36 @@
 from constants import RECAPTCHA_SECRET, RECAPTCHA_API_URL
 from requests import post
-from constants import TEXTOS
 
 
-def manejador_errores(error: str, idioma: str) -> str:
+def manejador_errores(error: str, idioma: str, textos: dict[str, str]) -> str:
     """
     Maneja los errores devueltos por la API de reCAPTCHA.
 
     Args:
         error (str): El mensaje de error devuelto por la API de reCAPTCHA.
         idioma (str): El idioma para los mensajes de error.
+        textos (dict[str, str]): El diccionario de textos para los mensajes de error.
 
     Returns:
         str: Un mensaje de error amigable para el usuario.
     """
     match error:
         case "invalid-input-response":
-            return f"{TEXTOS[idioma]['errCaptchaTokenErroneo']}"
+            return f"{textos[idioma]['errCaptchaTokenErroneo']}"
         case "timeout-or-duplicate":
-            return f"{TEXTOS[idioma]['errCaptchaTokenInvalido']}"
+            return f"{textos[idioma]['errCaptchaTokenInvalido']}"
         case _:
             return error
 
 
-def verificar_peticion_recaptcha(token: str, idioma: str) -> dict:
+def verificar_peticion_recaptcha(token: str, idioma: str, textos: dict[str, str]) -> dict:
     """
     Envía una petición al API de ReCAPTCHA para verificar que el token es válido.
 
     Args:
         token (str): El token de ReCAPTCHA a verificar.
         idioma (str): El idioma para los mensajes de error.
+        textos (dict[str, str]): El diccionario de textos para los mensajes de error.
 
     Returns:
         dict: La respuesta del API de ReCAPTCHA.
@@ -44,6 +45,6 @@ def verificar_peticion_recaptcha(token: str, idioma: str) -> dict:
 
     for key, value in res.items():
         if key == "error-codes":
-            res[key] = [manejador_errores(i, idioma) for i in value]
+            res[key] = [manejador_errores(i, idioma, textos) for i in value]
 
     return res

app/constants.py

@@ -1,30 +1,10 @@
 from os import getenv
-from dill import load as dload
-from json import load as jload
-from pathlib import Path
 from utils.Dominios import obtener_lista_dominios
 
-with open(f"{Path(__file__).resolve().parent}/bin/explicador.pkl", "rb") as archivo:
-    EXPLAINER = dload(archivo)
-
-with open(f"{Path(__file__).resolve().parent}/bin/textos.json") as archivo:
-    TEXTOS = jload(archivo)
-
 ROL_ADMIN = 1001
 CORS_ORIGINS = obtener_lista_dominios(getenv("CORS_ORIGINS", "http://localhost:5173,"))
 ORIGENES_AUTORIZADOS = obtener_lista_dominios(getenv("ORIGENES_AUTORIZADOS", "http://localhost:5173,"))
 ACTIVAR_DOCS = getenv("ACTIVAR_DOCS", "false").lower() == "true"
 ALLOWED_HOSTS = obtener_lista_dominios(getenv("ALLOWED_HOSTS", "localhost,"))
 RECAPTCHA_SECRET = getenv("CAPTCHA_SECRET")
-RECAPTCHA_API_URL = getenv("API_RECAPTCHA_URL", "https://www.google.com/recaptcha/api/siteverify")
-CREDS_FIREBASE_CLIENTE = {
-    "apiKey": getenv("CLIENTE_FIREBASE_API_KEY"),
-    "authDomain": getenv("CLIENTE_FIREBASE_AUTH_DOMAIN"),
-    "projectId": getenv("CLIENTE_FIREBASE_PROJECT_ID"),
-    "storageBucket": getenv("CLIENTE_FIREBASE_STORAGE_BUCKET"),
-    "messagingSenderId": getenv("CLIENTE_FIREBASE_MESSAGING_SENDER_ID"),
-    "appId": getenv("CLIENTE_FIREBASE_APP_ID"),
-    "measurementId": getenv("CLIENTE_FIREBASE_MEASUREMENT_ID"),
-    "driveScopes": obtener_lista_dominios(getenv("CLIENTE_DRIVE_SCOPES","")),
-    "reCAPTCHA": getenv("CLIENTE_CAPTCHA"),
-}
+RECAPTCHA_API_URL = getenv("API_RECAPTCHA_URL", "https://www.google.com/recaptcha/api/siteverify")

app/dependencies/usuarios_dependencies.py

@@ -1,29 +1,30 @@
-from firebase_admin_config import firebase_app
 from apis.FirebaseAuth import ver_datos_token
 from apis.Firestore import verificar_rol_usuario
-from fastapi import Header
+from fastapi import Header, Request
 from fastapi.responses import JSONResponse
-#from utils.Diccionario import ver_si_existe_clave
-from constants import TEXTOS
 
 
 async def verificar_usuario_administrador(
+    peticion: Request,
     language: str | None = Header(default="es"),
     authorization: str | None = Header(default=""),
 ) -> tuple[bool, JSONResponse | None]:
     """
     Verifica si el usuario está autenticado y es administrador antes de permitir el acceso a las rutas protegidas.
 
     Args:
+        peticion (Request): La solicitud HTTP entrante.
         language (str | None): El idioma de la solicitud HTTP.
         authorization (str | None): El token de autorización de Firebase.
     Returns:
         tuple: Un tuple que contiene un booleano indicando si el usuario está autenticado y es administrador, y
         en caso de error, un JSONResponse con el mensaje de error y el código de estado correspondiente.
     """
+    firebase_app = peticion.state.firebase_app
+    TEXTOS = peticion.state.textos
     idioma = language if language in ("es", "en") else "es"
     try:
-        RES, DATOS = ver_datos_token(authorization, firebase_app, idioma)
+        RES, DATOS = ver_datos_token(authorization, firebase_app, idioma, TEXTOS)
 
         if RES in (-1, 0):
             return False, JSONResponse(

app/firebase_admin_config.py

@@ -1,16 +1,21 @@
-import firebase_admin
+from firebase_admin import initialize_app, App
+from firebase_admin.credentials import Certificate
 from pathlib import Path
 from os.path import join, exists
-from dotenv import load_dotenv
 from os import getenv
 
-load_dotenv()
+def inicializar_firebase() -> App:
+    """
+    Inicializa la aplicación de Firebase utilizando las credenciales proporcionadas en un archivo JSON.
 
-ALT_PATH = Path(getenv("FIREBASE_ADMIN_CREDS_PATH", ""))
+    Returns:
+        firebase_admin.App: Una instancia de la aplicación de Firebase inicializada.
+    """
+    ALT_PATH = Path(getenv("FIREBASE_ADMIN_CREDS_PATH", ""))
 
-BASE_DIR = Path(__file__).resolve().parent.parent
-PATH = join(BASE_DIR, "firebase_token.json") if (ALT_PATH != "" and (not exists(ALT_PATH))) else ALT_PATH
+    BASE_DIR = Path(__file__).resolve().parent.parent
+    PATH = join(BASE_DIR, "firebase_token.json") if (ALT_PATH != "" and (not exists(ALT_PATH))) else ALT_PATH
 
-# Inicialización del servicio de Firebase
-cred = firebase_admin.credentials.Certificate(PATH)
-firebase_app = firebase_admin.initialize_app(cred)
+    # Inicialización del servicio de Firebase
+    cred = Certificate(PATH)
+    return initialize_app(cred)