Samuel_4.0 / database.py
Lukeetah's picture
Upload 13 files
5b0bb4b verified
# database.py
# Define los modelos de la base de datos (SQLAlchemy ORM) y la lógica de sesión.
import os
from sqlalchemy import create_engine, Column, Integer, String, DateTime, Boolean, ForeignKey
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime
# --- Configuración de la Base de Datos ---
# En un entorno de solo lectura, DEBEMOS escribir en el único directorio escribible: /tmp
DATA_DIR = "/tmp/data" # RUTA ABSOLUTA a un directorio escribible.
if not os.path.exists(DATA_DIR):
# Esto ahora intentará crear el directorio en /tmp/data, que sí tiene permisos.
os.makedirs(DATA_DIR)
DATABASE_URL = f"sqlite:///{DATA_DIR}/samuel.db"
# El engine es el punto de entrada a la base de datos.
# connect_args es necesario para SQLite para permitir su uso en múltiples hilos (como en FastAPI).
engine = create_engine(
DATABASE_URL, connect_args={"check_same_thread": False}
)
# SessionLocal es una fábrica de sesiones de base de datos.
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# Base es una clase base para nuestros modelos ORM.
Base = declarative_base()
# --- Modelos de Datos ORM ---
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
email = Column(String, unique=True, index=True, nullable=False)
hashed_password = Column(String, nullable=False)
has_completed_initiation = Column(Boolean, default=False)
memories = relationship("Memory", back_populates="owner")
class Memory(Base):
__tablename__ = "memories"
id = Column(Integer, primary_key=True, index=True)
content = Column(String, nullable=False)
timestamp = Column(DateTime, default=datetime.utcnow)
user_id = Column(Integer, ForeignKey("users.id"))
owner = relationship("User", back_populates="memories")
# --- Funciones de Ayuda ---
def create_db_and_tables():
"""Crea todas las tablas en la base de datos si no existen."""
Base.metadata.create_all(bind=engine)
def get_db():
"""Generador de dependencia de FastAPI para obtener una sesión de DB."""
db = SessionLocal()
try:
yield db
finally:
db.close()