Skip to content
Merged
Show file tree
Hide file tree
Changes from all 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
70 changes: 70 additions & 0 deletions scripts/download_regions.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,70 @@
"""
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
import requests

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)

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()
1 change: 1 addition & 0 deletions setup.py
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,7 @@ def get_version(package):
long_description_content_type="text/markdown",
url="https://github.com/contentstack/contentstack-python",
packages=['contentstack'],
package_data={'contentstack': ['assets/regions.json']},
license='MIT',
test_suite='tests',
install_requires=requirements,
Expand Down
Loading
Loading