Temporarily checked in Buzz appengine wrapper class. Will move to contrib and figure out how to handle the relative import once this is more stable.
diff --git a/buzz_gae_client.py b/buzz_gae_client.py
new file mode 100644
index 0000000..41b6c6b
--- /dev/null
+++ b/buzz_gae_client.py
@@ -0,0 +1,101 @@
+# Copyright (C) 2010 Google Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+__author__ = 'ade@google.com'
+
+import apiclient.discovery
+import logging
+import oauth_wrap
+import oauth2 as oauth
+import urllib
+import urlparse
+
+try:
+ from urlparse import parse_qs, parse_qsl, urlparse
+except ImportError:
+ from cgi import parse_qs, parse_qsl
+
+# TODO(ade) Replace user-agent with something specific
+HEADERS = {
+ 'user-agent': 'gdata-python-v3-sample-client/0.1',
+ 'content-type': 'application/x-www-form-urlencoded'
+ }
+
+REQUEST_TOKEN_URL = 'https://www.google.com/accounts/OAuthGetRequestToken?domain=anonymous&scope=https://www.googleapis.com/auth/buzz'
+AUTHORIZE_URL = 'https://www.google.com/buzz/api/auth/OAuthAuthorizeToken?domain=anonymous&scope=https://www.googleapis.com/auth/buzz'
+ACCESS_TOKEN_URL = 'https://www.google.com/accounts/OAuthGetAccessToken'
+
+class BuzzGaeClient(object):
+ def __init__(self, consumer_key='anonymous', consumer_secret='anonymous'):
+ self.consumer = oauth.Consumer(consumer_key, consumer_secret)
+ self.consumer_key = consumer_key
+ self.consumer_secret = consumer_secret
+
+ def _make_post_request(self, client, url, parameters):
+ resp, content = client.request(url, 'POST', headers=HEADERS,
+ body=urllib.urlencode(parameters, True))
+
+ if resp['status'] != '200':
+ logging.warn('Request: %s failed with status: %s. Content was: %s' % (url, resp['status'], content))
+ raise Exception('Invalid response %s.' % resp['status'])
+ return resp, content
+
+ def get_request_token(self, callback_url, display_name = None):
+ parameters = {
+ 'oauth_callback': callback_url
+ }
+
+ if display_name is not None:
+ parameters['xoauth_displayname'] = display_name
+
+ client = oauth.Client(self.consumer)
+ resp, content = self._make_post_request(client, REQUEST_TOKEN_URL, parameters)
+
+ request_token = dict(parse_qsl(content))
+ return request_token
+
+ def generate_authorisation_url(self, request_token):
+ """Returns the URL the user should be redirected to in other to gain access to their account."""
+
+ base_url = urlparse.urlparse(AUTHORIZE_URL)
+ query = parse_qs(base_url.query)
+ query['oauth_token'] = request_token['oauth_token']
+
+ logging.info(urllib.urlencode(query, True))
+
+ url = (base_url.scheme, base_url.netloc, base_url.path, base_url.params,
+ urllib.urlencode(query, True), base_url.fragment)
+ authorisation_url = urlparse.urlunparse(url)
+ return authorisation_url
+
+ def upgrade_to_access_token(self, request_token, oauth_verifier):
+ token = oauth.Token(request_token['oauth_token'],
+ request_token['oauth_token_secret'])
+ token.set_verifier(oauth_verifier)
+ client = oauth.Client(self.consumer, token)
+
+ parameters = {}
+ resp, content = self._make_post_request(client, ACCESS_TOKEN_URL, parameters)
+ access_token = dict(parse_qsl(content))
+
+ d = {
+ 'consumer_key' : self.consumer_key,
+ 'consumer_secret' : self.consumer_secret
+ }
+ d.update(access_token)
+ return d
+
+ def build_api_client(self, oauth_params):
+ http = oauth_wrap.get_authorised_http(oauth_params)
+ return apiclient.discovery.build('buzz', 'v1', http = http)
diff --git a/buzz_gae_client_tests.py b/buzz_gae_client_tests.py
new file mode 100644
index 0000000..fea5f68
--- /dev/null
+++ b/buzz_gae_client_tests.py
@@ -0,0 +1,67 @@
+# Copyright (C) 2010 Google Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+__author__ = 'ade@google.com'
+
+import buzz_gae_client
+import unittest
+
+class BuzzGaeClientTest(unittest.TestCase):
+ def test_can_build_client(self):
+ client = buzz_gae_client.BuzzGaeClient()
+ self.assertNotEquals(None, client)
+
+ def test_can_get_request_token(self):
+ client = buzz_gae_client.BuzzGaeClient()
+ callback_url = 'http://example.com'
+ request_token = client.get_request_token(callback_url)
+ self.assertTrue(request_token['oauth_token'] is not None)
+ self.assertTrue(request_token['oauth_token_secret'] is not None)
+ self.assertEquals(request_token['oauth_callback_confirmed'], 'true')
+
+ def test_can_generate_authorisation_url(self):
+ client = buzz_gae_client.BuzzGaeClient()
+ callback_url = 'http://example.com'
+ request_token = client.get_request_token(callback_url)
+ authorisation_url = client.generate_authorisation_url(request_token)
+ self.assertTrue(authorisation_url is not None)
+ self.assertTrue(authorisation_url.startswith('https://www.google.com/buzz/api/auth/OAuthAuthorizeToken?scope='))
+
+ # TODO(ade) this test can't work outside of the AppEngine environment. Find an equivalent that works.
+ def _test_can_upgrade_tokens(self):
+ client = buzz_gae_client.BuzzGaeClient()
+ callback_url = 'http://example.com'
+ request_token = client.get_request_token(callback_url)
+ oauth_verifier = 'some verifier'
+
+ consumer_key = 'anonymous'
+ consumer_secret = 'anonymous'
+ usable_token = client.upgrade_to_access_token(request_token, oauth_verifier)
+ self.assertTrue(usable_token is not None)
+ self.assertEquals(consumer_key, usable_token['consumer_key'])
+ self.assertEquals(consumer_secret, usable_token['consumer_secret'])
+ self.assertTrue(usable_token['access_token'] is not None)
+
+ def test_can_build_apiary_client_with_access_token(self):
+ client = buzz_gae_client.BuzzGaeClient()
+ oauth_parameters = {}
+ oauth_parameters['oauth_token'] = ''
+ oauth_parameters['oauth_token_secret'] = ''
+ oauth_parameters['consumer_key'] = ''
+ oauth_parameters['consumer_secret'] = ''
+ api_client = client.build_api_client(oauth_parameters)
+ self.assertTrue(api_client is not None)
+
+if __name__ == '__main__':
+ unittest.main()
\ No newline at end of file