diff --git a/hapi/cos_blog.py b/hapi/cos_blog.py new file mode 100644 index 0000000..4cb5bd6 --- /dev/null +++ b/hapi/cos_blog.py @@ -0,0 +1,209 @@ +from base import BaseClient +import simplejson as json +from urllib import urlencode + +BLOG_API_VERSION = '2' + +class COSBlogClient(BaseClient): + + # Blogs + def _get_path(self, subpath): + path_overrides = { + 'blog-authors': 'blogs/v3/blog-authors', + 'topics': 'blogs/v3/topics' + } + + if subpath in path_overrides: + return path_overrides.get(subpath) + + return 'content/api/v%s/%s' % (BLOG_API_VERSION, subpath) + + def get_blogs(self, **options): + '''https://developers.hubspot.com/docs/methods/blogv2/get_blogs''' + return self._call('blogs', **options) + + def get_blog(self, blog_id, **options): + '''https://developers.hubspot.com/docs/methods/blogv2/get_blogs_blog_id''' + return self._call('blogs/%s' % blog_id, **options) + + def get_blog_versions(self, blog_id, **options): + '''https://developers.hubspot.com/docs/methods/blogv2/get_blogs_blog_id_versions''' + return self._call('blogs/%s/versions' % blog_id, **options) + + def get_blog_version(self, blog_id, version_id, **options): + '''https://developers.hubspot.com/docs/methods/blogv2/get_blogs_blog_id_versions_version_id''' + return self._call('blogs/%s/versions/%s' % (blog_id, version_id), **options) + + # Blog Posts + @staticmethod + def _args_to_dict(allowed_fields, **kwargs): + data = {} + for k in allowed_fields: + if kwargs.get(k) is not None: + data[k] = kwargs.get(k) + return data + + @classmethod + def _post_data(cls, **kwargs): + allowed_fields = ('blog_author_id', 'campaign', 'campaign_name', 'content_group_id', + 'footer_html', 'head_html', 'is_draft', 'meta_description', 'meta_keywords', + 'name', 'post_body', 'post_summary', 'publish_date', 'publish_immediately', + 'slug', 'topic_ids', 'widgets') + return cls._args_to_dict(allowed_fields, **kwargs) + + def create_post(self, content_group_id, name, blog_author_id=None, + campaign=None, campaign_name=None, footer_html=None, head_html=None, + is_draft=None, meta_description=None, meta_keywords=None, + post_body=None, post_summary=None, publish_date=None, + publish_immediately=None, slug=None, topic_ids=None, widgets=None, **options): + '''https://developers.hubspot.com/docs/methods/blogv2/post_blog_posts''' + data = self._post_data(**locals()) + return self._call('blog-posts', data=json.dumps(data), method='POST', + content_type='application/json',**options) + + def get_posts(self, query={}, **options): + '''https://developers.hubspot.com/docs/methods/blogv2/get_blog_posts''' + return self._call('blog-posts', query=urlencode(query), **options) + + def update_post(self, blog_post_id, content_group_id=None, name=None, blog_author_id=None, + campaign=None, campaign_name=None, footer_html=None, head_html=None, + is_draft=None, meta_description=None, meta_keywords=None, + post_body=None, post_summary=None, publish_date=None, + publish_immediately=None, slug=None, topic_ids=None, widgets=None, **options): + '''https://developers.hubspot.com/docs/methods/blogv2/put_blog_posts_blog_post_id''' + data = self._post_data(**locals()) + return self._call('blog-posts/%s' % blog_post_id, data=json.dumps(data), method='PUT', + content_type='application/json',**options) + + def delete_post(self, blog_post_id, **options): + '''https://developers.hubspot.com/docs/methods/blogv2/delete_blog_posts_blog_post_id''' + return self._call('blog-posts/%s' % blog_post_id, method='DELETE', **options) + + def get_post(self, blog_post_id, **options): + '''https://developers.hubspot.com/docs/methods/blogv2/get_blog_posts_blog_post_id''' + return self._call('blog-posts/%s' % blog_post_id, **options) + + def update_auto_save_buffer(self, blog_post_id, content_group_id=None, name=None, blog_author_id=None, + campaign=None, campaign_name=None, footer_html=None, head_html=None, + is_draft=None, meta_description=None, meta_keywords=None, + post_body=None, post_summary=None, publish_date=None, + publish_immediately=None, slug=None, topic_ids=None, widgets=None, **options): + '''https://developers.hubspot.com/docs/methods/blogv2/put_blog_posts_blog_post_id_buffer''' + data = self._post_data(**locals()) + return self._call('blog-posts/%s/buffer' % blog_post_id, data=json.dumps(data), method='PUT', + content_type='application/json',**options) + + def get_auto_save_buffer(self, blog_post_id, **options): + '''https://developers.hubspot.com/docs/methods/blogv2/get_blog_posts_blog_post_id_buffer''' + return self._call('blog-posts/%s/buffer' % blog_post_id, **options) + + def clone_post(self, blog_post_id, content_group_id=None, name=None, blog_author_id=None, + campaign=None, campaign_name=None, footer_html=None, head_html=None, + is_draft=None, meta_description=None, meta_keywords=None, + post_body=None, post_summary=None, publish_date=None, + publish_immediately=None, slug=None, topic_ids=None, widgets=None, **options): + '''https://developers.hubspot.com/docs/methods/blogv2/post_blog_posts_blog_post_id_clone''' + data = self._post_data(**locals()) + return self._call('blog-posts/%s/clone' % blog_post_id, data=json.dumps(data), method='POST', + content_type='application/json',**options) + + def get_buffered_changes(self, blog_post_id, **options): + '''https://developers.hubspot.com/docs/methods/blogv2/get_blog_posts_blog_post_id_has_buffered_changes''' + return self._call('blog-posts/%s/has-buffered-changes' % blog_post_id, **options) + + def publish_post(self, blog_post_id, action, **options): + '''https://developers.hubspot.com/docs/methods/blogv2/post_blog_posts_blog_post_id_publish_action''' + if action not in ('push-buffer-live', 'schedule-publish', 'cancel-publish'): + raise ValueError('%s is not a valid action' % action) + data = {'action': action} + return self._call('blog-posts/%s/publish-action' % blog_post_id, data=json.dumps(data), method='POST', **options) + + def push_buffer_live(self, blog_post_id, **options): + '''https://developers.hubspot.com/docs/methods/blogv2/post_blog_posts_blog_post_id_push_buffer_live''' + return self._call('blog-posts/%s/push-buffer-live' % blog_post_id, method='POST', **options) + + def undelete_post(self, blog_post_id, **options): + '''https://developers.hubspot.com/docs/methods/blogv2/post_blog_posts_blog_post_id_restore_deleted''' + return self._call('blog-posts/%s/restore-deleted' % blog_post_id, method='POST', **options) + + def validate_buffer(self, blog_post_id, **options): + '''https://developers.hubspot.com/docs/methods/blogv2/post_blog_posts_blog_post_id_validate_buffer''' + return self._call('blog-posts/%s/validate-buffer' % blog_post_id, method='POST', **options) + + def get_post_versions(self, blog_post_id, **options): + '''https://developers.hubspot.com/docs/methods/blogv2/get_blog_posts_blog_post_id_versions''' + return self._call('blog-posts/%s/versions' % blog_post_id, **options) + + def get_post_version(self, blog_post_id, version_id, **options): + '''https://developers.hubspot.com/docs/methods/blogv2/get_blog_posts_blog_post_id_versions_version_id''' + return self._call('blog-posts/%s/versions/%s' % (blog_post_id, version_id), **options) + + def restore_post_version(self, blog_post_id, version_id, **options): + '''https://developers.hubspot.com/docs/methods/blogv2/post_blog_posts_blog_post_id_versions_restore''' + data = {'version_id': version_id} + return self._call('blog-posts/%s/versions/restore' % blog_post_id, data=json.dumps(data), method='POST', **options) + + + # Blog Authors + def create_author(self, email, full_name, user_id=None, username=None, **options): + '''https://developers.hubspot.com/docs/methods/blogv2/post_blog_authors''' + data = { + 'email': email, + 'fullName': full_name, + 'userId': user_id, + 'username': username + } + return self._call('blog-authors', data=json.dumps(data), method='POST', content_type='application/json', **options) + + def get_authors(self, query={}, **options): + '''https://developers.hubspot.com/docs/methods/blogv2/get_blog_authors''' + return self._call('blog-authors', query=urlencode(query), **options) + + def update_author(self, author_id, email=None, full_name=None, user_id=None, username=None, **options): + '''https://developers.hubspot.com/docs/methods/blogv2/put_blog_authors_blog_author_id''' + data = self._args_to_dict(('email', 'fullName', 'userId', 'username'), **locals()) + return self._call('blog-authors/%s' % author_id, data=json.dumps(data), + method='PUT', content_type='application/json', **options) + + def delete_author(self, author_id, **options): + '''https://developers.hubspot.com/docs/methods/blogv2/delete_blog_authors_blog_author_id''' + return self._call('blog-authors/%s' % author_id, method='DELETE', **options) + + def get_author(self, author_id, **options): + '''https://developers.hubspot.com/docs/methods/blogv2/get_blog_authors_blog_author_id''' + return self._call('blog-authors/%s' % author_id, **options) + + def undelete_author(self, author_id, **options): + '''https://developers.hubspot.com/docs/methods/blogv2/post_blog_authors_blog_author_id_restore_deleted''' + return self._call('blog-authors/%s/restore-deleted' % author_id, method='POST', **options) + + # Topics + def create_topic(self, name, slug=None, **options): + '''https://developers.hubspot.com/docs/methods/blogv2/post_topics''' + data = {'name': name} + if slug: + data['slug'] = slug + return self._call('topics', data=json.dumps(data), method='POST', **options) + + def get_topics(self, query={}, **options): + '''https://developers.hubspot.com/docs/methods/blogv2/get_topics''' + return self._call('topics', query=urlencode(query), **options) + + def update_topic(self, topic_id, name=None, slug=None, **options): + '''https://developers.hubspot.com/docs/methods/blogv2/put_topics_topic_id''' + data = self._args_to_dict(('name', 'slug'), **locals()) + return self._call('topics/%s' % topic_id, data=json.dumps(data), + method='PUT', content_type='application/json', **options) + + def delete_topic(self, topic_id, **options): + '''https://developers.hubspot.com/docs/methods/blogv2/delete_topics_topic_id''' + return self._call('topics/%s' % topic_id, method='DELETE', **options) + + def get_topic(self, topic_id, **options): + '''https://developers.hubspot.com/docs/methods/blogv2/get_topics_topic_id''' + return self._call('topics/%s' % topic_id, **options) + + def undelete_topic(self, topic_id, **options): + '''https://developers.hubspot.com/docs/methods/blogv2/post_topics_topic_id_restore_deleted''' + return self._call('topics/%s/restore-deleted' % topic_id, method='POST', **options) + diff --git a/hapi/nurturing.py b/hapi/nurturing.py deleted file mode 100644 index 87889de..0000000 --- a/hapi/nurturing.py +++ /dev/null @@ -1,25 +0,0 @@ -from base import BaseClient - -NURTURING_API_VERSION = '1' - -class NurturingClient(BaseClient): - - def _get_path(self, subpath): - return 'nurture/v%s/%s' % (NURTURING_API_VERSION, subpath) - - def get_campaigns(self, **options): - return self._call('campaigns', **options) - - def get_leads(self, campaign_guid, **options): - return self._call('campaign/%s/list' % campaign_guid, **options) - - def get_history(self, lead_guid, **options): - return self._call('lead/%s' % lead_guid, **options) - - def enroll_lead(self, campaign_guid, lead_guid, **options): - return self._call('campaign/%s/add' % campaign_guid, data=lead_guid, method='POST', **options) - - def unenroll_lead(self, campaign_guid, lead_guid, **options): - return self._call('campaign/%s/remove' % campaign_guid, data=lead_guid, method='POST', **options) - - diff --git a/hapi/test/test_cos_blog.py b/hapi/test/test_cos_blog.py new file mode 100644 index 0000000..8bc1a54 --- /dev/null +++ b/hapi/test/test_cos_blog.py @@ -0,0 +1,233 @@ +from __future__ import with_statement +import unittest2 +import helper +import logger +from pprint import pprint +import simplejson as json +from nose.plugins.attrib import attr +from datetime import datetime + +from hapi.cos_blog import COSBlogClient + +class BlogClientTest(unittest2.TestCase): + """ + Unit tests for the HubSpot Blog API Python wrapper (hapipy) client. + + This file contains some unittest tests for the Blog API. + + Questions, comments, etc: http://docs.hubapi.com/wiki/Discussion_Group + """ + BLOG_ID = 351076997 + VERSION_ID = 12926383 + + def setUp(self): + self.client = COSBlogClient(**helper.get_options()) + + def tearDown(self): + pass + + # Blogs + @attr('api') + def test_get_blogs(self): + blogs = self.client.get_blogs() + self.assertTrue(len(blogs)) + pprint(blogs) + + @attr('api') + def test_get_blog(self): + blog = self.client.get_blog(self.BLOG_ID) + self.assertTrue(blog) + self.assertTrue(blog['id'] == self.BLOG_ID) + pprint(blog) + + @attr('api') + def test_get_blog_versions(self): + versions = self.client.get_blog_versions(self.BLOG_ID) + self.assertTrue(versions) + pprint(versions) + + @attr('api') + def test_get_blog_version(self): + version = self.client.get_blog_version(self.BLOG_ID, self.VERSION_ID) + self.assertTrue(version) + pprint(version) + + # Blog Posts + @attr('api') + def test_create_post(self): + post = self.client.create_post(content_group_id=self.BLOG_ID, name="Test Blog Post") + pprint(post) + self.assertTrue(post) + + @attr('api') + def test_get_posts(self): + posts = self.client.get_posts(query={'name': 'Demonstration Blog Post'}) + pprint(posts) + self.assertTrue(posts) + + @attr('api') + def test_update_post(self): + post = self.client.update_post(blog_post_id=348109414, name="Updated Test Blog Post") + pprint(post) + self.assertTrue(post) + + @attr('api') + def test_delete_post(self): + post = self.client.delete_post(blog_post_id=348109414) + pprint(post) + self.assertTrue(post) + post = self.client.undelete_post(blog_post_id=348109414) + pprint(post) + self.assertTrue(post) + + @attr('api') + def test_undelete_post(self): + post = self.client.undelete_post(blog_post_id=348109414) + pprint(post) + self.assertTrue(post) + + @attr('api') + def test_get_post(self): + post = self.client.get_post(blog_post_id=348109414) + pprint(post) + self.assertTrue(post) + + @attr('api') + def test_update_auto_save_buffer(self): + post = self.client.update_auto_save_buffer(blog_post_id=348109414, name="Updated Test Post Buffer") + pprint(post) + self.assertTrue(post) + + @attr('api') + def test_get_auto_save_buffer(self): + post = self.client.get_auto_save_buffer(blog_post_id=348109414) + pprint(post) + self.assertTrue(post) + + @attr('api') + def test_clone_post(self): + post = self.client.clone_post(blog_post_id=348109414, name="Cloned Blog Post") + pprint(post) + self.assertTrue(post) + + @attr('api') + def test_get_buffered_changes(self): + post = self.client.get_buffered_changes(blog_post_id=348109414) + pprint(post) + self.assertTrue(post) + + @attr('api') + def test_publish_post(self): + post = self.client.publish_post(blog_post_id=348109414, action='schedule-publish') + pprint(post) + self.assertTrue(post) + post = self.client.publish_post(blog_post_id=348109414, action='push-buffer-live') + pprint(post) + self.assertTrue(post) + post = self.client.publish_post(blog_post_id=348109414, action='cancel-publish') + pprint(post) + self.assertTrue(post) + with self.assertRaises(ValueError): + self.client.publish_post(blog_post_id=348109414, action='invalid-action') + + @attr('api') + def test_push_buffer_live(self): + post = self.client.push_buffer_live(blog_post_id=348109414) + pprint(post) + self.assertTrue(post) + + @attr('api') + def test_validate_buffer(self): + post = self.client.validate_buffer(blog_post_id=348109414) + pprint(post) + self.assertTrue(post) + + @attr('api') + def test_get_post_versions(self): + response = self.client.get_post_versions(blog_post_id=348109414) + pprint(response) + self.assertTrue(response) + + version_id = response[0]['version_id'] + response = self.client.get_post_version(blog_post_id=348109414, version_id=version_id) + pprint(response) + self.assertTrue(response) + + response = self.client.restore_post_version(blog_post_id=348109414, version_id=version_id) + pprint(response) + self.assertTrue(response) + + @attr('api') + def test_create_delete_author(self): + full_name = "Test Author " + datetime.now().strftime('%Y%m%d%H%M%S') + response = self.client.create_author(email='test@example.com', full_name=full_name) + pprint(response) + self.assertTrue(response) + + author_id = response['id'] + + response = self.client.update_author(author_id=author_id, email="new-email@example.com") + pprint(response) + self.assertTrue(response) + + response = self.client.delete_author(author_id=author_id) + pprint(response) + self.assertTrue(response) + + authors = self.client.undelete_author(author_id=author_id) + pprint(authors) + self.assertTrue(authors) + + response = self.client.delete_author(author_id=author_id) + pprint(response) + self.assertTrue(response) + + @attr('api') + def test_get_authors(self): + response = self.client.get_authors(query={'full_name': 'Hub Spot'}) + pprint(response) + self.assertTrue(response) + + author = self.client.get_author(author_id=response['objects'][0]['id']) + pprint(author) + self.assertTrue(author) + + @attr('api') + def test_get_topics(self): + response = self.client.get_topics() + pprint(response) + self.assertTrue(response) + + topic_id = response['objects'][0]['id'] + response = self.client.get_topic(topic_id=topic_id) + pprint(response) + self.assertTrue(response) + + @attr('api') + def test_create_update_delete_undelete_topic(self): + name = "Test Topic " + datetime.now().strftime('%Y%m%d%H%M%S') + response = self.client.create_topic(name=name) + pprint(response) + self.assertTrue(response) + + topic_id = response['id'] + + name = "New " + name + response = self.client.update_topic(topic_id=topic_id, name=name) + pprint(response) + self.assertTrue(response) + + response = self.client.delete_topic(topic_id=topic_id) + pprint(response) + self.assertTrue(response) + + response = self.client.undelete_topic(topic_id=topic_id) + pprint(response) + self.assertTrue(response) + + response = self.client.delete_topic(topic_id=topic_id) + pprint(response) + self.assertTrue(response) + +if __name__ == "__main__": + unittest2.main() diff --git a/hapi/test/test_nurturing.py b/hapi/test/test_nurturing.py deleted file mode 100644 index d1897a4..0000000 --- a/hapi/test/test_nurturing.py +++ /dev/null @@ -1,44 +0,0 @@ -import unittest2 -import helper -from nose.plugins.attrib import attr -from hapi.nurturing import NurturingClient -from hapi.mixins.threading import PyCurlMixin - -class NurturingClientTest(unittest2.TestCase): - def setUp(self): - self.client = NurturingClient(**helper.get_options()) - - def tearDown(self): - pass - - @attr('api') - def test_get_campaigns(self): - campaigns = self.client.get_campaigns() - self._check_result(campaigns) - - def test_parallel_get_campaigns(self): - # Because the client established in setup does not have PyCurlMixin, - # make another client to handle parallel get campaigns. - # All the options are the same, except mixins is PyCurlMixin - opts = helper.get_options() - - opts['mixins'] = [PyCurlMixin] - - self.parallel_client = NurturingClient(**opts) - - self.parallel_client.get_campaigns() - self.parallel_client.get_campaigns() - - results = self.parallel_client.process_queue() - for result in results: - self.assertTrue('data' in result) - self.assertTrue('code' in result) - self._check_result(result['data']) - - def _check_result(self, campaigns): - print "\n\n------ current lead nurturing campaigns ------" - print "%s\n" % '\n'.join([str(c) for c in campaigns]) - self.assertTrue(len(campaigns)) - -if __name__ == "__main__": - unittest2.main() diff --git a/setup.py b/setup.py index 4db4c28..e4804ed 100755 --- a/setup.py +++ b/setup.py @@ -3,13 +3,12 @@ setup( name='hapipy', - version='2.10.3', + version='2.10.6-curata-0', description="A python wrapper around HubSpot's APIs", long_description=open('README.md').read(), author='HubSpot Dev Team', author_email='devteam+hapi@hubspot.com', - url='https://github.com/HubSpot/hapipy', - download_url='https://github.com/HubSpot/hapipy/tarball/v2.10.3', + url='https://github.com/CurataEng/hapipy', license='LICENSE.txt', packages=['hapi', 'hapi.mixins'], install_requires=[