feat: Add document history command to python client

Author: dgiegold

terminusdb_client/client/Client.py

@@ -563,6 +563,109 @@ def get_commit_history(self, max_history: int = 500) -> list:
             raise ValueError("max_history needs to be non-negative.")
         return self.log(count=max_history)
 
+    def get_document_history(
+        self,
+        doc_id: str,
+        team: Optional[str] = None,
+        db: Optional[str] = None,
+        start: int = 0,
+        count: int = -1,
+        created: bool = False,
+        updated: bool = False,
+    ) -> list:
+        """Get the commit history for a specific document
+
+        Returns the history of changes made to a document, ordered backwards
+        in time from the most recent change. Only commits where the specified
+        document was created, modified, or deleted are included.
+
+        Parameters
+        ----------
+        doc_id : str
+            The document ID (IRI) to retrieve history for (e.g., "Person/alice")
+        team : str, optional
+            The team from which the database is. Defaults to the class property.
+        db : str, optional
+            The database. Defaults to the class property.
+        start : int, optional
+            Starting index for pagination. Defaults to 0.
+        count : int, optional
+            Maximum number of history entries to return. Defaults to -1 (all).
+        created : bool, optional
+            If True, return only the creation time. Defaults to False.
+        updated : bool, optional
+            If True, return only the last update time. Defaults to False.
+
+        Raises
+        ------
+        InterfaceError
+            If the client is not connected to a database
+        DatabaseError
+            If the API request fails or document is not found
+
+        Returns
+        -------
+        list
+            List of history entry dictionaries containing commit information
+            for the specified document:
+            ```
+            [
+              {
+                "author": "admin",
+                "identifier": "tbn15yq6rw1l4e9bgboyu3vwcoxgri5",
+                "message": "Updated document",
+                "timestamp": datetime.datetime(2023, 4, 6, 19, 1, 14, 324928)
+              },
+              {
+                "author": "admin",
+                "identifier": "3v3naa8jrt8612dg5zryu4vjqwa2w9s",
+                "message": "Created document",
+                "timestamp": datetime.datetime(2023, 4, 6, 19, 0, 47, 406387)
+              }
+            ]
+            ```
+
+        Example
+        -------
+        >>> from terminusdb_client import Client
+        >>> client = Client("http://127.0.0.1:6363")
+        >>> client.connect(db="example_db")
+        >>> history = client.get_document_history("Person/Jane")
+        >>> print(f"Document modified {len(history)} times")
+        >>> print(f"Last change by: {history[0]['author']}")
+        """
+        self._check_connection(check_db=(not team or not db))
+        team = team if team else self.team
+        db = db if db else self.db
+
+        params = {
+            'id': doc_id,
+            'start': start,
+            'count': count,
+        }
+
+        if created:
+            params['created'] = created
+        if updated:
+            params['updated'] = updated
+
+        result = self._session.get(
+            f"{self.api}/history/{team}/{db}",
+            params=params,
+            headers=self._default_headers,
+            auth=self._auth(),
+        )
+
+        history = json.loads(_finish_response(result))
+
+        # Post-process timestamps from Unix timestamp to datetime objects
+        if isinstance(history, list):
+            for entry in history:
+                if 'timestamp' in entry and isinstance(entry['timestamp'], (int, float)):
+                    entry['timestamp'] = datetime.fromtimestamp(entry['timestamp'])
+
+        return history
+
     def _get_current_commit(self):
         descriptor = self.db
         if self.branch:

terminusdb_client/tests/integration_tests/test_client.py

@@ -211,6 +211,70 @@ def test_log(docker_url):
     assert log[0]['@type'] == 'InitialCommit'
 
 
+def test_get_document_history(docker_url):
+    # Create client
+    client = Client(docker_url, user_agent=test_user_agent)
+    client.connect()
+
+    # Create test database
+    db_name = "testDB" + str(random())
+    client.create_database(db_name, team="admin")
+    client.connect(db=db_name)
+
+    # Add a schema
+    schema = {
+        "@type": "Class",
+        "@id": "Person",
+        "name": "xsd:string",
+        "age": "xsd:integer"
+    }
+    client.insert_document(schema, graph_type=GraphType.SCHEMA)
+
+    # Insert a document
+    person = {"@type": "Person", "@id": "Person/Jane", "name": "Jane", "age": 30}
+    client.insert_document(person, commit_msg="Created Person/Jane")
+
+    # Update the document to create history
+    person["name"] = "Jane Doe"
+    person["age"] = 31
+    client.update_document(person, commit_msg="Updated Person/Jane name and age")
+
+    # Update again
+    person["age"] = 32
+    client.update_document(person, commit_msg="Updated Person/Jane age")
+
+    # Get document history
+    history = client.get_document_history("Person/Jane")
+
+    # Assertions
+    assert isinstance(history, list)
+    assert len(history) >= 3  # At least insert and two updates
+    assert all('timestamp' in entry for entry in history)
+    assert all(isinstance(entry['timestamp'], dt.datetime) for entry in history)
+    assert all('author' in entry for entry in history)
+    assert all('message' in entry for entry in history)
+    assert all('identifier' in entry for entry in history)
+
+    # Verify messages are in the history (order may vary)
+    messages = [entry['message'] for entry in history]
+    assert "Created Person/Jane" in messages
+    assert "Updated Person/Jane name and age" in messages
+    assert "Updated Person/Jane age" in messages
+
+    # Test with pagination
+    paginated_history = client.get_document_history("Person/Jane", start=0, count=2)
+    assert len(paginated_history) == 2
+
+    # Test with team/db override
+    history_override = client.get_document_history(
+        "Person/Jane", team="admin", db=db_name
+    )
+    assert len(history_override) == len(history)
+
+    # Cleanup
+    client.delete_database(db_name, "admin")
+
+
 def test_get_triples(docker_url):
     client = Client(docker_url, user_agent=test_user_agent, team="admin")
     client.connect()

terminusdb_client/tests/test_Client.py

@@ -712,3 +712,39 @@ def test_get_user(mocked_requests):
         auth=("admin", "root"),
         headers={"user-agent": f"terminusdb-client-python/{__version__}"},
     )
+
+
+@mock.patch.object(requests.Session, 'head', side_effect=mocked_request_success)
+@mock.patch.object(requests.Session, 'get', side_effect=mocked_request_success)
+def test_get_document_history(mocked_get, mocked_head):
+    client = Client(
+        "http://localhost:6363", user="admin", key="root", team="admin"
+    )
+    client.connect(db="myDBName")
+
+    client.get_document_history("Person/Jane", start=0, count=10)
+
+    # Get the last call to get (should be our get_document_history call)
+    last_call = client._session.get.call_args_list[-1]
+    assert last_call[0][0] == "http://localhost:6363/api/history/admin/myDBName"
+    assert last_call[1]["params"] == {"id": "Person/Jane", "start": 0, "count": 10}
+    assert last_call[1]["headers"] == {"user-agent": f"terminusdb-client-python/{__version__}"}
+    assert last_call[1]["auth"] == ("admin", "root")
+
+
+@mock.patch.object(requests.Session, 'head', side_effect=mocked_request_success)
+@mock.patch.object(requests.Session, 'get', side_effect=mocked_request_success)
+def test_get_document_history_with_created_updated(mocked_get, mocked_head):
+    client = Client(
+        "http://localhost:6363", user="admin", key="root", team="admin"
+    )
+    client.connect(db="myDBName")
+
+    client.get_document_history("Person/Jane", created=True, updated=True)
+
+    # Get the last call to get (should be our get_document_history call)
+    last_call = client._session.get.call_args_list[-1]
+    assert last_call[0][0] == "http://localhost:6363/api/history/admin/myDBName"
+    assert last_call[1]["params"] == {"id": "Person/Jane", "start": 0, "count": -1, "created": True, "updated": True}
+    assert last_call[1]["headers"] == {"user-agent": f"terminusdb-client-python/{__version__}"}
+    assert last_call[1]["auth"] == ("admin", "root")