From b8dee7191e595e86a73ca91bc37320cbd4926e04 Mon Sep 17 00:00:00 2001 From: Dennis Giegold Date: Sun, 16 Nov 2025 22:19:12 +0100 Subject: [PATCH] feat: Add document history command to python client --- terminusdb_client/client/Client.py | 103 ++++++++++++++++++ .../tests/integration_tests/test_client.py | 64 +++++++++++ terminusdb_client/tests/test_Client.py | 36 ++++++ 3 files changed, 203 insertions(+) diff --git a/terminusdb_client/client/Client.py b/terminusdb_client/client/Client.py index 1b7ea8d6..f7860b11 100644 --- a/terminusdb_client/client/Client.py +++ b/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: diff --git a/terminusdb_client/tests/integration_tests/test_client.py b/terminusdb_client/tests/integration_tests/test_client.py index 2762b65d..63997ae2 100644 --- a/terminusdb_client/tests/integration_tests/test_client.py +++ b/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() diff --git a/terminusdb_client/tests/test_Client.py b/terminusdb_client/tests/test_Client.py index fadd11dd..a1e459a6 100644 --- a/terminusdb_client/tests/test_Client.py +++ b/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")