Source code for langchain.storage.file_system

import os
import re
import time
from pathlib import Path
from typing import Iterator, List, Optional, Sequence, Tuple, Union

from langchain_core.stores import ByteStore

from langchain.storage.exceptions import InvalidKeyException


class LocalFileStore(ByteStore):
    """基于本地文件系统的BaseStore接口。

    示例:
        创建一个LocalFileStore实例并对其执行操作:

        .. code-block:: python

            from langchain.storage import LocalFileStore

            # 使用根路径实例化LocalFileStore
            file_store = LocalFileStore("/path/to/root")

            # 为键设置值
            file_store.mset([("key1", b"value1"), ("key2", b"value2")])

            # 获取键的值
            values = file_store.mget(["key1", "key2"])  # 返回[b"value1", b"value2"]

            # 删除键
            file_store.mdelete(["key1"])

            # 遍历键
            for key in file_store.yield_keys():
                print(key)  # noqa: T201"""

    def __init__(
        self,
        root_path: Union[str, Path],
        *,
        chmod_file: Optional[int] = None,
        chmod_dir: Optional[int] = None,
        update_atime: bool = False,
    ) -> None:
        """为本地文件系统实现BaseStore接口。

参数:
    root_path (Union[str, Path]): 文件存储的根路径。所有键都被解释为相对于此根路径的路径。
    chmod_file: (可选，默认为`None`) 如果指定，设置新创建文件的权限，必要时覆盖当前的`umask`。
    chmod_dir: (可选，默认为`None`) 如果指定，设置新创建目录的权限，必要时覆盖当前的`umask`。
    update_atime: (可选，默认为`False`) 如果为`True`，在读取文件时更新文件系统的访问时间（但不更新修改时间）。
        这允许在禁用访问时间更新的文件系统上实现MRU/LRU缓存策略。
"""
        self.root_path = Path(root_path).absolute()
        self.chmod_file = chmod_file
        self.chmod_dir = chmod_dir
        self.update_atime = update_atime

    def _get_full_path(self, key: str) -> Path:
        """获取相对于根路径的给定键的完整路径。

参数：
    key（str）：相对于根路径的键。

返回：
    Path：给定键的完整路径。
"""
        if not re.match(r"^[a-zA-Z0-9_.\-/]+$", key):
            raise InvalidKeyException(f"Invalid characters in key: {key}")
        full_path = os.path.abspath(self.root_path / key)
        common_path = os.path.commonpath([str(self.root_path), full_path])
        if common_path != str(self.root_path):
            raise InvalidKeyException(
                f"Invalid key: {key}. Key should be relative to the full path."
                f"{self.root_path} vs. {common_path} and full path of {full_path}"
            )

        return Path(full_path)

    def _mkdir_for_store(self, dir: Path) -> None:
        """创建一个具有指定权限的存储目录路径（包括父目录）

这是因为`Path.mkdir()`受当前`umask`的限制，
而这里使用的显式`os.chmod()`则没有受到限制。

参数：
    dir: (Path) 要创建的存储目录

返回：
    无
"""
        if not dir.exists():
            self._mkdir_for_store(dir.parent)
            dir.mkdir(exist_ok=True)
            if self.chmod_dir is not None:
                os.chmod(dir, self.chmod_dir)

    def mget(self, keys: Sequence[str]) -> List[Optional[bytes]]:
        """获取与给定键相关联的值。

参数：
    keys：一系列键。

返回：
    与键相关联的可选值序列。
    如果未找到键，则相应的值将为None。
"""
        values: List[Optional[bytes]] = []
        for key in keys:
            full_path = self._get_full_path(key)
            if full_path.exists():
                value = full_path.read_bytes()
                values.append(value)
                if self.update_atime:
                    # update access time only; preserve modified time
                    os.utime(full_path, (time.time(), os.stat(full_path).st_mtime))
            else:
                values.append(None)
        return values

    def mset(self, key_value_pairs: Sequence[Tuple[str, bytes]]) -> None:
        """设置给定键的值。

参数：
    key_value_pairs：一组键值对。

返回：
    无。
"""
        for key, value in key_value_pairs:
            full_path = self._get_full_path(key)
            self._mkdir_for_store(full_path.parent)
            full_path.write_bytes(value)
            if self.chmod_file is not None:
                os.chmod(full_path, self.chmod_file)

    def mdelete(self, keys: Sequence[str]) -> None:
        """删除给定的键及其关联的值。

参数：
    keys（Sequence[str]）：要删除的键的序列。

返回：
    无
"""
        for key in keys:
            full_path = self._get_full_path(key)
            if full_path.exists():
                full_path.unlink()

    def yield_keys(self, prefix: Optional[str] = None) -> Iterator[str]:
        """获取与给定前缀匹配的键的迭代器。

参数：
    prefix（可选[str]）：要匹配的前缀。

返回：
    Iterator[str]：一个迭代器，用于匹配给定前缀的键。
"""
        prefix_path = self._get_full_path(prefix) if prefix else self.root_path
        for file in prefix_path.rglob("*"):
            if file.is_file():
                relative_path = file.relative_to(self.root_path)
                yield str(relative_path)