import os import base64 from dotenv import load_dotenv from sqlalchemy import Column, Integer, String from src.ConexionSql.Base_conexion import ConexionBase from src.base import Base from src.ApiKeys.openai_apikey import OpenAICredencial from src.Security.Encriptar import Encriptar_fernet from entrypoint import ENV_PATH from src.ArquitectureLayer.Mapper import Mapper_base from sqlalchemy import Column, String from src.ArquitectureLayer.Model import Model_base from src.ArquitectureLayer.Repo import Repo_base # ---------------------- # Cargar clave maestra # ---------------------- load_dotenv(ENV_PATH) pssword = os.getenv('MASTER_PASSWORD') if pssword is None: raise ValueError("MASTER_PASSWORD no está definida en el archivo .env") # ---------------------- # MODELO (SQLAlchemy) # ---------------------- class OpenAICredencialModel(Base, Model_base): __tablename__ = 'openai_credenciales' id = Column(String, primary_key=True) titulo = Column(String, nullable=False) api_key = Column(String, nullable=False) # Encriptada como base64 string organizacion = Column(String, nullable=True) # ---------------------- # MAPPER # ---------------------- class OpenAICredencialMapper(Mapper_base[OpenAICredencial, OpenAICredencialModel]): @staticmethod def to_model(obj: OpenAICredencial) -> OpenAICredencialModel: return OpenAICredencialModel( id=obj.id, titulo=obj.titulo, api_key=base64.b64encode( Encriptar_fernet.encriptar(obj.api_key, pssword) ).decode("utf-8"), organizacion=obj.organizacion ) @staticmethod def from_model(model: OpenAICredencialModel) -> OpenAICredencial: return OpenAICredencial( id=model.id, titulo=model.titulo, api_key=Encriptar_fernet.desencriptar( base64.b64decode(model.api_key), pssword ), organizacion=model.organizacion ) @staticmethod def to_dict(obj: OpenAICredencial) -> dict: return { "id": obj.id, "titulo": obj.titulo, "api_key": base64.b64encode( Encriptar_fernet.encriptar(obj.api_key, pssword) ).decode("utf-8"), "organizacion": obj.organizacion } @staticmethod def from_dict(data: dict) -> OpenAICredencial: return OpenAICredencial( id=data["id"], titulo=data["titulo"], api_key=Encriptar_fernet.desencriptar( base64.b64decode(data["api_key"]), pssword ), organizacion=data.get("organizacion") ) # ---------------------- # REPO # ---------------------- class OpenAICredencialRepo(Repo_base[OpenAICredencialModel, OpenAICredencial]): def __init__(self, conexion: ConexionBase): super().__init__( session=conexion.get_session(), modelo=OpenAICredencialModel, mapper=OpenAICredencialMapper ) def get_by_titulo(self, titulo: str) -> OpenAICredencial | None: model = ( self.session.query(self.Modelo) .filter_by(titulo=titulo, sys_deleted_at=None) .first() ) return self.Mapper.from_model(model) if model else None