pangea-sdk: add base_url_template

kenany · pangea-andrest · kenany · commit c1a05692e574 · 2025-04-18T16:37:10.000-04:00
This allows for greater control over the complete API URL.

Co-authored-by: "H. Andres Tournour" &lt;andres.tournour@pangea.cloud&gt;
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -13,6 +13,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - AI Guard: detector overrides.
 - AI Guard: topic detector.
 - AI Guard: `ignore_recipe` in detector overrides.
+- `base_url_template` has been added to `PangeaConfig` to allow for greater
+  control over the complete API URL. This option may be a full URL with the
+  optional `{SERVICE_NAME}` placeholder, which will be replaced by the slug of
+  the respective service name. This supersedes `environment` and `insecure`.
 
 ### Changed
 
diff --git a/examples/asyncio/sanitize/sanitize_examples/sanitize_and_share.py b/examples/asyncio/sanitize/sanitize_examples/sanitize_and_share.py
@@ -19,9 +19,9 @@ async def main() -> None:
     domain = os.getenv("PANGEA_DOMAIN")
     assert domain
 
-    config = PangeaConfig(domain)
-    # Create Sanitize client with its token and its config
+    config = PangeaConfig(domain=domain)
 
+    # Create Sanitize client with its token and its config
     client = SanitizeAsync(token, config)
     try:
         # Create Sanitize file information
diff --git a/examples/asyncio/sanitize/sanitize_examples/sanitize_multipart.py b/examples/asyncio/sanitize/sanitize_examples/sanitize_multipart.py
@@ -19,9 +19,9 @@ async def main() -> None:
     domain = os.getenv("PANGEA_DOMAIN")
     assert domain
 
-    config = PangeaConfig(domain)
-    # Create Sanitize client with its token and its config
+    config = PangeaConfig(domain=domain)
 
+    # Create Sanitize client with its token and its config
     client = SanitizeAsync(token, config)
     try:
         # Create Sanitize file information
diff --git a/examples/asyncio/sanitize/sanitize_examples/sanitize_no_share.py b/examples/asyncio/sanitize/sanitize_examples/sanitize_no_share.py
@@ -19,9 +19,9 @@ async def main() -> None:
     domain = os.getenv("PANGEA_DOMAIN")
     assert domain
 
-    config = PangeaConfig(domain)
-    # Create Sanitize client with its token and its config
+    config = PangeaConfig(domain=domain)
 
+    # Create Sanitize client with its token and its config
     client = SanitizeAsync(token, config)
     try:
         # Create Sanitize file information
diff --git a/examples/asyncio/sanitize/sanitize_examples/sanitize_split_upload_post_url.py b/examples/asyncio/sanitize/sanitize_examples/sanitize_split_upload_post_url.py
@@ -26,9 +26,9 @@ async def main() -> None:
     domain = os.getenv("PANGEA_DOMAIN")
     assert domain
 
-    config = PangeaConfig(domain)
-    # Create Sanitize client with its token and its config
+    config = PangeaConfig(domain=domain)
 
+    # Create Sanitize client with its token and its config
     client = SanitizeAsync(token, config)
     try:
         # Create Sanitize file information
diff --git a/examples/asyncio/sanitize/sanitize_examples/sanitize_split_upload_put_url.py b/examples/asyncio/sanitize/sanitize_examples/sanitize_split_upload_put_url.py
@@ -25,9 +25,9 @@ async def main() -> None:
     domain = os.getenv("PANGEA_DOMAIN")
     assert domain
 
-    config = PangeaConfig(domain)
-    # Create Sanitize client with its token and its config
+    config = PangeaConfig(domain=domain)
 
+    # Create Sanitize client with its token and its config
     client = SanitizeAsync(token, config)
     try:
         # Create Sanitize file information
diff --git a/examples/sanitize/sanitize_examples/sanitize_and_share.py b/examples/sanitize/sanitize_examples/sanitize_and_share.py
@@ -18,9 +18,9 @@ def main() -> None:
     domain = os.getenv("PANGEA_DOMAIN")
     assert domain
 
-    config = PangeaConfig(domain)
-    # Create Sanitize client with its token and its config
+    config = PangeaConfig(domain=domain)
 
+    # Create Sanitize client with its token and its config
     client = Sanitize(token, config)
     try:
         # Create Sanitize file information
diff --git a/examples/sanitize/sanitize_examples/sanitize_multipart.py b/examples/sanitize/sanitize_examples/sanitize_multipart.py
@@ -18,9 +18,9 @@ def main() -> None:
     domain = os.getenv("PANGEA_DOMAIN")
     assert domain
 
-    config = PangeaConfig(domain)
-    # Create Sanitize client with its token and its config
+    config = PangeaConfig(domain=domain)
 
+    # Create Sanitize client with its token and its config
     client = Sanitize(token, config)
     try:
         # Create Sanitize file information
diff --git a/examples/sanitize/sanitize_examples/sanitize_no_share.py b/examples/sanitize/sanitize_examples/sanitize_no_share.py
@@ -18,9 +18,9 @@ def main() -> None:
     domain = os.getenv("PANGEA_DOMAIN")
     assert domain
 
-    config = PangeaConfig(domain)
-    # Create Sanitize client with its token and its config
+    config = PangeaConfig(domain=domain)
 
+    # Create Sanitize client with its token and its config
     client = Sanitize(token, config)
     try:
         # Create Sanitize file information
diff --git a/examples/sanitize/sanitize_examples/sanitize_split_upload_post_url.py b/examples/sanitize/sanitize_examples/sanitize_split_upload_post_url.py
@@ -26,9 +26,9 @@ def main() -> None:
     domain = os.getenv("PANGEA_DOMAIN")
     assert domain
 
-    config = PangeaConfig(domain)
-    # Create Sanitize client with its token and its config
+    config = PangeaConfig(domain=domain)
 
+    # Create Sanitize client with its token and its config
     client = Sanitize(token, config)
     try:
         # Create Sanitize file information
diff --git a/examples/sanitize/sanitize_examples/sanitize_split_upload_put_url.py b/examples/sanitize/sanitize_examples/sanitize_split_upload_put_url.py
@@ -25,9 +25,9 @@ def main() -> None:
     domain = os.getenv("PANGEA_DOMAIN")
     assert domain
 
-    config = PangeaConfig(domain)
-    # Create Sanitize client with its token and its config
+    config = PangeaConfig(domain=domain)
 
+    # Create Sanitize client with its token and its config
     client = Sanitize(token, config)
     try:
         # Create Sanitize file information
diff --git a/examples/vault/vault_examples/encrypt_structured.py b/examples/vault/vault_examples/encrypt_structured.py
@@ -17,7 +17,7 @@ def main() -> None:
     domain = os.getenv("PANGEA_DOMAIN")
     assert domain
 
-    config = PangeaConfig(domain)
+    config = PangeaConfig(domain=domain)
     vault = Vault(token, config)
 
     # First create an encryption key, either from the Pangea Console or
diff --git a/packages/pangea-sdk/pangea/config.py b/packages/pangea-sdk/pangea/config.py
@@ -1,32 +1,30 @@
 # Copyright 2022 Pangea Cyber Corporation
 # Author: Pangea Cyber Corporation
+from __future__ import annotations
 
-from dataclasses import dataclass
-from typing import Literal, Optional
+from typing import Any, Optional
 
+from pydantic import BaseModel, model_validator
 
-@dataclass
-class PangeaConfig:
-    """Holds run time configuration information used by SDK components."""
 
-    domain: str = "aws.us.pangea.cloud"
-    """
-    Used to set Pangea domain (and port if needed), it should not include service subdomain
-    just for particular use cases when environment = "local", domain could be set to an url including:
-    scheme (http:// or https://), subdomain, domain and port.
-    """
+class PangeaConfig(BaseModel):
+    """Holds run time configuration information used by SDK components."""
 
-    environment: Literal["production", "local"] = "production"
+    base_url_template: str = "https://{SERVICE_NAME}.aws.us.pangea.cloud"
     """
-    Pangea environment, used to construct service URLs.
-
-    If set to "local", then `domain` must be the full host (i.e., hostname and
-    port) for the Pangea service that this `PangeaConfig` will be used for.
+    Template for constructing the base URL for API requests. The placeholder
+    `{SERVICE_NAME}` will be replaced with the service name slug. This is a
+    more powerful version of `domain` that allows for setting more than just
+    the host of the API server. Defaults to
+    `https://{SERVICE_NAME}.aws.us.pangea.cloud`.
     """
 
-    insecure: bool = False
+    domain: str = "aws.us.pangea.cloud"
     """
-    Set to true to use plain http
+    Base domain for API requests. This is a weaker version of `base_url_template`
+    that only allows for setting the host of the API server. Use
+    `base_url_template` for more control over the URL, such as setting
+    service-specific paths. Defaults to `aws.us.pangea.cloud`.
     """
 
     request_retries: int = 3
@@ -58,3 +56,10 @@ class PangeaConfig:
     """
     Extra user agent to be added to request user agent
     """
+
+    @model_validator(mode="before")
+    @classmethod
+    def _domain_backwards_compat(cls, data: Any) -> Any:
+        if isinstance(data, dict) and "base_url_template" not in data and "domain" in data:
+            return {**data, "base_url_template": f"https://{{SERVICE_NAME}}.{data['domain']}"}
+        return data
diff --git a/packages/pangea-sdk/pangea/request.py b/packages/pangea-sdk/pangea/request.py
@@ -14,6 +14,7 @@
 from requests.adapters import HTTPAdapter, Retry
 from requests_toolbelt import MultipartDecoder  # type: ignore[import-untyped]
 from typing_extensions import TypeVar
+from yarl import URL
 
 import pangea
 import pangea.exceptions as pe
@@ -107,24 +108,16 @@ def _get_poll_path(self, request_id: str):
         return f"request/{request_id}"
 
     def _url(self, path: str) -> str:
-        if self.config.domain.startswith("http://") or self.config.domain.startswith("https://"):
-            # it's URL
-            url = f"{self.config.domain}/{path}"
-        else:
-            schema = "http://" if self.config.insecure else "https://"
-            domain = (
-                self.config.domain if self.config.environment == "local" else f"{self.service}.{self.config.domain}"
-            )
-            url = f"{schema}{domain}/{path}"
-        return url
+        url = URL(self.config.base_url_template.format(SERVICE_NAME=self.service))
+        return str(url / path)
 
     def _headers(self) -> dict:
         headers = {
             "User-Agent": self._user_agent,
             "Authorization": f"Bearer {self.token}",
         }
 
-        # We want to ignore previous headers if user tryed to set them, so we will overwrite them.
+        # We want to ignore previous headers if user tried to set them, so we will overwrite them.
         self._extra_headers.update(headers)
         return self._extra_headers
 
@@ -628,9 +621,7 @@ def _init_session(self) -> requests.Session:
         adapter = HTTPAdapter(max_retries=retry_config)
         session = requests.Session()
 
-        if self.config.insecure:
-            session.mount("http://", adapter)
-        else:
-            session.mount("https://", adapter)
+        session.mount("http://", adapter)
+        session.mount("https://", adapter)
 
         return session
diff --git a/packages/pangea-sdk/poetry.lock b/packages/pangea-sdk/poetry.lock
diff --git a/packages/pangea-sdk/pyproject.toml b/packages/pangea-sdk/pyproject.toml
diff --git a/packages/pangea-sdk/tests/unit/test_config.py b/packages/pangea-sdk/tests/unit/test_config.py
