From 899442993953152b3388aaa92d20196df5286549 Mon Sep 17 00:00:00 2001 From: sunil-lakshman <104969541+sunil-lakshman@users.noreply.github.com> Date: Mon, 14 Jul 2025 20:04:37 +0530 Subject: [PATCH] Added Taxonomy support. --- CHANGELOG.md | 6 +++ contentstack/__init__.py | 2 +- contentstack/stack.py | 9 ++++ contentstack/taxonomy.py | 64 ++++++++++++++++++++++++++ requirements.txt | 2 +- tests/__init__.py | 5 +- tests/test_taxonomies.py | 99 ++++++++++++++++++++++++++++++++++++++++ 7 files changed, 184 insertions(+), 3 deletions(-) create mode 100644 contentstack/taxonomy.py create mode 100644 tests/test_taxonomies.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 22b14f4..c3dfaaf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,11 @@ # CHANGELOG +## _v2.3.0_ + +### **Date: 21-July-2025** + +- Taxonomy Support Added. + ## _v2.2.0_ ### **Date: 14-July-2025** diff --git a/contentstack/__init__.py b/contentstack/__init__.py index d8f02b0..a44eb95 100644 --- a/contentstack/__init__.py +++ b/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__ = 'support@contentstack.com' __developer_email__ = 'mobile@contentstack.com' diff --git a/contentstack/stack.py b/contentstack/stack.py index 856d287..eb3567c 100644 --- a/contentstack/stack.py +++ b/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 diff --git a/contentstack/taxonomy.py b/contentstack/taxonomy.py new file mode 100644 index 0000000..b859a68 --- /dev/null +++ b/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) + diff --git a/requirements.txt b/requirements.txt index 36a1196..e47143c 100644 --- a/requirements.txt +++ b/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 diff --git a/tests/__init__.py b/tests/__init__.py index 33fb399..752c4f3 100644 --- a/tests/__init__.py +++ b/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 ]) diff --git a/tests/test_taxonomies.py b/tests/test_taxonomies.py new file mode 100644 index 0000000..b4c14c0 --- /dev/null +++ b/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()