Replace deprecated APIs with modern equivalents (#143)

Author: wnqqnw19

examples/projects/twitter/services/pathway-app/app/main.py

@@ -91,23 +91,26 @@ def get_data_table(dataset_path, poll_new_objects):
         "user": "postgres",
         "password": "changeme",
     }
-    pw.io.postgres.write_snapshot(
+    pw.io.postgres.write(
         author_meta,
         postgres_settings=postgres_settings,
         table_name="author_meta",
-        primary_key=["tweet_to_author_id"],
+        output_table_type="snapshot",
+        primary_key=[author_meta.tweet_to_author_id],
     )
-    pw.io.postgres.write_snapshot(
+    pw.io.postgres.write(
         grouped,
         postgres_settings=postgres_settings,
         table_name="grouped",
-        primary_key=["tweet_to_author_id", "time_bucket"],
+        output_table_type="snapshot",
+        primary_key=[grouped.tweet_to_author_id, grouped.time_bucket],
     )
-    pw.io.postgres.write_snapshot(
+    pw.io.postgres.write(
         tweet_pairs,
         postgres_settings=postgres_settings,
         table_name="tweet_pairs",
-        primary_key=["tweet_from_id", "tweet_to_id"],
+        output_table_type="snapshot",
+        primary_key=[tweet_pairs.tweet_from_id, tweet_pairs.tweet_to_id],
     )
 
     pw.run()

integration_tests/rag_evals/connector.py

@@ -1,8 +1,6 @@
-import json
-
 import httpx
-import requests
 
+from pathway.xpacks.llm.document_store import DocumentStoreClient
 from pathway.xpacks.llm.question_answering import RAGClient, send_post_request
 
 
@@ -15,125 +13,8 @@ async def a_send_post_request(
         return response.json()
 
 
-class VectorStoreClient:
-    """
-    A client you can use to query VectorStoreServer.
-
-    Please provide either the `url`, or `host` and `port`.
-
-    Args:
-        host: host on which `VectorStoreServer </developers/api-docs/pathway-xpacks-llm/vectorstore#pathway.xpacks.llm.vector_store.VectorStoreServer>`_ listens
-        port: port on which `VectorStoreServer </developers/api-docs/pathway-xpacks-llm/vectorstore#pathway.xpacks.llm.vector_store.VectorStoreServer>`_ listens
-        url: url at which `VectorStoreServer </developers/api-docs/pathway-xpacks-llm/vectorstore#pathway.xpacks.llm.vector_store.VectorStoreServer>`_ listens
-        timeout: timeout for the post requests in seconds
-    """  # noqa
-
-    def __init__(
-        self,
-        host: str | None = None,
-        port: int | None = None,
-        url: str | None = None,
-        timeout: int | None = 180,
-        additional_headers: dict | None = None,
-    ):
-        err = "Either (`host` and `port`) or `url` must be provided, but not both."
-        if url is not None:
-            if host or port:
-                raise ValueError(err)
-            self.url = url
-        else:
-            if host is None:
-                raise ValueError(err)
-            port = port or 80
-            self.url = f"http://{host}:{port}"
-
-        self.timeout = timeout
-        self.additional_headers = additional_headers or {}
-
-    def query(
-        self,
-        query: str,
-        k: int = 3,
-        metadata_filter: str | None = None,
-        filepath_globpattern: str | None = None,
-    ) -> list[dict]:
-        """
-        Perform a query to the vector store and fetch results.
-
-        Args:
-            query:
-            k: number of documents to be returned
-            metadata_filter: optional string representing the metadata filtering query
-                in the JMESPath format. The search will happen only for documents
-                satisfying this filtering.
-            filepath_globpattern: optional glob pattern specifying which documents
-                will be searched for this query.
-        """
-
-        data = {"query": query, "k": k}
-        if metadata_filter is not None:
-            data["metadata_filter"] = metadata_filter
-        if filepath_globpattern is not None:
-            data["filepath_globpattern"] = filepath_globpattern
-        url = self.url + "/v1/retrieve"
-        response = requests.post(
-            url,
-            data=json.dumps(data),
-            headers=self._get_request_headers(),
-            timeout=self.timeout,
-        )
-
-        responses = response.json()
-        return sorted(responses, key=lambda x: x["dist"])
-
-    # Make an alias
-    __call__ = query
-
-    def get_vectorstore_statistics(self):
-        """Fetch basic statistics about the vector store."""
-
-        url = self.url + "/v1/statistics"
-        response = requests.post(
-            url,
-            json={},
-            headers=self._get_request_headers(),
-            timeout=self.timeout,
-        )
-        responses = response.json()
-        return responses
-
-    def get_input_files(
-        self,
-        metadata_filter: str | None = None,
-        filepath_globpattern: str | None = None,
-    ):
-        """
-        Fetch information on documents in the the vector store.
-
-        Args:
-            metadata_filter: optional string representing the metadata filtering query
-                in the JMESPath format. The search will happen only for documents
-                satisfying this filtering.
-            filepath_globpattern: optional glob pattern specifying which documents
-                will be searched for this query.
-        """
-        url = self.url + "/v1/inputs"
-        response = requests.post(
-            url,
-            json={
-                "metadata_filter": metadata_filter,
-                "filepath_globpattern": filepath_globpattern,
-            },
-            headers=self._get_request_headers(),
-            timeout=self.timeout,
-        )
-        responses = response.json()
-        return responses
-
-    def _get_request_headers(self):
-        request_headers = {"Content-Type": "application/json"}
-        request_headers.update(self.additional_headers)
-        return request_headers
+# Use DocumentStoreClient from pathway instead of local implementation
+VectorStoreClient = DocumentStoreClient
 
 
 class RagConnector:
@@ -142,7 +23,7 @@ class RagConnector:
     def __init__(self, base_url: str):
         self.base_url = base_url
 
-        self.index_client = VectorStoreClient(
+        self.index_client = DocumentStoreClient(
             url=base_url,
         )
 

integration_tests/webserver/test_llm_xpack.py

@@ -17,10 +17,11 @@
 import pathway as pw
 from pathway.internals.udfs.caches import InMemoryCache
 from pathway.tests.utils import wait_result_with_checker
+from pathway.xpacks.llm.document_store import DocumentStoreClient
 from pathway.xpacks.llm.question_answering import BaseRAGQuestionAnswerer, RAGClient
 from pathway.xpacks.llm.tests.mocks import FakeChatModel, fake_embeddings_model
 from pathway.xpacks.llm.tests.utils import build_vector_store, create_build_rag_app
-from pathway.xpacks.llm.vector_store import VectorStoreClient, VectorStoreServer
+from pathway.xpacks.llm.vector_store import VectorStoreServer
 
 PATHWAY_HOST = "127.0.0.1"
 
@@ -85,7 +86,7 @@ def test_similarity_search_without_metadata(tmp_path: pathlib.Path, port: int):
     with open(tmp_path / "file_one.txt", "w+") as f:
         f.write("foo")
 
-    client = VectorStoreClient(host=PATHWAY_HOST, port=port)
+    client = DocumentStoreClient(host=PATHWAY_HOST, port=port)
 
     def checker() -> bool:
         output = []
@@ -109,7 +110,7 @@ def test_vector_store_with_langchain(tmp_path: pathlib.Path, port) -> None:
     with open(tmp_path / "file_one.txt", "w+") as f:
         f.write("foo\n\nbar")
 
-    client = VectorStoreClient(host=PATHWAY_HOST, port=port)
+    client = DocumentStoreClient(host=PATHWAY_HOST, port=port)
 
     def checker() -> bool:
         output = []
@@ -281,7 +282,7 @@ def fake_embeddings_model(x: str) -> list[float]:
 
     def checker() -> bool:
         try:
-            client = VectorStoreClient(host=PATHWAY_HOST, port=port)
+            client = DocumentStoreClient(host=PATHWAY_HOST, port=port)
             inputs = client.get_input_files()
 
             assert len(inputs) == 1
@@ -529,7 +530,7 @@ def test_serve_callable_with_search(port: int):
 
     @rag_app.serve_callable(route=f"/{TEST_ENDPOINT}")
     async def return_top_doc_text(query):
-        vs_client = VectorStoreClient(host=PATHWAY_HOST, port=port)
+        vs_client = DocumentStoreClient(host=PATHWAY_HOST, port=port)
         return vs_client.query(query, k=1)[0]["text"]
 
     def checker() -> bool:

integration_tests/webserver/test_rest_connector.py

@@ -45,9 +45,18 @@ def target() -> None:
         r.raise_for_status()
         assert r.text == '"TWO"', r.text
 
-    queries, response_writer = pw.io.http.rest_connector(
-        host="127.0.0.1", port=port, schema=InputSchema, delete_completed_queries=True
-    )
+    if isinstance(port, str):
+        queries, response_writer = pw.io.http.rest_connector(
+            host="127.0.0.1",
+            port=port,
+            schema=InputSchema,
+            delete_completed_queries=True,
+        )
+    else:
+        webserver = pw.io.http.PathwayWebserver(host="127.0.0.1", port=port)
+        queries, response_writer = pw.io.http.rest_connector(
+            webserver=webserver, schema=InputSchema, delete_completed_queries=True
+        )
     responses = logic(queries)
     response_writer(responses)
     pw.io.csv.write(queries, output_path)
@@ -89,9 +98,9 @@ def target() -> None:
             json={"query": "two"},
         ).raise_for_status()
 
+    webserver = pw.io.http.PathwayWebserver(host="127.0.0.1", port=port)
     queries, response_writer = pw.io.http.rest_connector(
-        host="127.0.0.1",
-        port=port,
+        webserver=webserver,
         schema=InputSchema,
         route="/endpoint",
         delete_completed_queries=True,
@@ -129,8 +138,9 @@ def target() -> None:
             json={"query": "two"},
         ).raise_for_status()
 
+    webserver = pw.io.http.PathwayWebserver(host="127.0.0.1", port=port)
     queries, response_writer = pw.io.http.rest_connector(
-        host="127.0.0.1", port=port, schema=InputSchema, delete_completed_queries=True
+        webserver=webserver, schema=InputSchema, delete_completed_queries=True
     )
     responses = logic(queries)
     response_writer(responses)
@@ -160,8 +170,9 @@ def target() -> None:
             json={"k": 1, "v": 2},
         ).raise_for_status()
 
+    webserver = pw.io.http.PathwayWebserver(host="127.0.0.1", port=port)
     queries, response_writer = pw.io.http.rest_connector(
-        host="127.0.0.1", port=port, schema=InputSchema, delete_completed_queries=False
+        webserver=webserver, schema=InputSchema, delete_completed_queries=False
     )
     response_writer(queries.select(query_id=queries.id, result=pw.this.v))
 
@@ -193,39 +204,20 @@ def target() -> None:
 
 
 def test_server_fail_on_duplicate_port(tmp_path: pathlib.Path, port: int) -> None:
-    output_path = tmp_path / "output.csv"
-
     class InputSchema(pw.Schema):
         k: int
         v: int
 
+    webserver = pw.io.http.PathwayWebserver(host="127.0.0.1", port=port)
     queries, response_writer = pw.io.http.rest_connector(
-        host="127.0.0.1", port=port, schema=InputSchema, delete_completed_queries=False
+        webserver=webserver, schema=InputSchema, delete_completed_queries=False
     )
     response_writer(queries.select(query_id=queries.id, result=pw.this.v))
 
-    queries_dup, response_writer_dup = pw.io.http.rest_connector(
-        host="127.0.0.1", port=port, schema=InputSchema, delete_completed_queries=False
-    )
-    response_writer_dup(queries_dup.select(query_id=queries_dup.id, result=pw.this.v))
-
-    sum = queries.groupby(pw.this.k).reduce(
-        key=pw.this.k, sum=pw.reducers.sum(pw.this.v)
-    )
-    sum_dup = queries_dup.groupby(pw.this.k).reduce(
-        key=pw.this.k, sum=pw.reducers.sum(pw.this.v)
-    )
-
-    pw.io.csv.write(sum, output_path)
-    pw.io.csv.write(sum_dup, output_path)
-
-    with pytest.raises(OSError) as exc_info:
-        pw.run()
-    error_message = str(exc_info.value)
-    assert (
-        "error while attempting to bind on address" in error_message
-        or "Address already in use" in error_message
-    )
+    with pytest.raises(RuntimeError, match="Added route will never be executed"):
+        queries_dup, response_writer_dup = pw.io.http.rest_connector(
+            webserver=webserver, schema=InputSchema, delete_completed_queries=False
+        )
 
 
 def _test_server_two_endpoints(

python/pathway/tests/test_column_properties.py

@@ -216,9 +216,9 @@ def test_rest_connector(delete_completed_queries: bool):
     class TestSchema(pw.Schema):
         a: int
 
+    webserver = io.http.PathwayWebserver(host="127.0.0.1", port=30000)
     table, response_writer = io.http.rest_connector(
-        host="127.0.0.1",
-        port=30000,  # server is not started, port number does not matter
+        webserver=webserver,
         schema=TestSchema,
         delete_completed_queries=delete_completed_queries,
     )

python/pathway/xpacks/llm/question_answering.py

@@ -14,16 +14,13 @@
 from pathway.xpacks.llm import Doc, llms, prompts
 from pathway.xpacks.llm.document_store import (
     DocumentStore,
+    DocumentStoreClient,
     SlidesDocumentStore,
     _get_jmespath_filter,
 )
 from pathway.xpacks.llm.llms import BaseChat, prompt_chat_single_qa
 from pathway.xpacks.llm.mcp_server import McpServable, McpServer
-from pathway.xpacks.llm.vector_store import (
-    SlidesVectorStoreServer,
-    VectorStoreClient,
-    VectorStoreServer,
-)
+from pathway.xpacks.llm.vector_store import SlidesVectorStoreServer, VectorStoreServer
 
 if TYPE_CHECKING:
     from pathway.xpacks.llm.servers import QARestServer, QASummaryRestServer
@@ -1106,7 +1103,7 @@ def __init__(
         self.timeout = timeout
         self.additional_headers = additional_headers or {}
 
-        self.index_client = VectorStoreClient(
+        self.index_client = DocumentStoreClient(
             url=self.url,
             timeout=self.timeout,
             additional_headers=self.additional_headers,