import os from dotenv import load_dotenv from sqlalchemy import Column, Integer, String from sqlalchemy.orm import relationship from sqlalchemy import DateTime, Text, func import base64 from domains.ArquitectureLayer.Mapper import Mapper_base from domains.ArquitectureLayer.Model import Model_base from domains.ArquitectureLayer.Repo import Repo_base from domains.ConexionSql.Base_conexion import ConexionBase from domains.base import Base from domains.Credenciales.postgres_credencial import PostgresCredencial from domains.Security.Encriptar import Encriptar_fernet # ---------------------- # Cargar clave maestra # ---------------------- from entrypoint import ENV_PATH 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 PostgresCredencialModel(Base, Model_base): __tablename__ = 'postgres_credenciales' id = Column(String, primary_key=True) titulo = Column(String, nullable=False) host = Column(String, nullable=False) port = Column(Integer, nullable=False) dbname = Column(String, nullable=False) user = Column(String, nullable=False) password = Column(String, nullable=False) # Encriptada como base64 string # ---------------------- # MAPPER # ---------------------- class PostgresCredencialMapper(Mapper_base[PostgresCredencial, PostgresCredencialModel]): @staticmethod def to_model(obj: PostgresCredencial) -> PostgresCredencialModel: return PostgresCredencialModel( id=obj.id, titulo=obj.titulo, host=obj.host, port=obj.port, dbname=obj.dbname, user=obj.user, password=base64.b64encode( Encriptar_fernet.encriptar(obj.password, pssword) ).decode('utf-8') ) @staticmethod def from_model(model: PostgresCredencialModel) -> PostgresCredencial: return PostgresCredencial( id=model.id, titulo=model.titulo, host=model.host, port=model.port, dbname=model.dbname, user=model.user, password=Encriptar_fernet.desencriptar( base64.b64decode(model.password), pssword ) ) @staticmethod def to_dict(obj: PostgresCredencial) -> dict: return { "id": obj.id, "titulo": obj.titulo, "host": obj.host, "port": obj.port, "dbname": obj.dbname, "user": obj.user, "password": base64.b64encode( Encriptar_fernet.encriptar(obj.password, pssword) ).decode('utf-8') } @staticmethod def from_dict(data: dict) -> PostgresCredencial: return PostgresCredencial( id=data["id"], titulo=data["titulo"], host=data["host"], port=data["port"], dbname=data["dbname"], user=data["user"], password=Encriptar_fernet.desencriptar( base64.b64decode(data["password"]), pssword ) ) # ---------------------- # REPO # ---------------------- class PostgresCredencialRepo(Repo_base[PostgresCredencialModel, PostgresCredencial]): def __init__(self, conexion: ConexionBase): super().__init__( session=conexion.get_session(), modelo=PostgresCredencialModel, mapper=PostgresCredencialMapper ) def get_by_titulo(self, titulo: str) -> PostgresCredencial | 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