misp-fixcor/db_setup.py

from sqlalchemy import create_engine, event, Column, Integer, String, DateTime, UniqueConstraint
from sqlalchemy.orm import sessionmaker, declarative_base

Base = declarative_base()

class ModificadosEv(Base):
    __tablename__ = "modificados_ev"
    id = Column(Integer, primary_key=True, autoincrement=True)
    evento_uuid = Column(String, nullable=False)
    publicado_fecha = Column(DateTime, nullable=False)
    attribute_count = Column(Integer, nullable=False)
    __table_args__ = (
        UniqueConstraint('evento_uuid', 'attribute_count', name='ev_attr'),
    )

def get_engine(db_path: str):
    # Uso de connect_args={"check_same_thread": False} para múltiples hilos
    return create_engine(
        f"sqlite:///{db_path}",
        echo=False,
        connect_args={"check_same_thread": False}
    )

def set_sqlite_pragma(dbapi_connection, connection_record):
    cursor = dbapi_connection.cursor()
    cursor.execute("PRAGMA journal_mode=WAL;")
    cursor.close()

def create_tables(engine):
    Base.metadata.create_all(bind=engine)

def get_session(engine):
    return sessionmaker(bind=engine, autoflush=False, autocommit=False)()