Add publish api

cloudinary/api.py

@@ -228,6 +228,27 @@ def create_transformation(name, definition, **options):
     return call_api("post", uri, {"transformation": transformation_string(definition)}, **options)
 
 
+def publish_by_ids(public_ids, **options):
+    resource_type = options.pop("resource_type", "image")
+    uri = ["resources", resource_type, "publish_resources"]
+    params = dict(only(options, "type", "overwrite", "invalidate"), public_ids=public_ids)
+    return call_api("post", uri, params, **options)
+
+
+def publish_by_prefix(prefix, **options):
+    resource_type = options.pop("resource_type", "image")
+    uri = ["resources", resource_type, "publish_resources"]
+    params = dict(only(options, "type", "overwrite", "invalidate"), prefix=prefix)
+    return call_api("post", uri, params, **options)
+
+
+def publish_by_tag(tag, **options):
+    resource_type = options.pop("resource_type", "image")
+    uri = ["resources", resource_type, "publish_resources"]
+    params = dict(only(options, "type", "overwrite", "invalidate"), tag=tag)
+    return call_api("post", uri, params, **options)
+
+
 def upload_presets(**options):
     uri = ["upload_presets"]
     return call_api("get", uri, only(options, "next_cursor", "max_results"), **options)

tests/api_test.py

@@ -592,5 +592,31 @@ def test_upload_mapping(self):
         result = api.upload_mappings()
         self.assertNotIn(MAPPING_TEST_ID, [mapping.get("folder") for mapping in result["mappings"]])
 
+    @patch('urllib3.request.RequestMethods.request')
+    @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret")
+    def test_publish_by_ids(self, mocker):
+        mocker.return_value = MOCK_RESPONSE
+        api.publish_by_ids(["pub1", "pub2"])
+        self.assertTrue(get_uri(mocker.call_args[0]).endswith('/resources/image/publish_resources'))
+        self.assertIn('pub1', get_list_param(mocker, 'public_ids'))
+        self.assertIn('pub2', get_list_param(mocker, 'public_ids'))
+
+    @patch('urllib3.request.RequestMethods.request')
+    @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret")
+    def test_publish_by_prefix(self, mocker):
+        mocker.return_value = MOCK_RESPONSE
+        api.publish_by_prefix("pub_prefix")
+        self.assertTrue(get_uri(mocker.call_args[0]).endswith('/resources/image/publish_resources'))
+        self.assertEqual(get_param(mocker, 'prefix'), 'pub_prefix')
+
+    @patch('urllib3.request.RequestMethods.request')
+    @unittest.skipUnless(cloudinary.config().api_secret, "requires api_key/api_secret")
+    def test_publish_by_tag(self, mocker):
+        mocker.return_value = MOCK_RESPONSE
+        api.publish_by_tag("pub_tag")
+        self.assertTrue(get_uri(mocker.call_args[0]).endswith('/resources/image/publish_resources'))
+        self.assertEqual(get_param(mocker, 'tag'), "pub_tag")
+
+
 if __name__ == '__main__':
     unittest.main()