Updating edge

Author: apetenchea

arangoasync/collection.py

@@ -2307,3 +2307,81 @@ def response_handler(resp: Response) -> Json:
             raise DocumentInsertError(resp, request, msg)
 
         return await self._executor.execute(request, response_handler)
+
+    async def update(
+        self,
+        edge: T,
+        wait_for_sync: Optional[bool] = None,
+        keep_null: Optional[bool] = None,
+        return_new: Optional[bool] = None,
+        return_old: Optional[bool] = None,
+        if_match: Optional[str] = None,
+    ) -> Result[Json]:
+        """Update a vertex in the graph.
+
+        Args:
+            edge (dict): Partial or full document with the updated values.
+                It must contain the "_key" or "_id" field, along with "_from" and
+                "_to" fields.
+            wait_for_sync (bool | None): Wait until document has been synced to disk.
+            keep_null (bool | None): If the intention is to delete existing attributes
+                with the patch command, set this parameter to `False`.
+            return_new (bool | None): Additionally return the complete new document
+                under the attribute `new` in the result.
+            return_old (bool | None): Additionally return the complete old document
+                under the attribute `old` in the result.
+            if_match (str | None): You can conditionally update a document based on a
+                target revision id by using the "if-match" HTTP header.
+
+        Returns:
+            dict: Document metadata (e.g. document id, key, revision).
+                If `return_new` or "return_old" are specified, the result contains
+                the document metadata in the "vertex" field and two additional fields
+                ("new" and "old").
+
+        Raises:
+            DocumentUpdateError: If update fails.
+
+        References:
+            - `update-an-edge <https://docs.arangodb.com/stable/develop/http-api/graphs/named-graphs/#update-an-edge>`__
+        """  # noqa: E501
+        params: Params = {}
+        if wait_for_sync is not None:
+            params["waitForSync"] = wait_for_sync
+        if keep_null is not None:
+            params["keepNull"] = keep_null
+        if return_new is not None:
+            params["returnNew"] = return_new
+        if return_old is not None:
+            params["returnOld"] = return_old
+
+        headers: RequestHeaders = {}
+        if if_match is not None:
+            headers["If-Match"] = if_match
+
+        request = Request(
+            method=Method.PATCH,
+            endpoint=f"/_api/gharial/{self._graph}/edge/"
+            f"{self._prep_from_doc(cast(Json, edge))}",
+            params=params,
+            headers=headers,
+            data=self._doc_serializer.dumps(edge),
+        )
+
+        def response_handler(resp: Response) -> Json:
+            if resp.is_success:
+                return self._parse_result(self.deserializer.loads(resp.raw_body))
+            msg: Optional[str] = None
+            if resp.status_code == HTTP_PRECONDITION_FAILED:
+                raise DocumentRevisionError(resp, request)
+            elif resp.status_code == HTTP_NOT_FOUND:
+                msg = (
+                    "The graph cannot be found or the edge collection is not "
+                    "part of the graph. It is also possible that the vertex "
+                    "collection referenced in the _from or _to attribute is not part "
+                    "of the graph or the vertex collection is part of the graph, but "
+                    "does not exist. Finally check that _from or _to vertex do exist."
+                )
+            raise DocumentUpdateError(resp, request, msg)
+
+        return await self._executor.execute(request, response_handler)

arangoasync/graph.py

@@ -828,3 +828,50 @@ async def insert_edge(
             wait_for_sync=wait_for_sync,
             return_new=return_new,
         )
+
+    async def update_edge(
+        self,
+        edge: T,
+        wait_for_sync: Optional[bool] = None,
+        keep_null: Optional[bool] = None,
+        return_new: Optional[bool] = None,
+        return_old: Optional[bool] = None,
+        if_match: Optional[str] = None,
+    ) -> Result[Json]:
+        """Update a vertex in the graph.
+
+        Args:
+            edge (dict): Partial or full document with the updated values.
+                It must contain the "_key" or "_id" field, along with "_from" and
+                "_to" fields.
+            wait_for_sync (bool | None): Wait until document has been synced to disk.
+            keep_null (bool | None): If the intention is to delete existing attributes
+                with the patch command, set this parameter to `False`.
+            return_new (bool | None): Additionally return the complete new document
+                under the attribute `new` in the result.
+            return_old (bool | None): Additionally return the complete old document
+                under the attribute `old` in the result.
+            if_match (str | None): You can conditionally update a document based on a
+                target revision id by using the "if-match" HTTP header.
+
+        Returns:
+            dict: Document metadata (e.g. document id, key, revision).
+                If `return_new` or "return_old" are specified, the result contains
+                the document metadata in the "vertex" field and two additional fields
+                ("new" and "old").
+
+        Raises:
+            DocumentUpdateError: If update fails.
+
+        References:
+            - `update-an-edge <https://docs.arangodb.com/stable/develop/http-api/graphs/named-graphs/#update-an-edge>`__
+        """  # noqa: E501
+        col = Collection.get_col_name(cast(Json | str, edge))
+        return await self.edge_collection(col).update(
+            edge,
+            wait_for_sync=wait_for_sync,
+            keep_null=keep_null,
+            return_new=return_new,
+            return_old=return_old,
+            if_match=if_match,
+        )

tests/conftest.py

@@ -245,7 +245,7 @@ def db_version():
     return global_data.db_version
 
 
-@pytest_asyncio.fixture(scope="session", autouse=True)
+@pytest_asyncio.fixture(autouse=True)
 async def teardown():
     yield
     async with ArangoClient(hosts=global_data.url) as client:

tests/test_graph.py

@@ -271,6 +271,14 @@ async def test_edge_collections(db, bad_graph):
     edge = await graph.edge(edge_id)
     assert edge is not None
 
+    # Update an edge
+    edge["subject"] = "Advanced Math"
+    updated_edge_meta = await graph.update_edge(edge, return_new=True, return_old=True)
+    assert "new" in updated_edge_meta
+    assert "old" in updated_edge_meta
+    edge = await graph.edge(edge_id)
+    assert edge["subject"] == "Advanced Math"
+
     # Replace the edge definition
     new_from_collections = [students_col_name]
     new_to_collections = [teachers_col_name]