Skip to content
Merged
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -118,3 +118,4 @@ venv.bak/
.mypy_cache/
.idea/
.vscode/
*/assets/regions.json
12 changes: 12 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
@@ -1,5 +1,17 @@
# CHANGELOG

## _v2.6.0_

### **Date: 15-June-2026**

- Dynamic endpoint resolution via new `Endpoint` class.
- Region-to-URL mapping is now loaded from a bundled `regions.json` (sourced from `artifacts.contentstack.com`) instead of hardcoded `if/elif` chains.
- Added `Endpoint.get_contentstack_endpoint(region, service, omit_https)` — resolves any supported region to its `contentDelivery`, `contentManagement`, or other service URL.
- Added `contentstack.get_contentstack_endpoint()` module-level proxy.
- `Stack` now auto-resolves `host` and `live_preview` management host via `Endpoint` on initialization.
- Added `scripts/download_regions.py` to pre-populate `regions.json` at install time.
- Runtime fallback: if `regions.json` is absent, the SDK downloads it live on first use.

## _v2.5.1_

### **Date: 15-April-2026**
Expand Down
20 changes: 19 additions & 1 deletion contentstack/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
from .entry import Entry
from .asset import Asset
from .contenttype import ContentType
from .endpoint import Endpoint
from .https_connection import HTTPSConnection
from contentstack.stack import Stack
from .utility import Utils
Expand All @@ -14,15 +15,32 @@
"Entry",
"Asset",
"ContentType",
"Endpoint",
"HTTPSConnection",
"Stack",
"Utils"
)


def get_contentstack_endpoint(region='us', service='', omit_https=False):
"""
Resolve a Contentstack service endpoint URL for a given region.

Proxy to :class:`Endpoint.get_contentstack_endpoint` for convenience —
mirrors ``Contentstack::getContentstackEndpoint()`` in the PHP SDK.

:param region: Region ID or alias ('us', 'eu', 'azure-na', 'gcp-eu', ...).
:param service: Service key ('contentDelivery', 'contentManagement', ...).
When empty, returns a dict of all endpoints for the region.
:param omit_https: When True, strips 'https://' from the returned URL(s).
:returns: str when service is provided, dict[str,str] otherwise.
"""
return Endpoint.get_contentstack_endpoint(region, service, omit_https)

__title__ = 'contentstack-delivery-python'
__author__ = 'contentstack'
__status__ = 'debug'
__version__ = 'v2.5.1'
__version__ = 'v2.6.0'
__endpoint__ = 'cdn.contentstack.io'
__email__ = 'support@contentstack.com'
__developer_email__ = 'mobile@contentstack.com'
Expand Down
1 change: 1 addition & 0 deletions contentstack/contenttype.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@

# ************* Module ContentType **************
# Your code has been rated at 10.00/10 by pylint
from __future__ import annotations
import json
import logging
from urllib import parse
Expand Down
180 changes: 180 additions & 0 deletions contentstack/endpoint.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,180 @@
"""
Endpoint — Contentstack region-to-URL resolver.

Resolves Contentstack service endpoint URLs for any supported region.
Region data is loaded from contentstack/assets/regions.json (bundled) and
cached in-memory for the lifetime of the process. When the bundled file is
absent the class attempts a live download from the Contentstack CDN so the
SDK continues to work even when the file was not created during installation.
"""

import json
import os
import re

REGIONS_URL = 'https://artifacts.contentstack.com/regions.json'


class Endpoint:
"""
Resolves Contentstack service endpoint URLs for any supported region.

Usage::

from contentstack.endpoint import Endpoint

# Single service URL
url = Endpoint.get_contentstack_endpoint('eu', 'contentDelivery')
# 'https://eu-cdn.contentstack.com'

# All services for a region
endpoints = Endpoint.get_contentstack_endpoint('azure-na')
# {'contentDelivery': 'https://...', 'contentManagement': 'https://...', ...}

# Strip scheme (useful when setting host directly)
host = Endpoint.get_contentstack_endpoint('gcp-eu', 'contentDelivery', omit_https=True)
# 'gcp-eu-cdn.contentstack.com'
"""

_regions_data = None # in-memory cache — shared across all instances

@staticmethod
def get_contentstack_endpoint(region='us', service='', omit_https=False):
"""
Resolve a Contentstack service endpoint URL for a given region.

:param region: Region ID or alias ('us', 'eu', 'azure-na', 'gcp-eu', etc.).
Defaults to 'us' (AWS North America).
:param service: Service key ('contentDelivery', 'contentManagement', ...).
When empty, returns a dict of all endpoints for the region.
:param omit_https: When True, strips 'https://' prefix from returned URL(s).
:returns: str when service is provided, dict[str,str] otherwise.
:raises ValueError: When region is empty, unknown, or service is not found.
:raises RuntimeError: When regions.json cannot be read or parsed.
"""
if not region:
raise ValueError('Empty region provided. Please put valid region.')

data = Endpoint._load_regions()
normalized = region.strip().lower()
region_row = Endpoint._find_region(data['regions'], normalized)

if region_row is None:
raise ValueError(f'Invalid region: {region}')

if service:
if service not in region_row['endpoints']:
raise ValueError(
f'Service "{service}" not found for region "{region_row["id"]}"'
)
url = region_row['endpoints'][service]
return Endpoint._strip_https(url) if omit_https else url

endpoints = region_row['endpoints']
if omit_https:
return {k: Endpoint._strip_https(v) for k, v in endpoints.items()}
return dict(endpoints)

@staticmethod
def _load_regions():
"""
Load and cache regions.json.

Resolution order:
1. In-memory static cache (zero I/O after first call)
2. contentstack/assets/regions.json on disk (written by download script)
3. Live download from artifacts.contentstack.com (fallback)
"""
if Endpoint._regions_data is not None:
return Endpoint._regions_data

assets_dir = os.path.join(os.path.dirname(__file__), 'assets')
path = os.path.join(assets_dir, 'regions.json')

if not os.path.exists(path):
Endpoint._download_and_save(path)

if not os.path.exists(path):
raise RuntimeError(
'contentstack: regions.json not found and could not be downloaded. '
'Run "python scripts/download_regions.py" and ensure network access.'
)

try:
with open(path, 'r', encoding='utf-8') as f:
decoded = json.load(f)
except (OSError, json.JSONDecodeError) as exc:
raise RuntimeError(
f'contentstack: Could not read or parse regions.json: {exc}. '
'Run "python scripts/download_regions.py" to re-download it.'
) from exc

if not isinstance(decoded, dict) or 'regions' not in decoded:
raise RuntimeError(
'contentstack: regions.json is corrupt. '
'Run "python scripts/download_regions.py" to re-download it.'
)

Endpoint._regions_data = decoded
return Endpoint._regions_data

@staticmethod
def _download_and_save(dest):
"""
Download regions.json from the Contentstack CDN and save to disk.
Uses the requests library (already an SDK dependency).
Silent on failure — the caller decides whether a missing file is fatal.

:param dest: Absolute path to write the file to.
"""
os.makedirs(os.path.dirname(dest), exist_ok=True)

try:
import requests
response = requests.get(REGIONS_URL, timeout=30)
response.raise_for_status()
data = response.text
except Exception: # noqa: BLE001
return

try:
decoded = json.loads(data)
except json.JSONDecodeError:
return

if isinstance(decoded, dict) and 'regions' in decoded:
try:
with open(dest, 'w', encoding='utf-8') as f:
f.write(data)
except OSError:
pass

@staticmethod
def _find_region(regions, input_str):
"""
Find a region entry by its id or any alias (case-insensitive).

Two-pass: exact id match first, then alias[] scan — mirrors PHP implementation.

:param regions: list of region dicts from regions.json
:param input_str: already-lowercased input
:returns: region dict or None
"""
for row in regions:
if row['id'] == input_str:
return row
for row in regions:
for alias in row.get('alias', []):
if alias.lower() == input_str:
return row
return None

@staticmethod
def _strip_https(url):
"""Strip the https:// (or http://) scheme from a URL string."""
return re.sub(r'^https?://', '', url)

@staticmethod
def reset_cache():
"""Reset the internal region cache. Intended for testing only."""
Endpoint._regions_data = None
1 change: 1 addition & 0 deletions contentstack/entry.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
API Reference: https://www.contentstack.com/docs/developers/apis/content-delivery-api/#single-entry
"""
#min-similarity-lines=10
from __future__ import annotations
import logging
from urllib import parse
from contentstack.error_messages import ErrorMessages
Expand Down
34 changes: 18 additions & 16 deletions contentstack/stack.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
from contentstack.asset import Asset
from contentstack.assetquery import AssetQuery
from contentstack.contenttype import ContentType
from contentstack.endpoint import Endpoint
from contentstack.taxonomy import Taxonomy
from contentstack.globalfields import GlobalField
from contentstack.https_connection import HTTPSConnection
Expand Down Expand Up @@ -119,20 +120,15 @@ def _validate_stack(self):
if self.environment is None or self.environment == "":
raise PermissionError(ErrorMessages.INVALID_ENVIRONMENT_TOKEN)

if self.region.value == 'eu' and self.host == DEFAULT_HOST:
self.host = 'eu-cdn.contentstack.com'
elif self.region.value == 'au' and self.host == DEFAULT_HOST:
self.host = 'au-cdn.contentstack.com'
elif self.region.value == 'azure-na' and self.host == DEFAULT_HOST:
self.host = 'azure-na-cdn.contentstack.com'
elif self.region.value == 'azure-eu' and self.host == DEFAULT_HOST:
self.host = 'azure-eu-cdn.contentstack.com'
elif self.region.value == 'gcp-na' and self.host == DEFAULT_HOST:
self.host = 'gcp-na-cdn.contentstack.com'
elif self.region.value == 'gcp-eu' and self.host == DEFAULT_HOST:
self.host = 'gcp-eu-cdn.contentstack.com'
elif self.region.value != 'us':
self.host = f'{self.region.value}-{DEFAULT_HOST}'
if self.host == DEFAULT_HOST:
try:
self.host = Endpoint.get_contentstack_endpoint(
self.region.value, 'contentDelivery', omit_https=True)
except (ValueError, RuntimeError):
# Unknown/custom region — fall back to legacy pattern so
# code written before this feature was added continues to work.
if self.region.value != 'us':
self.host = f'{self.region.value}-{DEFAULT_HOST}'
self.endpoint = f'https://{self.host}/{self.version}'

def _setup_headers(self):
Expand Down Expand Up @@ -365,8 +361,14 @@ def image_transform(self, image_url, **kwargs):

def _setup_live_preview(self):
if self.live_preview and self.live_preview.get("enable"):
region_prefix = "" if self.region.value == "us" else f"{self.region.value}-"
self.live_preview["host"] = f"{region_prefix}rest-preview.contentstack.com"
if not self.live_preview.get("host"):
try:
mgmt_host = Endpoint.get_contentstack_endpoint(
self.region.value, 'contentManagement', omit_https=True)
except (ValueError, RuntimeError):
region_prefix = "" if self.region.value == "us" else f"{self.region.value}-"
mgmt_host = f"{region_prefix}api.contentstack.io"
self.live_preview["host"] = mgmt_host

if self.live_preview.get("preview_token"):
self.headers["preview_token"] = self.live_preview["preview_token"]
Expand Down
78 changes: 78 additions & 0 deletions scripts/download_regions.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,78 @@
"""
Downloads the Contentstack regions registry from the official source and
saves it to contentstack/assets/regions.json.

Run manually:
python scripts/download_regions.py

Can also be wired into setup.py post-install hooks or tox envsetup if needed.
"""

import json
import os
import sys

REGIONS_URL = 'https://artifacts.contentstack.com/regions.json'

DEST = os.path.join(
os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
'contentstack', 'assets', 'regions.json'
)


def download():
dest_dir = os.path.dirname(DEST)
os.makedirs(dest_dir, exist_ok=True)

try:
import requests
Comment thread
OMpawar-21 marked this conversation as resolved.
Outdated
except ImportError:
sys.stderr.write(
'contentstack: requests library not available. '
'Install dependencies first: pip install -r requirements.txt\n'
)
sys.exit(1)

print(f'contentstack: Downloading regions.json from {REGIONS_URL} ...')

try:
response = requests.get(REGIONS_URL, timeout=30)
response.raise_for_status()
data = response.text
except Exception as exc:
sys.stderr.write(
f'contentstack: Warning — could not download regions.json: {exc}. '
'The SDK will attempt to download it at runtime on first use.\n'
)
sys.exit(0) # non-fatal

try:
decoded = json.loads(data)
except json.JSONDecodeError:
sys.stderr.write(
'contentstack: Warning — downloaded data is not valid JSON.\n'
)
sys.exit(0)

if not isinstance(decoded, dict) or 'regions' not in decoded or \
not isinstance(decoded['regions'], list):
sys.stderr.write(
'contentstack: Warning — downloaded data is not a valid regions.json.\n'
)
sys.exit(0)

try:
with open(DEST, 'w', encoding='utf-8') as f:
f.write(data)
except OSError as exc:
sys.stderr.write(
f'contentstack: Warning — could not write regions.json to {DEST}: {exc}\n'
)
sys.exit(0)

region_count = len(decoded['regions'])
print(f'contentstack: regions.json downloaded ({region_count} regions) → {DEST}')


if __name__ == '__main__':
download()
Loading
Loading