Source code for langchain_core.indexing.base

from __future__ import annotations

from abc import ABC, abstractmethod
from typing import List, Optional, Sequence


class RecordManager(ABC):
    """抽象基类，表示记录管理器的接口。"""

    def __init__(
        self,
        namespace: str,
    ) -> None:
        """初始化记录管理器。

参数：
    namespace (str)：记录管理器的命名空间。
"""
        self.namespace = namespace

    @abstractmethod
    def create_schema(self) -> None:
        """为记录管理器创建数据库模式。"""

    @abstractmethod
    async def acreate_schema(self) -> None:
        """异步创建记录管理器的数据库模式。"""

    @abstractmethod
    def get_time(self) -> float:
        """获取当前服务器时间作为高分辨率时间戳！

从服务器获取这个时间是很重要的，以确保一个单调时钟，否则在清理旧文档时可能会丢失数据！

返回：
    当前服务器时间作为浮点时间戳。
"""

    @abstractmethod
    async def aget_time(self) -> float:
        """异步获取当前服务器时间作为高分辨率时间戳。
重要的是从服务器获取此时间以确保单调时钟，否则在清理旧文档时可能会丢失数据！
返回：
当前服务器时间作为浮点时间戳。
"""

    @abstractmethod
    def update(
        self,
        keys: Sequence[str],
        *,
        group_ids: Optional[Sequence[Optional[str]]] = None,
        time_at_least: Optional[float] = None,
    ) -> None:
        """将记录插入数据库。

参数：
    keys：需要插入的记录键的列表。
    group_ids：与键对应的组ID列表。
    time_at_least：可选的时间戳。实现可以使用此参数
        可选地验证时间戳至少是存储数据的系统中的时间。

        例如，用于验证Postgres数据库中的时间是否等于或大于给定时间戳，如果不是，则
        抛出错误。

        这旨在帮助防止时间漂移问题，因为
        时间可能不是单调递增的！

引发：
    ValueError：如果键的长度与group_ids的长度不匹配。
"""

    @abstractmethod
    async def aupdate(
        self,
        keys: Sequence[str],
        *,
        group_ids: Optional[Sequence[Optional[str]]] = None,
        time_at_least: Optional[float] = None,
    ) -> None:
        """将记录异步更新到数据库中。

参数：
    keys：需要更新的记录键列表。
    group_ids：与键对应的组ID列表。
    time_at_least：可选的时间戳。实现可以使用此参数
        可选地验证时间戳至少等于数据存储系统中的时间。

        例如，用于验证postgres数据库中的时间
        是否等于或大于给定的时间戳，如果不是，则
        抛出错误。

        这旨在帮助防止时间漂移问题，因为
        时间可能不是单调递增的！

抛出：
    ValueError：如果键的长度与group_ids的长度不匹配。
"""

    @abstractmethod
    def exists(self, keys: Sequence[str]) -> List[bool]:
        """检查提供的键是否存在于数据库中。

参数：
    keys：要检查的键列表。

返回：
    一个布尔值列表，指示每个键的存在情况。
"""

    @abstractmethod
    async def aexists(self, keys: Sequence[str]) -> List[bool]:
        """异步检查提供的键是否存在于数据库中。

参数：
    keys：要检查的键列表。

返回：
    一个布尔值列表，指示每个键的存在情况。
"""

    @abstractmethod
    def list_keys(
        self,
        *,
        before: Optional[float] = None,
        after: Optional[float] = None,
        group_ids: Optional[Sequence[str]] = None,
        limit: Optional[int] = None,
    ) -> List[str]:
        """根据提供的过滤条件列出数据库中的记录。

参数：
    before：过滤条件，列出在此时间之前更新的记录。
    after：过滤条件，列出在此时间之后更新的记录。
    group_ids：过滤条件，列出具有特定组ID的记录。
    limit：可选参数，限制要返回的记录数。

返回：
    匹配记录的键列表。
"""

    @abstractmethod
    async def alist_keys(
        self,
        *,
        before: Optional[float] = None,
        after: Optional[float] = None,
        group_ids: Optional[Sequence[str]] = None,
        limit: Optional[int] = None,
    ) -> List[str]:
        """根据提供的过滤条件异步列出数据库中的记录。

参数：
    before: 过滤条件，列出在此时间之前更新的记录。
    after: 过滤条件，列出在此时间之后更新的记录。
    group_ids: 过滤条件，列出具有特定组ID的记录。
    limit: 可选参数，限制要返回的记录数量。

返回：
    匹配记录的键列表。
"""

    @abstractmethod
    def delete_keys(self, keys: Sequence[str]) -> None:
        """从数据库中删除指定的记录。

参数：
    keys：要删除的键的列表。
"""

    @abstractmethod
    async def adelete_keys(self, keys: Sequence[str]) -> None:
        """从数据库中异步删除指定记录。

参数：
    keys：要删除的键列表。
"""