#!/usr/bin/python2.4 # # 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. """Base class to use OAuth to talk to the wave service.""" import httplib import logging import urllib import urlparse import oauth import simplejson import ops import blip import errors import events import search import util import wavelet class WaveService(object): # Google OAuth URLs REQUEST_TOKEN_URL = 'https://www.google.com/accounts/OAuthGetRequestToken' ACCESS_TOKEN_URL = 'https://www.google.com/accounts/OAuthGetAccessToken' AUTHORIZATION_URL = 'https://www.google.com/accounts/OAuthAuthorizeToken' # Wave OAuth URLS SCOPE = 'http://wave.googleusercontent.com/api/rpc' SIGNATURE_METHOD = oauth.OAuthSignatureMethod_HMAC_SHA1() # Wave RPC URLs RPC_URL = 'https://www-opensocial.googleusercontent.com/api/rpc' SANDBOX_RPC_URL = ( 'https://www-opensocial-sandbox.googleusercontent.com/api/rpc') def __init__(self, use_sandbox=False, server_rpc_base=None, consumer_key='anonymous', consumer_secret='anonymous', http_post=None): """Initializes a service that can perform the various OAuth steps. Args: use_sandbox: A boolean indicating whether to use Wave Sandbox URLs server_rpc_base: optional explicit url to use for rpc, overriding use_sandbox. consumer_key: A string for the consumer key, defaults to 'anonymous' consumer_secret: A string for the consumer secret, defaults to 'anonymous' http_post: handler to call to execute a http post. """ self._consumer = oauth.OAuthConsumer(consumer_key, consumer_secret) logging.info('server_rpc_base: %s', server_rpc_base) if server_rpc_base: self._server_rpc_base = server_rpc_base elif use_sandbox: self._server_rpc_base = WaveService.SANDBOX_RPC_URL else: self._server_rpc_base = WaveService.RPC_URL logging.info('server:' + self._server_rpc_base) self._http_post = self.http_post self._connection = httplib.HTTPSConnection('www.google.com') self._access_token = None def _make_token(self, token): """If passed an oauth token, return that. If passed a string, convert.""" if isinstance(token, basestring): return oauth.OAuthToken.from_string(token) else: return token def set_http_post(self, http_post): """Set the http_post handler to use when posting.""" self._http_post = http_post def get_token_from_request(self, oauth_request): """Convenience function to returning the token from a request. Args: oauth_request: An OAuthRequest object Returns: An OAuthToken object """ # Send request to the request token URL self._connection.request(oauth_request.http_method, oauth_request.to_url()) # Extract token from response response = self._connection.getresponse().read() self._request_token = oauth.OAuthToken.from_string(response) return self._request_token def fetch_request_token(self, callback=None): """Fetches the request token to start the oauth dance. Args: callback: the URL to where the service will redirect to after access is granted. Returns: An OAuthToken object """ # Create and sign OAuth request params = {'scope': WaveService.SCOPE} if callback: params['oauth_callback'] = callback oauth_request = oauth.OAuthRequest.from_consumer_and_token(self._consumer, http_url=WaveService.REQUEST_TOKEN_URL, parameters=params) oauth_request.sign_request(WaveService.SIGNATURE_METHOD, self._consumer, None) return self.get_token_from_request(oauth_request) def generate_authorization_url(self, request_token=None): """Generates the authorization URL (Step 2). Args: request_token: An OAuthToken object Returns: An authorization URL """ # Create Authorization URL request if request_token is None: request_token = self._request_token oauth_request = oauth.OAuthRequest.from_token_and_callback( token=request_token, http_url=WaveService.AUTHORIZATION_URL) # Send request self._connection.request(oauth_request.http_method, oauth_request.to_url()) # Extract location from the response response = self._connection.getresponse() return response.getheader('location') def upgrade_to_access_token(self, request_token, verifier=None): """Upgrades the request_token to an access token (Step 3). Args: request_token: An OAuthToken object or string verifier: A verifier string Returns: An OAuthToken object """ request_token = self._make_token(request_token) params = {} if verifier: params['oauth_verifier'] = verifier oauth_request = oauth.OAuthRequest.from_consumer_and_token(self._consumer, token=request_token, http_url=WaveService.ACCESS_TOKEN_URL, parameters=params) oauth_request.sign_request(WaveService.SIGNATURE_METHOD, self._consumer, request_token) self._access_token = self.get_token_from_request(oauth_request) return self._access_token def set_access_token(self, access_token): self._access_token = self._make_token(access_token) def http_post(self, url, data, headers): """Execute an http post. You can provide a different method to use in the constructor. This is mostly useful when running on app engine and you want to set the time out to something different than the default 5 seconds. Args: url: to post to body: post body headers: extra headers to pass along Returns: response_code, returned_page """ import urllib2 req = urllib2.Request(url, data=data, headers=headers) try: f = urllib2.urlopen(req) return f.code, f.read() except urllib2.HTTPError, e: return e.code, e.read() def make_rpc(self, operations): """Make an rpc call, submitting the specified operations.""" rpc_host = urlparse.urlparse(self._server_rpc_base).netloc # We either expect an operationqueue, a single op or a list # of ops: if (not isinstance(operations, ops.OperationQueue)): if not isinstance(operations, list): operations = [operations] queue = ops.OperationQueue() queue.copy_operations(operations) else: queue = operations data = simplejson.dumps(queue.serialize(method_prefix='wave')) oauth_request = oauth.OAuthRequest.from_consumer_and_token(self._consumer, token=self._access_token, http_method='POST', http_url=self._server_rpc_base) oauth_request.sign_request(WaveService.SIGNATURE_METHOD, self._consumer, self._access_token) logging.info('Active URL: %s' % self._server_rpc_base) logging.info('Active Outgoing: %s' % data) headers = {'Content-Type': 'application/json'} headers.update(oauth_request.to_header()); status, content = self._http_post( url=self._server_rpc_base, data=data, headers=headers) if status != 200: raise errors.RpcError('code: %s\n%s' % (status, content)) return simplejson.loads(content) def _first_rpc_result(self, result): """result is returned from make_rpc. Get the first data record or throw an exception if it was an error. Ignore responses to NOTIFY_OP_ID.""" result = [record for record in result if record['id'] != ops.NOTIFY_OP_ID] if not result: raise errors.RpcError('No results found.') result = result[0] error = result.get('error') if error: raise errors.RpcError(str(error['code']) + ': ' + error['message']) data = result.get('data') if data is not None: return data raise errors.Error('RPC Error: No data record.') def _wavelet_from_json(self, json, pending_ops): """Construct a wavelet from the passed json. The json should either contain a wavelet and a blips record that define those respective object. The returned wavelet will be constructed using the passed pending_ops OperationQueue. Alternatively the json can be the result of a previous wavelet.serialize() call. In that case the blips will be contaned in the wavelet record. """ if isinstance(json, basestring): json = simplejson.loads(json) # Create blips dict so we can pass into BlipThread objects blips = {} # Setup threads first, as the Blips and Wavelet need to know about them threads = {} # In case of blind_wavelet or new_wave, we may not have threads indo threads_data = json.get('threads', {}) # Create remaining thread objects for thread_id, raw_thread_data in threads_data.items(): threads[thread_id] = wavelet.BlipThread(thread_id, raw_thread_data.get('location'), raw_thread_data.get('blipIds', []), blips, pending_ops) # If being called from blind_wavelet, wavelet is top level info if 'wavelet' in json: raw_wavelet_data = json['wavelet'] elif 'waveletData' in json: raw_wavelet_data = json['waveletData'] else: raw_wavelet_data = json root_thread_data = raw_wavelet_data.get('rootThread') root_thread = wavelet.BlipThread('', root_thread_data.get('location'), root_thread_data.get('blipIds', []), blips, pending_ops) threads[''] = root_thread # Setup the blips, pass in reply threads for blip_id, raw_blip_data in json['blips'].items(): reply_threads = [threads[id] for id in raw_blip_data.get('replyThreadIds', [])] thread = threads.get(raw_blip_data.get('threadId')) blips[blip_id] = blip.Blip(raw_blip_data, blips, pending_ops, thread=thread, reply_threads=reply_threads) result = wavelet.Wavelet(raw_wavelet_data, blips, root_thread, pending_ops, raw_deltas=json.get('rawDeltas')) robot_address = json.get('robotAddress') if robot_address: result.robot_address = robot_address return result def search(self, query, index=None, num_results=None): """Execute a search request. Args: query: what to search for, for example [in:inbox] index: index of the first result to return num_results: how many results to return """ operation_queue = ops.OperationQueue() operation_queue.robot_search(query, index, num_results) result = self._first_rpc_result(self.make_rpc(operation_queue)) return search.Results(result) def new_wave(self, domain, participants=None, message='', proxy_for_id=None, submit=False): """Create a new wave with the initial participants on it. A new wave is returned with its own operation queue. It the responsibility of the caller to make sure this wave gets submitted to the server, either by calling robot.submit() or by calling .submit_with() on the returned wave. Args: domain: the domain to create the wavelet on. This should in general correspond to the domain of the incoming wavelet. (wavelet.domain). Exceptions are situations where the robot is calling new_wave outside of an event or when the server is handling multiple domains. participants: initial participants on the wave. The robot as the creator of the wave is always added. message: a string that will be passed back to the robot when the WAVELET_CREATOR event is fired. This is a lightweight way to pass around state. submit: if true, use the active gateway to make a round trip to the server. This will return immediately an actual waveid/waveletid and blipId for the root blip. """ util.check_is_valid_proxy_for_id(proxy_for_id) operation_queue = ops.OperationQueue(proxy_for_id) if not isinstance(message, basestring): message = simplejson.dumps(message) # Create temporary wavelet data blip_data, wavelet_data = operation_queue.robot_create_wavelet( domain=domain, participants=participants, message=message) # Create temporary blips dictionary blips = {} root_blip = blip.Blip(blip_data, blips, operation_queue) blips[root_blip.blip_id] = root_blip if submit: # Submit operation to server and return actual wave/blip IDs temp_wavelet = wavelet.Wavelet(wavelet_data, blips=blips, root_thread=None, operation_queue=operation_queue) result = self._first_rpc_result(self.submit(temp_wavelet)) if isinstance(result, list): result = result[0] if 'blipId' in result: blip_data['blipId'] = result['blipId'] wavelet_data['rootBlipId'] = result['blipId'] for field in 'waveId', 'waveletId': if field in result: wavelet_data[field] = result[field] blip_data[field] = result[field] blips = {} root_blip = blip.Blip(blip_data, blips, operation_queue) blips[root_blip.blip_id] = root_blip root_thread = wavelet.BlipThread('', -1, [root_blip.blip_id], blips, operation_queue) new_wavelet = wavelet.Wavelet(wavelet_data, blips=blips, root_thread=root_thread, operation_queue=operation_queue) return new_wavelet def fetch_wavelet(self, wave_id, wavelet_id=None, proxy_for_id=None, raw_deltas_from_version=-1, return_raw_snapshot=False): """Use the REST interface to fetch a wave and return it. The returned wavelet contains a snapshot of the state of the wavelet at that point. It can be used to modify the wavelet, but the wavelet might change in between, so treat carefully. Also note that the wavelet returned has its own operation queue. It the responsibility of the caller to make sure this wavelet gets submited to the server, either by calling robot.submit() or by calling .submit_with() on the returned wavelet. Args: wave_id: the wave id wavelet_id: the wavelet_id proxy_for_id: on whose behalf to execute the operation raw_deltas_from_version: If specified, return a raw dump of the delta history of this wavelet, starting at the given version. This may return only part of the history; use additional requests with higher raw_deltas_from_version parameters to get the rest. return_raw_snapshot: if true, return the raw data for this wavelet. """ util.check_is_valid_proxy_for_id(proxy_for_id) if not wavelet_id: domain, id = wave_id.split('!', 1) wavelet_id = domain + '!conv+root' operation_queue = ops.OperationQueue(proxy_for_id) operation_queue.robot_fetch_wave(wave_id, wavelet_id, raw_deltas_from_version, return_raw_snapshot) result = self._first_rpc_result(self.make_rpc(operation_queue)) return self._wavelet_from_json(result, ops.OperationQueue(proxy_for_id)) def blind_wavelet(self, json, proxy_for_id=None): """Construct a blind wave from a json string. Call this method if you have a snapshot of a wave that you want to operate on outside of an event. Since the wave might have changed since you last saw it, you should take care to submit operations that are as safe as possible. Args: json: a json object or string containing at least a key wavelet defining the wavelet and a key blips defining the blips in the view. proxy_for_id: the proxying information that will be set on the wavelet's operation queue. Returns: A new wavelet with its own operation queue. It the responsibility of the caller to make sure this wavelet gets submited to the server, either by calling robot.submit() or by calling .submit_with() on the returned wavelet. """ util.check_is_valid_proxy_for_id(proxy_for_id) return self._wavelet_from_json(json, ops.OperationQueue(proxy_for_id)) def submit(self, wavelet_to_submit): """Submit the pending operations associated with wavelet_to_submit. Typically the wavelet will be the result of fetch_wavelet, blind_wavelet or new_wave. """ pending = wavelet_to_submit.get_operation_queue() res = self.make_rpc(pending) pending.clear() return res