allow temporary credential overrides with api calls for multi-cloud codebases

Author: ojhilt

README.md

@@ -75,6 +75,27 @@ cloudinary.utils.cloudinary_url("sample.jpg", width=100, height=150, crop="fill"
 cloudinary.uploader.upload("my_picture.jpg")
 ```
 
+### Per-call account configuration
+For multi-account or multi-tenant applications, you can override configuration for a single API call by passing
+`cloudinary_config`. This does not mutate the global SDK configuration.
+
+```python
+cloudinary.api.ping(cloudinary_config={
+    "cloud_name": "tenant_cloud",
+    "api_key": "tenant_key",
+    "api_secret": "tenant_secret",
+})
+
+cloudinary.uploader.upload(
+    "my_picture.jpg",
+    cloudinary_config={
+        "cloud_name": "tenant_cloud",
+        "api_key": "tenant_key",
+        "api_secret": "tenant_secret",
+    },
+)
+```
+
 ### Django
 - [See full documentation](https://cloudinary.com/documentation/django_image_and_video_upload#django_forms_and_models).
 

cloudinary/api_client/call_account_api.py

@@ -1,7 +1,7 @@
 import cloudinary
 from cloudinary.api_client.execute_request import execute_request
 from cloudinary.provisioning.account_config import account_config
-from cloudinary.utils import get_http_connector, normalize_params
+from cloudinary.utils import get_http_connector, normalize_params, consume_cloudinary_config
 
 PROVISIONING_SUB_PATH = "provisioning"
 ACCOUNT_SUB_PATH = "accounts"
@@ -10,13 +10,15 @@
 
 # Account-scoped, authenticated call: provisioning/accounts/{account_id}/...
 def _call_account_api(method, uri, params=None, headers=None, **options):
+    options = consume_cloudinary_config(options)
     account_uri = [ACCOUNT_SUB_PATH, _account_id(options)] + uri
     return _execute_account_request(method, account_uri, _account_auth(options),
                                     params=params, headers=headers, **options)
 
 
 # Public, unauthenticated call: provisioning/... with no account_id or credentials
 def _call_public_account_api(method, uri, params=None, headers=None, **options):
+    options = consume_cloudinary_config(options)
     return _execute_account_request(method, uri, {"anonymous": True},
                                     params=params, headers=headers, **options)
 
@@ -41,6 +43,7 @@ def _account_auth(options):
 # Core transport: builds the provisioning URL and dispatches with the resolved auth.
 # The API version can be overridden via the "api_version" option (defaults to cloudinary.API_VERSION).
 def _execute_account_request(method, uri, auth, params=None, headers=None, **options):
+    options = consume_cloudinary_config(options)
     prefix = options.pop("upload_prefix",
                          cloudinary.config().upload_prefix) or "https://api.cloudinary.com"
     api_version = options.pop("api_version", cloudinary.API_VERSION)

cloudinary/api_client/call_api.py

@@ -2,7 +2,7 @@
 
 import cloudinary
 from cloudinary.api_client.execute_request import execute_request
-from cloudinary.utils import get_http_connector, normalize_params
+from cloudinary.utils import get_http_connector, normalize_params, consume_cloudinary_config
 
 logger = cloudinary.logger
 _http = get_http_connector(cloudinary.config(), cloudinary.CERT_KWARGS)
@@ -52,6 +52,7 @@ def call_api(method, uri, params, **options):
 
 
 def _call_api(method, uri, params=None, body=None, headers=None, extra_headers=None, **options):
+    options = consume_cloudinary_config(options)
     prefix = options.pop("upload_prefix",
                          cloudinary.config().upload_prefix) or "https://api.cloudinary.com"
     cloud_name = options.pop("cloud_name", cloudinary.config().cloud_name)

cloudinary/uploader.py

@@ -194,6 +194,7 @@ def upload(file, **options):
         - etc.
     :rtype: dict
     """
+    options = utils.consume_cloudinary_config(options)
     params = utils.build_upload_params(**options)
     return call_cacheable_api("upload", params, file=file, **options)
 
@@ -275,12 +276,13 @@ def _upload_large_part_with_auth_retry(file, http_headers, options):
     :rtype: dict
     """
     # Pin the token so the value handed to the callback is the one actually sent.
-    token = cloudinary.config().oauth_token
+    options = utils.consume_cloudinary_config(options)
+    token = options.get("oauth_token", cloudinary.config().oauth_token)
     pinned = dict(options, oauth_token=token) if token else options
     try:
         return upload_large_part(file, http_headers=http_headers, **pinned)
     except AuthorizationRequired:
-        callback = cloudinary.config().oauth_token_refresh_callback
+        callback = options.get("oauth_token_refresh_callback", cloudinary.config().oauth_token_refresh_callback)
         if not callback:
             raise
         callback(token)
@@ -303,6 +305,8 @@ def upload_large(file, **options):
     :return: The result of the upload API call.
     :rtype: dict
     """
+    options = utils.consume_cloudinary_config(options)
+
     if utils.is_remote_url(file):
         return upload(file, **options)
 
@@ -357,6 +361,7 @@ def upload_large_part(file, **options):
     :return: The result of the chunk upload API call.
     :rtype: dict
     """
+    options = utils.consume_cloudinary_config(options)
     params = utils.build_upload_params(**options)
 
     if 'resource_type' not in options:
@@ -859,6 +864,7 @@ def call_cacheable_api(action, params, http_headers=None, return_error=False, un
     :return: The parsed JSON response from Cloudinary.
     :rtype: dict
     """
+    options = utils.consume_cloudinary_config(options)
     result = call_api(action, params, http_headers, return_error, unsigned, file, timeout, **options)
     if "use_cache" in options or cloudinary.config().use_cache:
         _save_responsive_breakpoints_to_cache(result)
@@ -892,6 +898,7 @@ def call_api(action, params, http_headers=None, return_error=False, unsigned=Fal
 
     :raises Error: If an HTTP error or a Cloudinary error occurs.
     """
+    options = utils.consume_cloudinary_config(options)
     params = utils.cleanup_params(params)
 
     headers = {"User-Agent": cloudinary.get_user_agent()}

cloudinary/utils.py

@@ -20,6 +20,11 @@
 from six import iteritems
 from urllib3 import ProxyManager, PoolManager
 
+try:
+    from collections.abc import Mapping
+except ImportError:
+    from collections import Mapping
+
 import cloudinary
 from cloudinary import auth_token
 from cloudinary.api_client.tcp_keep_alive_manager import TCPKeepAlivePoolManager, TCPKeepAliveProxyManager
@@ -171,6 +176,8 @@
     SIGNATURE_SHA256: hashlib.sha256,
 }
 
+CLOUDINARY_CONFIG_OPTION = "cloudinary_config"
+
 
 def compute_hex_hash(s, algorithm=SIGNATURE_SHA1):
     """
@@ -196,6 +203,21 @@ def build_array(arg):
     return [arg]
 
 
+def consume_cloudinary_config(options):
+    config_overrides = options.pop(CLOUDINARY_CONFIG_OPTION, None)
+
+    if config_overrides is None:
+        return options
+
+    if not isinstance(config_overrides, Mapping):
+        raise ValueError("{} must be a dictionary".format(CLOUDINARY_CONFIG_OPTION))
+
+    for key, value in config_overrides.items():
+        options.setdefault(key, value)
+
+    return options
+
+
 def build_list_of_dicts(val):
     """
     Converts a value that can be presented as a list of dict.
@@ -614,6 +636,7 @@ def normalize_params(params):
 
 
 def sign_request(params, options):
+    options = consume_cloudinary_config(options)
     api_key = options.get("api_key", cloudinary.config().api_key)
     if not api_key:
         raise ValueError("Must supply api_key")
@@ -798,6 +821,7 @@ def unsigned_download_url_prefix(source, cloud_name, private_cdn, cdn_subdomain,
 
 
 def build_distribution_domain(options):
+    options = consume_cloudinary_config(options)
     source = options.pop('source', '')
     cloud_name = options.pop("cloud_name", cloudinary.config().cloud_name or None)
     if cloud_name is None:
@@ -906,6 +930,7 @@ def cloudinary_url(source, **options):
 
 
 def base_api_url(path, **options):
+    options = consume_cloudinary_config(options)
     cloudinary_prefix = options.get("upload_prefix", cloudinary.config().upload_prefix) \
                         or "https://api.cloudinary.com"
     cloud_name = options.get("cloud_name", cloudinary.config().cloud_name)
@@ -1161,6 +1186,7 @@ def build_custom_headers(headers):
 
 
 def build_upload_params(**options):
+    options = consume_cloudinary_config(options)
     params = {param_name: options.get(param_name) for param_name in __SIMPLE_UPLOAD_PARAMS if param_name in options}
     params["upload_preset"] = params.pop("upload_preset", cloudinary.config().upload_preset)
 

test/test_api_authorization.py

@@ -5,7 +5,7 @@
 import cloudinary
 from cloudinary import api
 from cloudinary import uploader
-from test.helper_test import TEST_IMAGE, get_headers, get_params, URLLIB3_REQUEST, patch
+from test.helper_test import TEST_IMAGE, get_headers, get_params, get_uri, URLLIB3_REQUEST, patch
 from test.test_api import MOCK_RESPONSE
 from test.test_config import OAUTH_TOKEN, CLOUD_NAME, API_KEY, API_SECRET
 from test.test_uploader import API_TEST_PRESET
@@ -61,6 +61,29 @@ def test_missing_credentials_admin_api(self, mocker):
         with six.assertRaisesRegex(self, Exception, "Must supply api_key"):
             api.ping()
 
+    @patch(URLLIB3_REQUEST)
+    def test_temporary_config_admin_api(self, mocker):
+        self.config.oauth_token = None
+        self.config.api_key = None
+        self.config.api_secret = None
+        mocker.return_value = MOCK_RESPONSE
+
+        api.ping(cloudinary_config={
+            "cloud_name": "temporary_cloud",
+            "api_key": "temporary_key",
+            "api_secret": "temporary_secret",
+        })
+
+        headers = get_headers(mocker)
+
+        self.assertTrue("authorization" in headers)
+        self.assertEqual("Basic dGVtcG9yYXJ5X2tleTp0ZW1wb3Jhcnlfc2VjcmV0", headers["authorization"])
+        self.assertTrue(get_uri(mocker).endswith("/temporary_cloud/ping"))
+
+        self.assertIsNone(self.config.api_key)
+        self.assertIsNone(self.config.api_secret)
+        self.assertEqual(CLOUD_NAME, self.config.cloud_name)
+
     @patch(URLLIB3_REQUEST)
     def test_oauth_token_upload_api(self, mocker):
         self.config.oauth_token = OAUTH_TOKEN
@@ -119,6 +142,31 @@ def test_missing_credentials_upload_api(self, mocker):
         params = get_params(mocker)
         self.assertTrue("upload_preset" in params)
 
+    @patch(URLLIB3_REQUEST)
+    def test_temporary_config_upload_api(self, mocker):
+        self.config.oauth_token = None
+        self.config.api_key = None
+        self.config.api_secret = None
+        mocker.return_value = MOCK_RESPONSE
+
+        uploader.upload(TEST_IMAGE, cloudinary_config={
+            "cloud_name": "temporary_cloud",
+            "api_key": "temporary_key",
+            "api_secret": "temporary_secret",
+            "upload_preset": API_TEST_PRESET,
+        })
+
+        self.assertTrue(get_uri(mocker).endswith("/temporary_cloud/image/upload"))
+
+        params = get_params(mocker)
+        self.assertEqual("temporary_key", params["api_key"])
+        self.assertEqual(API_TEST_PRESET, params["upload_preset"])
+        self.assertIn("signature", params)
+
+        self.assertIsNone(self.config.api_key)
+        self.assertIsNone(self.config.api_secret)
+        self.assertEqual(CLOUD_NAME, self.config.cloud_name)
+
 
 if __name__ == '__main__':
     unittest.main()

test/test_provisioning_api.py

@@ -355,5 +355,34 @@ def test_create_agent_account_parses_response(self):
         self.assertIn("guidance", res)
 
 
+class AccountApiTemporaryConfigTest(unittest.TestCase):
+    def tearDown(self):
+        reset_config()
+
+    def test_temporary_config_account_api(self):
+        with patch(URLLIB3_REQUEST) as mocker:
+            mocker.return_value = api_response_mock()
+            cloudinary.provisioning.sub_accounts(cloudinary_config={
+                "account_id": "temporary_account",
+                "provisioning_api_key": "temporary_key",
+                "provisioning_api_secret": "temporary_secret",
+                "upload_prefix": "https://custom.example.com",
+            })
+
+        self.assertEqual("GET", get_method(mocker))
+        self.assertTrue(
+            get_uri(mocker).endswith("/v1_1/provisioning/accounts/temporary_account/sub_accounts")
+        )
+
+        headers = get_headers(mocker)
+        self.assertTrue("authorization" in headers)
+        self.assertEqual("Basic dGVtcG9yYXJ5X2tleTp0ZW1wb3Jhcnlfc2VjcmV0", headers["authorization"])
+
+        config = account_config()
+        self.assertIsNone(config.account_id)
+        self.assertIsNone(config.provisioning_api_key)
+        self.assertIsNone(config.provisioning_api_secret)
+
+
 if __name__ == '__main__':
     unittest.main()