Add prefix management

Author: hoijnet

terminusdb_client/client/Client.py

@@ -670,7 +670,177 @@ def _get_prefixes(self):
             headers=self._default_headers,
             auth=self._auth(),
         )
-        return json.loads(_finish_response(result))
+        result.raise_for_status()
+        return result.json()
+
+    def get_prefix(self, prefix_name: str) -> str:
+        """Get a single prefix IRI by name.
+
+        Parameters
+        ----------
+        prefix_name : str
+            The prefix name to retrieve.
+
+        Returns
+        -------
+        str
+            The IRI (namespace URL) this prefix expands to.
+
+        Raises
+        ------
+        DatabaseError
+            If the prefix does not exist (404) or other API error.
+
+        Examples
+        --------
+        >>> client.get_prefix("schema")
+        'http://schema.org/'
+        """
+        self._check_connection()
+        result = self._session.get(
+            self._prefix_url(prefix_name),
+            headers=self._default_headers,
+            auth=self._auth(),
+        )
+        result.raise_for_status()
+        return result.json()["api:prefix_uri"]
+
+    def add_prefix(self, prefix_name: str, uri: str) -> dict:
+        """Add a new prefix mapping.
+
+        Parameters
+        ----------
+        prefix_name : str
+            The prefix name to create (must follow NCName rules).
+        uri : str
+            The IRI (namespace URL) this prefix expands to.
+
+        Returns
+        -------
+        dict
+            API response with status and details.
+
+        Raises
+        ------
+        DatabaseError
+            If prefix already exists or validation fails.
+
+        Examples
+        --------
+        >>> client.add_prefix("ex", "http://example.org/")
+        {'@type': 'api:PrefixAddResponse', 'api:status': 'api:success', ...}
+        """
+        self._check_connection()
+        result = self._session.post(
+            self._prefix_url(prefix_name),
+            json={"uri": uri},
+            headers=self._default_headers,
+            auth=self._auth(),
+        )
+        result.raise_for_status()
+        return result.json()
+
+    def update_prefix(self, prefix_name: str, uri: str) -> dict:
+        """Update an existing prefix mapping.
+
+        Parameters
+        ----------
+        prefix_name : str
+            The prefix name to update.
+        uri : str
+            The new IRI for this prefix.
+
+        Returns
+        -------
+        dict
+            API response with status and details.
+
+        Raises
+        ------
+        DatabaseError
+            If prefix does not exist (404) or validation fails.
+
+        Examples
+        --------
+        >>> client.update_prefix("ex", "http://example.com/")
+        {'@type': 'api:PrefixUpdateResponse', 'api:status': 'api:success', ...}
+        """
+        self._check_connection()
+        result = self._session.put(
+            self._prefix_url(prefix_name),
+            json={"uri": uri},
+            headers=self._default_headers,
+            auth=self._auth(),
+        )
+        result.raise_for_status()
+        return result.json()
+
+    def upsert_prefix(self, prefix_name: str, uri: str) -> dict:
+        """Create or update a prefix mapping (upsert).
+
+        Parameters
+        ----------
+        prefix_name : str
+            The prefix name.
+        uri : str
+            The IRI for this prefix.
+
+        Returns
+        -------
+        dict
+            API response with status and details.
+
+        Raises
+        ------
+        DatabaseError
+            If validation fails.
+
+        Examples
+        --------
+        >>> client.upsert_prefix("ex", "http://example.org/")
+        {'@type': 'api:PrefixUpdateResponse', 'api:status': 'api:success', ...}
+        """
+        self._check_connection()
+        result = self._session.put(
+            self._prefix_url(prefix_name) + "?create=true",
+            json={"uri": uri},
+            headers=self._default_headers,
+            auth=self._auth(),
+        )
+        result.raise_for_status()
+        return result.json()
+
+    def delete_prefix(self, prefix_name: str) -> dict:
+        """Delete a prefix mapping.
+
+        Parameters
+        ----------
+        prefix_name : str
+            The prefix name to delete.
+
+        Returns
+        -------
+        dict
+            API response with status.
+
+        Raises
+        ------
+        DatabaseError
+            If prefix does not exist (404) or is reserved.
+
+        Examples
+        --------
+        >>> client.delete_prefix("ex")
+        {'@type': 'api:PrefixDeleteResponse', 'api:status': 'api:success', ...}
+        """
+        self._check_connection()
+        result = self._session.delete(
+            self._prefix_url(prefix_name),
+            headers=self._default_headers,
+            auth=self._auth(),
+        )
+        result.raise_for_status()
+        return result.json()
 
     def create_database(
         self,
@@ -3086,3 +3256,12 @@ def _push_url(self):
 
     def _db_url(self):
         return self._db_base("db")
+
+    def _prefix_url(self, prefix_name: Optional[str] = None):
+        """Get URL for prefix operations"""
+        base = self._db_base("prefix")
+        if self._db == "_system":
+            return base if prefix_name is None else f"{base}/{urlparse.quote(prefix_name)}"
+        # For regular databases, include repo and branch
+        base = self._branch_base("prefix")
+        return base if prefix_name is None else f"{base}/{urlparse.quote(prefix_name)}"

terminusdb_client/tests/integration_tests/test_prefix_management.py

@@ -0,0 +1,135 @@
+"""Integration tests for prefix management operations."""
+import pytest
+from terminusdb_client.client import Client
+
+
+@pytest.fixture
+def client():
+    """Create a test client connected to TerminusDB."""
+    client = Client("http://127.0.0.1:6363")
+    client.connect(user="admin", key="root", team="admin")
+    return client
+
+
+@pytest.fixture
+def test_db(client):
+    """Create and cleanup a test database."""
+    import time
+    db_name = f"test_prefix_{int(time.time() * 1000)}"
+    
+    client.create_database(db_name, label="Test Prefix DB", description="Database for testing prefix operations")
+    client.connect(db=db_name)
+    
+    yield db_name
+    
+    # Cleanup
+    try:
+        client.delete_database(db_name)
+    except Exception:
+        pass
+
+
+class TestPrefixManagement:
+    """Test suite for prefix management operations."""
+
+    def test_add_prefix_success(self, client, test_db):
+        """Test adding a new prefix successfully."""
+        result = client.add_prefix("ex", "http://example.org/")
+        assert result["api:status"] == "api:success"
+        assert result["api:prefix_name"] == "ex"
+        assert result["api:prefix_uri"] == "http://example.org/"
+
+    def test_add_prefix_duplicate(self, client, test_db):
+        """Test that adding duplicate prefix fails."""
+        client.add_prefix("ex", "http://example.org/")
+        
+        with pytest.raises(Exception) as exc_info:
+            client.add_prefix("ex", "http://example.com/")
+        
+        # Should get 400 error with PrefixAlreadyExists
+        assert exc_info.value.response.status_code == 400
+
+    def test_add_prefix_invalid_iri(self, client, test_db):
+        """Test that invalid IRI is rejected."""
+        with pytest.raises(Exception) as exc_info:
+            client.add_prefix("ex", "not-a-valid-uri")
+        
+        assert exc_info.value.response.status_code == 400
+
+    def test_add_prefix_reserved(self, client, test_db):
+        """Test that reserved prefix names are rejected."""
+        with pytest.raises(Exception) as exc_info:
+            client.add_prefix("@custom", "http://example.org/")
+        
+        assert exc_info.value.response.status_code == 400
+
+    def test_get_prefix_success(self, client, test_db):
+        """Test retrieving an existing prefix."""
+        client.add_prefix("ex", "http://example.org/")
+        uri = client.get_prefix("ex")
+        assert uri == "http://example.org/"
+
+    def test_get_prefix_not_found(self, client, test_db):
+        """Test that getting non-existent prefix fails."""
+        with pytest.raises(Exception) as exc_info:
+            client.get_prefix("nonexistent")
+        
+        assert exc_info.value.response.status_code == 404
+
+    def test_update_prefix_success(self, client, test_db):
+        """Test updating an existing prefix."""
+        client.add_prefix("ex", "http://example.org/")
+        result = client.update_prefix("ex", "http://example.com/")
+        
+        assert result["api:status"] == "api:success"
+        assert result["api:prefix_uri"] == "http://example.com/"
+        
+        # Verify the update
+        uri = client.get_prefix("ex")
+        assert uri == "http://example.com/"
+
+    def test_update_prefix_not_found(self, client, test_db):
+        """Test that updating non-existent prefix fails."""
+        with pytest.raises(Exception) as exc_info:
+            client.update_prefix("nonexistent", "http://example.org/")
+        
+        assert exc_info.value.response.status_code == 404
+
+    def test_upsert_prefix_create(self, client, test_db):
+        """Test upsert creates prefix if it doesn't exist."""
+        result = client.upsert_prefix("ex", "http://example.org/")
+        assert result["api:status"] == "api:success"
+        assert result["api:prefix_uri"] == "http://example.org/"
+
+    def test_upsert_prefix_update(self, client, test_db):
+        """Test upsert updates prefix if it already exists."""
+        client.add_prefix("ex", "http://example.org/")
+        result = client.upsert_prefix("ex", "http://example.com/")
+        
+        assert result["api:status"] == "api:success"
+        assert result["api:prefix_uri"] == "http://example.com/"
+
+    def test_delete_prefix_success(self, client, test_db):
+        """Test deleting an existing prefix."""
+        client.add_prefix("ex", "http://example.org/")
+        result = client.delete_prefix("ex")
+        assert result["api:status"] == "api:success"
+        
+        # Verify deletion
+        with pytest.raises(Exception) as exc_info:
+            client.get_prefix("ex")
+        assert exc_info.value.response.status_code == 404
+
+    def test_delete_prefix_not_found(self, client, test_db):
+        """Test that deleting non-existent prefix fails."""
+        with pytest.raises(Exception) as exc_info:
+            client.delete_prefix("nonexistent")
+        
+        assert exc_info.value.response.status_code == 404
+
+    def test_delete_prefix_reserved(self, client, test_db):
+        """Test that deleting reserved prefix fails."""
+        with pytest.raises(Exception) as exc_info:
+            client.delete_prefix("@base")
+        
+        assert exc_info.value.response.status_code == 400