Indexing pipeline API

locallore.indexing.pipeline

update_index(settings, *, embedder=None)

Import changed sessions and embed any messages that are out of date.

Source code in src/locallore/indexing/pipeline.py

def update_index(
    settings: Settings,
    *,
    embedder: Embedder | None = None,
) -> tuple[ImportResult, int]:
    """Import changed sessions and embed any messages that are out of date."""
    with acquire_index_lock(settings.database_path):
        connection = connect(settings.database_path)
        try:
            migrate(connection)
            result = import_sessions(connection, settings.sessions_path)
            model_id = (
                embedder.model_id
                if embedder is not None
                else embedding_model_id(
                    settings.embedding_model,
                    settings.model_path,
                )
            )
            if not has_pending_messages(
                connection,
                model_id,
                settings.embedding_dimension,
            ):
                return result, 0
            if embedder is None:
                embedder = FastEmbedder(
                    settings.embedding_model,
                    settings.model_path,
                    settings.embedding_dimension,
                    model_id=model_id,
                )
            embedded = embed_pending_messages(
                connection,
                embedder,
                batch_size=settings.embedding_batch_size,
            )
            return result, embedded
        finally:
            connection.close()

locallore.indexing.locking

Coordinate indexing across processes.

logger = logging.getLogger(__name__) module-attribute

index_lock_path(database_path)

Source code in src/locallore/indexing/locking.py

def index_lock_path(database_path: Path) -> Path:
    return database_path.with_name(f"{database_path.name}.index.lock")

acquire_index_lock(database_path)

Allow only one process to update an index at a time.

Source code in src/locallore/indexing/locking.py

@contextmanager
def acquire_index_lock(database_path: Path) -> Iterator[None]:
    """Allow only one process to update an index at a time."""
    lock_path = index_lock_path(database_path)
    lock_path.parent.mkdir(parents=True, exist_ok=True)
    descriptor = os.open(lock_path, os.O_CREAT | os.O_RDWR, 0o600)
    with os.fdopen(descriptor, "rb+", closefd=True) as lock_file:
        try:
            fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
        except BlockingIOError:
            logger.info(
                "Another LocalLore indexing operation is running; "
                "waiting for it to complete"
            )
            fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX)
        try:
            yield
        finally:
            fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN)