Added Taxonomy support.

CHANGELOG.md

@@ -1,5 +1,11 @@
 # CHANGELOG
 
+## _v2.3.0_
+
+### **Date: 21-July-2025**
+
+- Taxonomy Support Added.
+
 ## _v2.2.0_
 
 ### **Date: 14-July-2025**

contentstack/__init__.py

@@ -22,7 +22,7 @@
 __title__ = 'contentstack-delivery-python'
 __author__ = 'contentstack'
 __status__ = 'debug'
-__version__ = 'v2.2.0'
+__version__ = 'v2.3.0'
 __endpoint__ = 'cdn.contentstack.io'
 __email__ = '[email protected]'
 __developer_email__ = '[email protected]'

contentstack/stack.py

@@ -6,6 +6,7 @@
 from contentstack.asset import Asset
 from contentstack.assetquery import AssetQuery
 from contentstack.contenttype import ContentType
+from contentstack.taxonomy import Taxonomy
 from contentstack.globalfields import GlobalField
 from contentstack.https_connection import HTTPSConnection
 from contentstack.image_transform import ImageTransform
@@ -204,6 +205,14 @@ def content_type(self, content_type_uid=None):
         """
         return ContentType(self.http_instance, content_type_uid)
 
+    def taxonomy(self):
+        """
+        taxonomy defines the structure or schema of a page or a section
+        of your web or mobile property.
+        :return: taxonomy
+        """
+        return Taxonomy(self.http_instance)
+
     def global_field(self, global_field_uid=None):
         """
         Global field defines the structure or schema of a page or a section

contentstack/taxonomy.py

@@ -0,0 +1,64 @@
+import json
+from urllib import parse
+from urllib.parse import quote
+
+
+
+class Taxonomy:
+    def __init__(self, http_instance):
+        self.http_instance = http_instance
+        self._filters: dict = {}
+
+    def _add(self, field: str, condition: dict) -> "TaxonomyQuery":
+        self._filters[field] = condition
+        return self
+
+    def in_(self, field: str, terms: list) -> "TaxonomyQuery":
+        return self._add(field, {"$in": terms})
+
+    def or_(self, *conds: dict) -> "TaxonomyQuery":
+        return self._add("$or", list(conds))
+
+    def and_(self, *conds: dict) -> "TaxonomyQuery":
+        return self._add("$and", list(conds))
+
+    def exists(self, field: str) -> "TaxonomyQuery":
+        return self._add(field, {"$exists": True})
+
+    def equal_and_below(self, field: str, term_uid: str, levels: int = 10) -> "TaxonomyQuery":
+        cond = {"$eq_below": term_uid, "levels": levels}
+        return self._add(field, cond)
+
+    def below(self, field: str, term_uid: str, levels: int = 10) -> "TaxonomyQuery":
+        cond = {"$below": term_uid, "levels": levels}
+        return self._add(field, cond)
+
+    def equal_and_above(self, field: str, term_uid: str, levels: int = 10) -> "TaxonomyQuery":
+        cond = {"$eq_above": term_uid, "levels": levels}
+        return self._add(field, cond)
+
+    def above(self, field: str, term_uid: str, levels: int = 10) -> "TaxonomyQuery":
+        cond = {"$above": term_uid, "levels": levels}
+        return self._add(field, cond)
+
+    def find(self, params=None):
+        """
+        This method fetches entries filtered by taxonomy from the stack.
+        """
+        self.local_param = {}
+        self.local_param['environment'] = self.http_instance.headers['environment']
+
+        # Ensure query param is always present
+        query_string = json.dumps(self._filters or {})
+        query_encoded = quote(query_string, safe='{}":,[]')  # preserves JSON characters
+
+        # Build the base URL
+        endpoint = self.http_instance.endpoint
+        url = f'{endpoint}/taxonomies/entries?environment={self.local_param["environment"]}&query={query_encoded}'
+
+        # Append any additional params manually
+        if params:
+            other_params = '&'.join(f'{k}={v}' for k, v in params.items())
+            url += f'&{other_params}'
+        return self.http_instance.get(url)
+

requirements.txt

@@ -8,7 +8,7 @@ tox==4.5.1
 virtualenv==20.26.6
 Sphinx==7.3.7
 sphinxcontrib-websupport==1.2.7
-pip==23.3.1
+pip==25.1.1
 build==0.10.0
 wheel==0.45.1
 lxml==5.3.1

tests/__init__.py

@@ -16,6 +16,7 @@
 from .test_early_fetch import TestGlobalFieldFetch
 from .test_early_find import TestGlobalFieldFind
 from .test_live_preview import TestLivePreviewConfig
+from .test_taxonomies import TestTaxonomyAPI
 
 
 def all_tests():
@@ -27,6 +28,7 @@ def all_tests():
     test_module_globalFields = TestLoader().loadTestsFromName(TestGlobalFieldInit)
     test_module_globalFields_fetch = TestLoader().loadTestsFromName(TestGlobalFieldFetch)
     test_module_globalFields_find = TestLoader().loadTestsFromName(TestGlobalFieldFind)
+    test_module_taxonomies = TestLoader().loadTestsFromTestCase(TestTaxonomyAPI)
     TestSuite([
         test_module_stack,
         test_module_asset,
@@ -35,5 +37,6 @@ def all_tests():
         test_module_live_preview,
         test_module_globalFields,
         test_module_globalFields_fetch,
-        test_module_globalFields_find
+        test_module_globalFields_find,
+        test_module_taxonomies
     ])

tests/test_taxonomies.py

@@ -0,0 +1,99 @@
+import logging
+import unittest
+import config
+import contentstack
+import pytest
+
+API_KEY = config.APIKEY
+DELIVERY_TOKEN = config.DELIVERYTOKEN
+ENVIRONMENT = config.ENVIRONMENT
+HOST = config.HOST
+
+class TestTaxonomyAPI(unittest.TestCase):
+    def setUp(self):
+        self.stack = contentstack.Stack(API_KEY, DELIVERY_TOKEN, ENVIRONMENT, host=HOST)
+
+    def test_01_taxonomy_complex_query(self):
+        """Test complex taxonomy query combining multiple filters"""
+        taxonomy = self.stack.taxonomy()
+        result = taxonomy.and_(
+            {"taxonomies.category": {"$in": ["test"]}},
+            {"taxonomies.test1": {"$exists": True}}
+        ).or_(
+            {"taxonomies.status": {"$in": ["active"]}},
+            {"taxonomies.priority": {"$in": ["high"]}}
+        ).find({'limit': 10})
+        if result is not None:
+            self.assertIn('entries', result)
+
+    def test_02_taxonomy_in_query(self):
+        """Test taxonomy query with $in filter"""
+        taxonomy = self.stack.taxonomy()
+        result = taxonomy.in_("taxonomies.category", ["category1", "category2"]).find()
+        if result is not None:
+            self.assertIn('entries', result)
+
+    def test_03_taxonomy_exists_query(self):
+        """Test taxonomy query with $exists filter"""
+        taxonomy = self.stack.taxonomy()
+        result = taxonomy.exists("taxonomies.test1").find()
+        if result is not None:
+            self.assertIn('entries', result)
+
+    def test_04_taxonomy_or_query(self):
+        """Test taxonomy query with $or filter"""
+        taxonomy = self.stack.taxonomy()
+        result = taxonomy.or_(
+            {"taxonomies.category": {"$in": ["category1"]}},
+            {"taxonomies.test1": {"$exists": True}}
+        ).find()
+        if result is not None:
+            self.assertIn('entries', result)
+
+    def test_05_taxonomy_and_query(self):
+        """Test taxonomy query with $and filter"""
+        taxonomy = self.stack.taxonomy()
+        result = taxonomy.and_(
+            {"taxonomies.category": {"$in": ["category1"]}},
+            {"taxonomies.test1": {"$exists": True}}
+        ).find()
+        if result is not None:
+            self.assertIn('entries', result)
+
+    def test_06_taxonomy_equal_and_below(self):
+        """Test taxonomy query with $eq_below filter"""
+        taxonomy = self.stack.taxonomy()
+        result = taxonomy.equal_and_below("taxonomies.color", "blue", levels=1).find()
+        if result is not None:
+            self.assertIn('entries', result)
+
+    def test_07_taxonomy_below(self):
+        """Test taxonomy query with $below filter"""
+        taxonomy = self.stack.taxonomy()
+        result = taxonomy.below("taxonomies.hierarchy", "parent_uid", levels=2).find()
+        if result is not None:
+            self.assertIn('entries', result)
+
+    def test_08_taxonomy_equal_and_above(self):
+        """Test taxonomy query with $eq_above filter"""
+        taxonomy = self.stack.taxonomy()
+        result = taxonomy.equal_and_above("taxonomies.hierarchy", "child_uid", levels=3).find()
+        if result is not None:
+            self.assertIn('entries', result)
+
+    def test_09_taxonomy_above(self):
+        """Test taxonomy query with $above filter"""
+        taxonomy = self.stack.taxonomy()
+        result = taxonomy.above("taxonomies.hierarchy", "child_uid", levels=2).find()
+        if result is not None:
+            self.assertIn('entries', result)
+
+    def test_10_taxonomy_find_with_params(self):
+        """Test taxonomy find with additional parameters"""
+        taxonomy = self.stack.taxonomy()
+        result = taxonomy.in_("taxonomies.category", ["test"]).find({'limit': 5})
+        if result is not None:
+            self.assertIn('entries', result)
+
+if __name__ == '__main__':
+    unittest.main()