test_upsert_overwrites() — langchain Function Reference

Architecture documentation for the test_upsert_overwrites() function in indexer.py from the langchain codebase.

Function python LangChainCore Runnables calls 1 called by 1

Entity Profile

LangChainCore→ Runnables→ test_upsert_overwrites() — langchain Function Reference

Dependency Diagram

graph TD
  16e17ca5_c91d_ab92_7bec_8ed212d850a3["test_upsert_overwrites()"]
  091c1ddc_0019_3806_d313_14532add9618["AsyncDocumentIndexTestSuite"]
  16e17ca5_c91d_ab92_7bec_8ed212d850a3 -->|defined in| 091c1ddc_0019_3806_d313_14532add9618
  55e9955a_6054_786f_ce69_5fd8209628dc["test_upsert_overwrites()"]
  55e9955a_6054_786f_ce69_5fd8209628dc -->|calls| 16e17ca5_c91d_ab92_7bec_8ed212d850a3
  55e9955a_6054_786f_ce69_5fd8209628dc["test_upsert_overwrites()"]
  16e17ca5_c91d_ab92_7bec_8ed212d850a3 -->|calls| 55e9955a_6054_786f_ce69_5fd8209628dc
  style 16e17ca5_c91d_ab92_7bec_8ed212d850a3 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/standard-tests/langchain_tests/integration_tests/indexer.py lines 283–304

    async def test_upsert_overwrites(self, index: DocumentIndex) -> None:
        """Test that upsert overwrites existing content."""
        foo_uuid = str(uuid.UUID(int=7))
        documents = [
            Document(id=foo_uuid, page_content="foo", metadata={"bar": 1}),
        ]
        response = await index.aupsert(documents)
        ids = response["succeeded"]
        assert response["failed"] == []

        assert await index.aget(ids) == [
            Document(page_content="foo", metadata={"bar": 1}, id=foo_uuid),
        ]

        # Now let's overwrite foo
        await index.aupsert(
            [Document(id=foo_uuid, page_content="foo2", metadata={"meow": 2})]
        )
        documents = await index.aget([foo_uuid])
        assert documents == [
            Document(page_content="foo2", metadata={"meow": 2}, id=foo_uuid)
        ]