feat: add multi-region URL resolver and bump version to 1.6.0

.gitignore

@@ -113,3 +113,4 @@ venv.bak/
 .mypy_cache/
 .idea/
 .vscode/
+*/assets/regions.json

CHANGELOG.md

@@ -1,5 +1,17 @@
 # Changelog
 
+## v1.6.0 (2026-06-05)
+
+### New feature: Multi-region endpoint resolution
+
+- Added `Endpoint.get_contentstack_endpoint()` for dynamic region-aware URL resolution across all Contentstack regions and services.
+- Added `Utils.get_contentstack_endpoint()` proxy for backward-compatible access via the existing `Utils` import path.
+- Added `getContentstackEndpoint` camelCase alias on both `Endpoint` and `Utils` for cross-SDK parity.
+- Bundled `contentstack_utils/assets/regions.json` — the authoritative registry of 7 regions (AWS NA/EU/AU, Azure NA/EU, GCP NA/EU) and 18 service endpoint keys.
+- Added runtime fallback in `Endpoint._load_regions()` — downloads `regions.json` from `artifacts.contentstack.com` on first use when the file is absent.
+- Added `scripts/refresh_regions.py` to manually pull the latest regions from Contentstack .
+- Exported `Endpoint` at package level in `__all__`.
+
 ## v1.5.0
 
 ### New feature: Variants utility (CDA entry variant aliases)

contentstack_utils/__init__.py

@@ -10,6 +10,7 @@
 
 from contentstack_utils.embedded.item_type import ItemType
 from contentstack_utils.embedded.styletype import StyleType
+from contentstack_utils.endpoint import Endpoint
 from contentstack_utils.helper.metadata import Metadata
 from contentstack_utils.helper.node_to_html import NodeToHtml
 from contentstack_utils.render.options import Options
@@ -19,6 +20,7 @@
 from contentstack_utils.entry_editable import addEditableTags, addTags, getTag
 
 __all__ = (
+"Endpoint",
 "Utils",
 "Options",
 "Metadata",
@@ -35,6 +37,6 @@
 __title__ = 'contentstack_utils'
 __author__ = 'contentstack'
 __status__ = 'debug'
-__version__ = '1.4.0'
+__version__ = '1.6.0'
 __endpoint__ = 'cdn.contentstack.io'
 __contact__ = 'support@contentstack.com'

contentstack_utils/endpoint.py

@@ -0,0 +1,242 @@
+"""
+Endpoint resolution for Contentstack services across all regions.
+
+Reads a bundled regions.json (src/assets/regions.json) and resolves
+the correct base URL for any region + service combination. No runtime
+HTTP calls — the file is shipped with the package and updated via
+``python scripts/refresh_regions.py``.
+"""
+
+import json
+import os
+import re
+import urllib.request
+from typing import Dict, Optional, Union
+
+_REGIONS_URL = "https://artifacts.contentstack.com/regions.json"
+_ASSETS_DIR = os.path.join(os.path.dirname(__file__), "assets")
+_REGIONS_FILE = os.path.join(_ASSETS_DIR, "regions.json")
+
+
+class Endpoint:
+    """
+    Resolve Contentstack service URLs for any region.
+
+    All public methods are static — no instantiation required.
+
+    Example::
+
+        from contentstack_utils import Endpoint
+
+        # Full URL
+        url = Endpoint.get_contentstack_endpoint("na", "contentDelivery")
+        # → "https://cdn.contentstack.io"
+
+        # Host only (strip https://) — useful for SDK setHost() calls
+        host = Endpoint.get_contentstack_endpoint("eu", "contentDelivery", omit_https=True)
+        # → "eu-cdn.contentstack.com"
+
+        # All endpoints for a region
+        all_endpoints = Endpoint.get_contentstack_endpoint("azure-na")
+        # → {"contentDelivery": "...", "contentManagement": "...", ...}
+    """
+
+    # Module-level cache — loaded once per Python process, shared across all calls.
+    _regions_data: Optional[Dict] = None
+
+    @staticmethod
+    def get_contentstack_endpoint(
+        region: str = "us",
+        service: str = "",
+        omit_https: bool = False,
+    ) -> Union[str, Dict[str, str]]:
+        """
+        Resolve a Contentstack service endpoint URL for the given region.
+
+        :param region:
+            Region ID or any accepted alias (case-insensitive, ``-`` and ``_``
+            are interchangeable). Examples: ``'na'``, ``'us'``, ``'eu'``,
+            ``'AWS-NA'``, ``'azure_eu'``, ``'gcp-na'``.
+            Defaults to ``'us'`` (AWS North America).
+        :param service:
+            Optional service key. When provided, a single URL string is
+            returned. When omitted, a dict of **all** service URLs is returned.
+            Valid keys include: ``'contentDelivery'``, ``'contentManagement'``,
+            ``'auth'``, ``'graphqlDelivery'``, ``'preview'``, ``'images'``,
+            ``'assets'``, ``'automate'``, ``'launch'``, ``'developerHub'``,
+            ``'brandKit'``, ``'genAI'``, ``'personalizeManagement'``,
+            ``'personalizeEdge'``, ``'composableStudio'``, ``'assetManagement'``.
+        :param omit_https:
+            When ``True``, strips the ``https://`` (or ``http://``) scheme from
+            every returned URL. Useful when passing the host to an SDK that
+            constructs its own URLs (e.g. ``stack.set_host(host)``).
+        :returns:
+            - A ``str`` URL when *service* is specified.
+            - A ``dict[str, str]`` mapping service keys → URLs when *service*
+              is omitted.
+        :raises ValueError:
+            If *region* is an empty string.
+        :raises LookupError:
+            If *region* does not match any known region ID or alias, or if
+            *service* is not present in the resolved region's endpoint map.
+        :raises RuntimeError:
+            If the bundled ``regions.json`` cannot be read or is malformed.
+
+        Examples::
+
+            Endpoint.get_contentstack_endpoint("na", "contentDelivery")
+            # → "https://cdn.contentstack.io"
+
+            Endpoint.get_contentstack_endpoint("eu", "contentDelivery", omit_https=True)
+            # → "eu-cdn.contentstack.com"
+
+            Endpoint.get_contentstack_endpoint("azure-na")
+            # → {"contentDelivery": "https://...", ...}
+        """
+        if not region:
+            raise ValueError("Empty region provided. Please put valid region.")
+
+        data = Endpoint._load_regions()
+        normalized = region.strip().lower()
+
+        if not normalized:
+            raise ValueError("Empty region provided. Please put valid region.")
+        region_row = Endpoint._find_region_by_id_or_alias(data["regions"], normalized)
+
+        if region_row is None:
+            raise LookupError(f"Invalid region: {region}")
+
+        if service:
+            endpoints = region_row["endpoints"]
+            if service not in endpoints:
+                raise LookupError(
+                    f'Service "{service}" not found for region "{region_row["id"]}"'
+                )
+            url = endpoints[service]
+            return Endpoint._strip_https(url) if omit_https else url
+
+        endpoints = dict(region_row["endpoints"])
+        return Endpoint._strip_https_from_map(endpoints) if omit_https else endpoints
+
+
+    getContentstackEndpoint = get_contentstack_endpoint
+
+    @staticmethod
+    def reset_cache() -> None:
+        """
+        Clear the in-memory region cache.
+
+        Intended for testing only — forces the next call to re-read
+        ``regions.json`` from disk.
+        """
+        Endpoint._regions_data = None
+
+    # ------------------------------------------------------------------
+    # Internal helpers
+    # ------------------------------------------------------------------
+
+    @staticmethod
+    def _load_regions() -> Dict:
+        """
+        Load and cache regions data.
+
+        Resolution order:
+          1. In-memory cache (zero I/O after the first call in a process)
+          2. Bundled ``contentstack_utils/assets/regions.json`` on disk
+          3. Live download from ``artifacts.contentstack.com`` (fallback when
+             the file is absent — e.g. an editable install without assets)
+        """
+        if Endpoint._regions_data is not None:
+            return Endpoint._regions_data
+
+        if not os.path.exists(_REGIONS_FILE):
+            Endpoint._download_and_save(_REGIONS_FILE)
+
+        if not os.path.exists(_REGIONS_FILE):
+            raise RuntimeError(
+                "contentstack_utils: regions.json not found and could not be downloaded. "
+                "Run 'python scripts/refresh_regions.py' and ensure network access."
+            )
+
+        try:
+            with open(_REGIONS_FILE, "r", encoding="utf-8") as fh:
+                raw = fh.read()
+        except OSError as exc:
+            raise RuntimeError(
+                f"contentstack_utils: Could not read regions.json: {exc}"
+            ) from exc
+
+        try:
+            decoded = json.loads(raw)
+        except json.JSONDecodeError as exc:
+            raise RuntimeError(
+                "contentstack_utils: regions.json is corrupt. "
+                "Run 'python scripts/refresh_regions.py' to re-download it."
+            ) from exc
+
+        if not isinstance(decoded, dict) or "regions" not in decoded:
+            raise RuntimeError(
+                "contentstack_utils: regions.json is corrupt. "
+                "Run 'python scripts/refresh_regions.py' to re-download it."
+            )
+
+        Endpoint._regions_data = decoded
+        return Endpoint._regions_data
+
+    @staticmethod
+    def _download_and_save(dest: str) -> None:
+        """
+        Fetch regions.json from the Contentstack CDN and write it to *dest*.
+
+        Silent on failure — the caller decides whether a missing file is fatal.
+        """
+        os.makedirs(os.path.dirname(dest), exist_ok=True)
+        try:
+            with urllib.request.urlopen(_REGIONS_URL, timeout=30) as resp:
+                data = resp.read().decode("utf-8")
+        except Exception:
+            return
+
+        try:
+            decoded = json.loads(data)
+        except json.JSONDecodeError:
+            return
+
+        if isinstance(decoded, dict) and "regions" in decoded:
+            with open(dest, "w", encoding="utf-8") as fh:
+                fh.write(data)
+
+    @staticmethod
+    def _find_region_by_id_or_alias(
+        regions: list, normalized_input: str
+    ) -> Optional[Dict]:
+        """
+        Find a region by its ``id`` field first, then by any alias.
+
+        Both passes are case-insensitive (caller must pass a lowercased string).
+        Two-pass approach mirrors the PHP implementation: ID match wins over alias
+        match, which avoids surprising behaviour when a future alias happens to
+        collide with another region's canonical ID.
+        """
+        # Pass 1 — exact id match
+        for row in regions:
+            if row["id"] == normalized_input:
+                return row
+
+        # Pass 2 — alias match
+        for row in regions:
+            for alias in row.get("alias", []):
+                if alias.lower() == normalized_input:
+                    return row
+
+        return None
+
+    @staticmethod
+    def _strip_https(url: str) -> str:
+        """Strip ``https://`` or ``http://`` from the start of a URL."""
+        return re.sub(r"^https?://", "", url)
+
+    @staticmethod
+    def _strip_https_from_map(endpoints: Dict[str, str]) -> Dict[str, str]:
+        """Return a new dict with the scheme stripped from every URL value."""
+        return {key: Endpoint._strip_https(url) for key, url in endpoints.items()}

contentstack_utils/utils.py

@@ -6,6 +6,7 @@
 from lxml import etree
 
 from contentstack_utils.automate import Automate
+from contentstack_utils.endpoint import Endpoint
 from contentstack_utils.entry_editable import addEditableTags as _addEditableTags
 from contentstack_utils.entry_editable import addTags as _addTags
 from contentstack_utils.entry_editable import getTag as _getTag
@@ -219,6 +220,34 @@ def __get_metadata(elements):
             metadata = Metadata(element.text, typeof, uid, content_type, style, outer_html, attributes)
             return metadata
 
+    # ------------------------------------------------------------------
+    # Endpoint resolution — thin proxy to Endpoint class
+    # ------------------------------------------------------------------
+
+    @staticmethod
+    def get_contentstack_endpoint(
+        region: str = "us",
+        service: str = "",
+        omit_https: bool = False,
+    ):
+        """
+        Resolve a Contentstack service URL for the given region.
+
+        Delegates entirely to :class:`~contentstack_utils.endpoint.Endpoint`.
+        Both ``Utils.get_contentstack_endpoint(...)`` and
+        ``Endpoint.get_contentstack_endpoint(...)`` produce identical results —
+        choose whichever import path suits your codebase.
+
+        :param region: Region ID or alias (e.g. ``'na'``, ``'eu'``, ``'azure-na'``).
+        :param service: Optional service key (e.g. ``'contentDelivery'``).
+        :param omit_https: Strip ``https://`` from the returned URL(s).
+        :returns: URL string when *service* is given, dict of all URLs otherwise.
+        """
+        return Endpoint.get_contentstack_endpoint(region, service, omit_https)
+
+    # camelCase alias for cross-SDK parity
+    getContentstackEndpoint = get_contentstack_endpoint
+
     ####################################################
     #                   SUPERCHARGED                   #
     ####################################################

scripts/refresh_regions.py

@@ -0,0 +1,63 @@
+#!/usr/bin/env python3
+"""
+Pull the latest regions.json from the Contentstack CDN and overwrite the
+bundled copy at contentstack_utils/assets/regions.json.
+
+Usage:
+    python3 scripts/refresh_regions.py
+
+Run this whenever Contentstack adds a new region or service, then commit the
+updated file so all consumers get the change on their next install.
+"""
+
+import json
+import os
+import sys
+import urllib.request
+
+REGIONS_URL = "https://artifacts.contentstack.com/regions.json"
+DEST = os.path.join(
+    os.path.dirname(__file__),
+    "..",
+    "contentstack_utils",
+    "assets",
+    "regions.json",
+)
+
+
+def main() -> int:
+    dest = os.path.normpath(DEST)
+    print(f"Fetching {REGIONS_URL} ...")
+
+    try:
+        with urllib.request.urlopen(REGIONS_URL, timeout=30) as resp:
+            data = resp.read().decode("utf-8")
+    except Exception as exc:
+        print(f"ERROR: Could not download regions.json: {exc}", file=sys.stderr)
+        return 1
+
+    try:
+        decoded = json.loads(data)
+    except json.JSONDecodeError as exc:
+        print(f"ERROR: Downloaded content is not valid JSON: {exc}", file=sys.stderr)
+        return 1
+
+    if not isinstance(decoded, dict) or "regions" not in decoded:
+        print("ERROR: Downloaded JSON does not contain a 'regions' key.", file=sys.stderr)
+        return 1
+
+    region_count = len(decoded["regions"])
+    os.makedirs(os.path.dirname(dest), exist_ok=True)
+    with open(dest, "w", encoding="utf-8") as fh:
+        json.dump(decoded, fh, indent=2, ensure_ascii=False)
+        fh.write("\n")
+
+    print(f"OK: Wrote {region_count} regions to {dest}")
+    print("Next steps:")
+    print("  git add contentstack_utils/assets/regions.json")
+    print('  git commit -m "chore: refresh regions.json"')
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())