Source code for langchain.indexes.base
from __future__ import annotations
import uuid
from abc import ABC, abstractmethod
from typing import List, Optional, Sequence
NAMESPACE_UUID = uuid.UUID(int=1984)
[docs]class RecordManager(ABC):
    """An abstract base class representing the interface for a record manager."""
[docs]    def __init__(
        self,
        namespace: str,
    ) -> None:
        """Initialize the record manager.
        Args:
            namespace (str): The namespace for the record manager.
        """
        self.namespace = namespace
[docs]    @abstractmethod
    def create_schema(self) -> None:
        """Create the database schema for the record manager."""
[docs]    @abstractmethod
    def get_time(self) -> float:
        """Get the current server time as a high resolution timestamp!
        It's important to get this from the server to ensure a monotonic clock,
        otherwise there may be data loss when cleaning up old documents!
        Returns:
            The current server time as a float timestamp.
        """
[docs]    @abstractmethod
    def update(
        self,
        keys: Sequence[str],
        *,
        group_ids: Optional[Sequence[Optional[str]]] = None,
        time_at_least: Optional[float] = None,
    ) -> None:
        """Upsert records into the database.
        Args:
            keys: A list of record keys to upsert.
            group_ids: A list of group IDs corresponding to the keys.
            time_at_least: if provided, updates should only happen if the
              updated_at field is at least this time.
        Raises:
            ValueError: If the length of keys doesn't match the length of group_ids.
        """
[docs]    @abstractmethod