CustomPersister Class — langchain Architecture

Source Code

libs/standard-tests/langchain_tests/conftest.py lines 55–89

class CustomPersister:
    """A custom persister for VCR that uses the `CustomSerializer`."""

    @classmethod
    def load_cassette(
        cls,
        cassette_path: str | PathLike[str],
        serializer: CustomSerializer,
    ) -> tuple[list[Any], list[Any]]:
        """Load a cassette from a file."""
        # If cassette path is already Path this is a no-op
        cassette_path = Path(cassette_path)
        if not cassette_path.is_file():
            msg = f"Cassette file {cassette_path} does not exist."
            raise CassetteNotFoundError(msg)
        with cassette_path.open(mode="rb") as f:
            data = f.read()
        deser = serializer.deserialize(data)
        return deser["requests"], deser["responses"]

    @staticmethod
    def save_cassette(
        cassette_path: str | PathLike[str],
        cassette_dict: dict[str, Any],
        serializer: CustomSerializer,
    ) -> None:
        """Save a cassette to a file."""
        data = serializer.serialize(cassette_dict)
        # if cassette path is already Path this is no operation
        cassette_path = Path(cassette_path)
        cassette_folder = cassette_path.parent
        if not cassette_folder.exists():
            cassette_folder.mkdir(parents=True)
        with cassette_path.open("wb") as f:
            f.write(data)