[DE-665]: GET /_api/transaction

arango/database.py

@@ -63,6 +63,7 @@
     TaskGetError,
     TaskListError,
     TransactionExecuteError,
+    TransactionListError,
     UserCreateError,
     UserDeleteError,
     UserGetError,
@@ -307,6 +308,25 @@ def response_handler(resp: Response) -> Any:
 
         return self._execute(request, response_handler)
 
+    def list_transactions(self) -> Result[Jsons]:
+        """Return the list of running stream transactions.
+
+        :return: The list of transactions, with each transaction
+          containing an "id" and a "state" field.
+        :rtype: List[Dict[str, Any]]
+        :raise arango.exceptions.TransactionListError: If retrieval fails.
+        """
+        request = Request(method="get", endpoint="/_api/transaction")
+
+        def response_handler(resp: Response) -> Jsons:
+            if not resp.is_success:
+                raise TransactionListError(resp, request)
+
+            result: Jsons = resp.body["transactions"]
+            return result
+
+        return self._execute(request, response_handler)
+
     def version(self, details: bool = False) -> Result[Any]:
         """Return ArangoDB server version.
         :param details: Return more detailed version output

arango/exceptions.py

@@ -724,6 +724,10 @@ class TransactionAbortError(ArangoServerError):
     """Failed to abort transaction."""
 
 
+class TransactionListError(ArangoServerError):
+    """Failed to retrieve transactions."""
+
+
 ###################
 # User Exceptions #
 ###################

tests/test_transaction.py

@@ -8,7 +8,7 @@
     TransactionInitError,
     TransactionStatusError,
 )
-from tests.helpers import extract
+from tests.helpers import extract, generate_db_name
 
 
 def test_transaction_execute_raw(db, col, docs):
@@ -149,3 +149,34 @@ def test_transaction_graph(db, graph, fvcol, fvdocs):
     assert len(vcol) == 0
 
     txn_db.commit_transaction()
+
+
+def test_transaction_list(client, sys_db, username, password):
+    db_name = generate_db_name()
+
+    sys_db.create_database(
+        name=db_name,
+        users=[{"username": username, "password": password, "active": True}],
+    )
+
+    db = client.db(db_name, username, password)
+
+    assert db.list_transactions() == []
+
+    txn_db = db.begin_transaction()
+    txn_db.aql.execute("RETURN 1")
+
+    txn_db_2 = db.begin_transaction()
+    txn_db_2.aql.execute("RETURN 1")
+
+    assert len(db.list_transactions()) == 2
+
+    txn_db.commit_transaction()
+
+    assert len(db.list_transactions()) == 1
+
+    txn_db_2.commit_transaction()
+
+    assert db.list_transactions() == []
+
+    sys_db.delete_database(db_name)