feat: Implement text manager API and database connection

- Added `text_manager.py` to handle the creation of text libraries via FastAPI.
- Introduced database connection management in `conexion.py` using PostgreSQL credentials from environment variables.
- Created abstract base class `EmbedderABC` in `Base_Embedder.py` for embedding models.
- Developed `OpenAIEmbedder` class to generate embeddings using OpenAI's API.
- Implemented `OpenAIEmbedderModel` and repository pattern for managing OpenAI embedders in `Openai_embedder_mmr.py`.
- Established `Biblioteca` class for managing text libraries and their associated notes in `biblioteca.py`.
- Created SQLAlchemy models and mappers for `Biblioteca` and `Nota` in `biblioteca_mmr.py` and `notas_biblioteca_mmr.py`.
- Added functionality for dynamic table generation for notes associated with libraries.
- Included comprehensive methods for adding, retrieving, and managing notes and libraries in their respective repositories.
This commit is contained in:
2025-05-10 17:52:43 +02:00
parent c646bc1fef
commit c47b9474f4
27 changed files with 844 additions and 44 deletions
+96
View File
@@ -0,0 +1,96 @@
import os
import base64
from dotenv import load_dotenv
from sqlalchemy import Column, String, Integer
from src.ConexionSql.Base_conexion import ConexionBase
from src.base import Base
from src.Security.Encriptar import Encriptar_fernet
from src.Security.GenerarIDs import GeneradorIDUnico
from src.Llms.Embedders.Base_Embedder import EmbedderABC
from src.TextManager.biblioteca import Biblioteca # Suponiendo que defines la clase lógica Biblioteca aquí
# ----------------------
# 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 BibliotecaModel(Base):
__tablename__ = "bibliotecas"
id = Column(String, primary_key=True)
nombre = Column(String, nullable=False)
descripcion = Column(String, default="")
vector_dim = Column(Integer, nullable=False)
embedder_info = Column(String, nullable=True) # Se puede guardar nombre de clase o config encriptada
# ----------------------
# MAPPER
# ----------------------
class BibliotecaMapper:
@staticmethod
def to_dict(obj: Biblioteca) -> dict:
embedder_info = type(obj.embedder).__name__ if obj.embedder else None
return {
"id": obj.id,
"nombre": obj.nombre,
"descripcion": obj.descripcion,
"vector_dim": obj.vector_dim,
"embedder_info": embedder_info # sin codificar ni encriptar
}
@staticmethod
def from_dict(data: dict) -> Biblioteca:
return Biblioteca(
id=data["id"],
nombre=data["nombre"],
descripcion=data["descripcion"],
vector_dim=data["vector_dim"],
embedder=None # Mantienes la lógica actual de no restaurarlo automáticamente
)
@staticmethod
def from_model(model: BibliotecaModel) -> Biblioteca:
return Biblioteca(
id=model.id,
nombre=model.nombre,
descripcion=model.descripcion,
vector_dim=model.vector_dim,
embedder=None # Se puede cargar manualmente si es necesario
)
# ----------------------
# REPO
# ----------------------
class BibliotecaRepo:
def __init__(self, conexion: ConexionBase):
self.session = conexion.get_session()
def add(self, biblioteca: Biblioteca) -> str:
data = BibliotecaMapper.to_dict(biblioteca)
model = BibliotecaModel(**data)
self.session.add(model)
self.session.commit()
return model.id
def get_all(self) -> list[Biblioteca]:
models = self.session.query(BibliotecaModel).all()
return [BibliotecaMapper.from_model(m) for m in models]
def get_by_nombre(self, nombre: str) -> Biblioteca | None:
model = self.session.query(BibliotecaModel).filter_by(nombre=nombre).first()
return BibliotecaMapper.from_model(model) if model else None
def get_by_id(self, id_: str) -> Biblioteca | None:
model = self.session.get(BibliotecaModel, id_)
return BibliotecaMapper.from_model(model) if model else None