A new push implementation. This updates us to the latest revision of the push
API which has a much simpler interface. Look in the module doc string for how
this API would be used.
Reviewed in https://codereview.appspot.com/9885043/.
diff --git a/apiclient/channel.py b/apiclient/channel.py
new file mode 100644
index 0000000..61a7ec4
--- /dev/null
+++ b/apiclient/channel.py
@@ -0,0 +1,285 @@
+"""Channel notifications support.
+
+Classes and functions to support channel subscriptions and notifications
+on those channels.
+
+Notes:
+ - This code is based on experimental APIs and is subject to change.
+ - Notification does not do deduplication of notification ids, that's up to
+ the receiver.
+ - Storing the Channel between calls is up to the caller.
+
+
+Example setting up a channel:
+
+ # Create a new channel that gets notifications via webhook.
+ channel = new_webhook_channel("https://example.com/my_web_hook")
+
+ # Store the channel, keyed by 'channel.id'. Store it before calling the
+ # watch method because notifications may start arriving before the watch
+ # method returns.
+ ...
+
+ resp = service.objects().watchAll(
+ bucket="some_bucket_id", body=channel.body()).execute()
+ channel.update(resp)
+
+ # Store the channel, keyed by 'channel.id'. Store it after being updated
+ # since the resource_id value will now be correct, and that's needed to
+ # stop a subscription.
+ ...
+
+
+An example Webhook implementation using webapp2. Note that webapp2 puts
+headers in a case insensitive dictionary, as headers aren't guaranteed to
+always be upper case.
+
+ id = self.request.headers[X_GOOG_CHANNEL_ID]
+
+ # Retrieve the channel by id.
+ channel = ...
+
+ # Parse notification from the headers, including validating the id.
+ n = notification_from_headers(channel, self.request.headers)
+
+ # Do app specific stuff with the notification here.
+ if n.resource_state == 'sync':
+ # Code to handle sync state.
+ elif n.resource_state == 'exists':
+ # Code to handle the exists state.
+ elif n.resource_state == 'not_exists':
+ # Code to handle the not exists state.
+
+
+Example of unsubscribing.
+
+ service.channels().stop(channel.body())
+"""
+
+import datetime
+import uuid
+
+from apiclient import errors
+from oauth2client import util
+
+
+# The unix time epoch starts at midnight 1970.
+EPOCH = datetime.datetime.utcfromtimestamp(0)
+
+# Map the names of the parameters in the JSON channel description to
+# the parameter names we use in the Channel class.
+CHANNEL_PARAMS = {
+ 'address': 'address',
+ 'id': 'id',
+ 'expiration': 'expiration',
+ 'params': 'params',
+ 'resourceId': 'resource_id',
+ 'resourceUri': 'resource_uri',
+ 'type': 'type',
+ 'token': 'token',
+ }
+
+X_GOOG_CHANNEL_ID = 'X-GOOG-CHANNEL-ID'
+X_GOOG_MESSAGE_NUMBER = 'X-GOOG-MESSAGE-NUMBER'
+X_GOOG_RESOURCE_STATE = 'X-GOOG-RESOURCE-STATE'
+X_GOOG_RESOURCE_URI = 'X-GOOG-RESOURCE-URI'
+X_GOOG_RESOURCE_ID = 'X-GOOG-RESOURCE-ID'
+
+
+def _upper_header_keys(headers):
+ new_headers = {}
+ for k, v in headers.iteritems():
+ new_headers[k.upper()] = v
+ return new_headers
+
+
+class Notification(object):
+ """A Notification from a Channel.
+
+ Notifications are not usually constructed directly, but are returned
+ from functions like notification_from_headers().
+
+ Attributes:
+ message_number: int, The unique id number of this notification.
+ state: str, The state of the resource being monitored.
+ uri: str, The address of the resource being monitored.
+ resource_id: str, The unique identifier of the version of the resource at
+ this event.
+ """
+ @util.positional(5)
+ def __init__(self, message_number, state, resource_uri, resource_id):
+ """Notification constructor.
+
+ Args:
+ message_number: int, The unique id number of this notification.
+ state: str, The state of the resource being monitored. Can be one
+ of "exists", "not_exists", or "sync".
+ resource_uri: str, The address of the resource being monitored.
+ resource_id: str, The identifier of the watched resource.
+ """
+ self.message_number = message_number
+ self.state = state
+ self.resource_uri = resource_uri
+ self.resource_id = resource_id
+
+
+class Channel(object):
+ """A Channel for notifications.
+
+ Usually not constructed directly, instead it is returned from helper
+ functions like new_webhook_channel().
+
+ Attributes:
+ type: str, The type of delivery mechanism used by this channel. For
+ example, 'web_hook'.
+ id: str, A UUID for the channel.
+ token: str, An arbitrary string associated with the channel that
+ is delivered to the target address with each event delivered
+ over this channel.
+ address: str, The address of the receiving entity where events are
+ delivered. Specific to the channel type.
+ expiration: int, The time, in milliseconds from the epoch, when this
+ channel will expire.
+ params: dict, A dictionary of string to string, with additional parameters
+ controlling delivery channel behavior.
+ resource_id: str, An opaque id that identifies the resource that is
+ being watched. Stable across different API versions.
+ resource_uri: str, The canonicalized ID of the watched resource.
+ """
+
+ @util.positional(5)
+ def __init__(self, type, id, token, address, expiration=None,
+ params=None, resource_id="", resource_uri=""):
+ """Create a new Channel.
+
+ In user code, this Channel constructor will not typically be called
+ manually since there are functions for creating channels for each specific
+ type with a more customized set of arguments to pass.
+
+ Args:
+ type: str, The type of delivery mechanism used by this channel. For
+ example, 'web_hook'.
+ id: str, A UUID for the channel.
+ token: str, An arbitrary string associated with the channel that
+ is delivered to the target address with each event delivered
+ over this channel.
+ address: str, The address of the receiving entity where events are
+ delivered. Specific to the channel type.
+ expiration: int, The time, in milliseconds from the epoch, when this
+ channel will expire.
+ params: dict, A dictionary of string to string, with additional parameters
+ controlling delivery channel behavior.
+ resource_id: str, An opaque id that identifies the resource that is
+ being watched. Stable across different API versions.
+ resource_uri: str, The canonicalized ID of the watched resource.
+ """
+ self.type = type
+ self.id = id
+ self.token = token
+ self.address = address
+ self.expiration = expiration
+ self.params = params
+ self.resource_id = resource_id
+ self.resource_uri = resource_uri
+
+ def body(self):
+ """Build a body from the Channel.
+
+ Constructs a dictionary that's appropriate for passing into watch()
+ methods as the value of body argument.
+
+ Returns:
+ A dictionary representation of the channel.
+ """
+ result = {
+ 'id': self.id,
+ 'token': self.token,
+ 'type': self.type,
+ 'address': self.address
+ }
+ if self.params:
+ result['params'] = self.params
+ if self.resource_id:
+ result['resourceId'] = self.resource_id
+ if self.resource_uri:
+ result['resourceUri'] = self.resource_uri
+ if self.expiration:
+ result['expiration'] = self.expiration
+
+ return result
+
+ def update(self, resp):
+ """Update a channel with information from the response of watch().
+
+ When a request is sent to watch() a resource, the response returned
+ from the watch() request is a dictionary with updated channel information,
+ such as the resource_id, which is needed when stopping a subscription.
+
+ Args:
+ resp: dict, The response from a watch() method.
+ """
+ for json_name, param_name in CHANNEL_PARAMS.iteritems():
+ value = resp.get(json_name)
+ if value is not None:
+ setattr(self, param_name, value)
+
+
+def notification_from_headers(channel, headers):
+ """Parse a notification from the webhook request headers, validate
+ the notification, and return a Notification object.
+
+ Args:
+ channel: Channel, The channel that the notification is associated with.
+ headers: dict, A dictionary like object that contains the request headers
+ from the webhook HTTP request.
+
+ Returns:
+ A Notification object.
+
+ Raises:
+ errors.InvalidNotificationError if the notification is invalid.
+ ValueError if the X-GOOG-MESSAGE-NUMBER can't be converted to an int.
+ """
+ headers = _upper_header_keys(headers)
+ channel_id = headers[X_GOOG_CHANNEL_ID]
+ if channel.id != channel_id:
+ raise errors.InvalidNotificationError(
+ 'Channel id mismatch: %s != %s' % (channel.id, channel_id))
+ else:
+ message_number = int(headers[X_GOOG_MESSAGE_NUMBER])
+ state = headers[X_GOOG_RESOURCE_STATE]
+ resource_uri = headers[X_GOOG_RESOURCE_URI]
+ resource_id = headers[X_GOOG_RESOURCE_ID]
+ return Notification(message_number, state, resource_uri, resource_id)
+
+
+@util.positional(2)
+def new_webhook_channel(url, token=None, expiration=None, params=None):
+ """Create a new webhook Channel.
+
+ Args:
+ url: str, URL to post notifications to.
+ token: str, An arbitrary string associated with the channel that
+ is delivered to the target address with each notification delivered
+ over this channel.
+ expiration: datetime.datetime, A time in the future when the channel
+ should expire. Can also be None if the subscription should use the
+ default expiration. Note that different services may have different
+ limits on how long a subscription lasts. Check the response from the
+ watch() method to see the value the service has set for an expiration
+ time.
+ params: dict, Extra parameters to pass on channel creation. Currently
+ not used for webhook channels.
+ """
+ expiration_ms = 0
+ if expiration:
+ delta = expiration - EPOCH
+ expiration_ms = delta.microseconds/1000 + (
+ delta.seconds + delta.days*24*3600)*1000
+ if expiration_ms < 0:
+ expiration_ms = 0
+
+ return Channel('web_hook', str(uuid.uuid4()),
+ token, url, expiration=expiration_ms,
+ params=params)
+
diff --git a/apiclient/errors.py b/apiclient/errors.py
index 2bf9149..ef2b161 100644
--- a/apiclient/errors.py
+++ b/apiclient/errors.py
@@ -102,6 +102,9 @@
"""The given chunksize is not valid."""
pass
+class InvalidNotificationError(Error):
+ """The channel Notification is invalid."""
+ pass
class BatchError(HttpError):
"""Error occured during batch operations."""
diff --git a/apiclient/push.py b/apiclient/push.py
deleted file mode 100644
index c520faf..0000000
--- a/apiclient/push.py
+++ /dev/null
@@ -1,274 +0,0 @@
-# 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.
-
-"""Push notifications support.
-
-This code is based on experimental APIs and is subject to change.
-"""
-
-__author__ = 'afshar@google.com (Ali Afshar)'
-
-import binascii
-import collections
-import os
-import urllib
-
-SUBSCRIBE = 'X-GOOG-SUBSCRIBE'
-SUBSCRIPTION_ID = 'X-GOOG-SUBSCRIPTION-ID'
-TOPIC_ID = 'X-GOOG-TOPIC-ID'
-TOPIC_URI = 'X-GOOG-TOPIC-URI'
-CLIENT_TOKEN = 'X-GOOG-CLIENT-TOKEN'
-EVENT_TYPE = 'X-GOOG-EVENT-TYPE'
-UNSUBSCRIBE = 'X-GOOG-UNSUBSCRIBE'
-
-
-class InvalidSubscriptionRequestError(ValueError):
- """The request cannot be subscribed."""
-
-
-def new_token():
- """Gets a random token for use as a client_token in push notifications.
-
- Returns:
- str, a new random token.
- """
- return binascii.hexlify(os.urandom(32))
-
-
-class Channel(object):
- """Base class for channel types."""
-
- def __init__(self, channel_type, channel_args):
- """Create a new Channel.
-
- You probably won't need to create this channel manually, since there are
- subclassed Channel for each specific type with a more customized set of
- arguments to pass. However, you may wish to just create it manually here.
-
- Args:
- channel_type: str, the type of channel.
- channel_args: dict, arguments to pass to the channel.
- """
- self.channel_type = channel_type
- self.channel_args = channel_args
-
- def as_header_value(self):
- """Create the appropriate header for this channel.
-
- Returns:
- str encoded channel description suitable for use as a header.
- """
- return '%s?%s' % (self.channel_type, urllib.urlencode(self.channel_args))
-
- def write_header(self, headers):
- """Write the appropriate subscribe header to a headers dict.
-
- Args:
- headers: dict, headers to add subscribe header to.
- """
- headers[SUBSCRIBE] = self.as_header_value()
-
-
-class WebhookChannel(Channel):
- """Channel for registering web hook notifications."""
-
- def __init__(self, url, app_engine=False):
- """Create a new WebhookChannel
-
- Args:
- url: str, URL to post notifications to.
- app_engine: bool, default=False, whether the destination for the
- notifications is an App Engine application.
- """
- super(WebhookChannel, self).__init__(
- channel_type='web_hook',
- channel_args={
- 'url': url,
- 'app_engine': app_engine and 'true' or 'false',
- }
- )
-
-
-class Headers(collections.defaultdict):
- """Headers for managing subscriptions."""
-
-
- ALL_HEADERS = set([SUBSCRIBE, SUBSCRIPTION_ID, TOPIC_ID, TOPIC_URI,
- CLIENT_TOKEN, EVENT_TYPE, UNSUBSCRIBE])
-
- def __init__(self):
- """Create a new subscription configuration instance."""
- collections.defaultdict.__init__(self, str)
-
- def __setitem__(self, key, value):
- """Set a header value, ensuring the key is an allowed value.
-
- Args:
- key: str, the header key.
- value: str, the header value.
- Raises:
- ValueError if key is not one of the accepted headers.
- """
- normal_key = self._normalize_key(key)
- if normal_key not in self.ALL_HEADERS:
- raise ValueError('Header name must be one of %s.' % self.ALL_HEADERS)
- else:
- return collections.defaultdict.__setitem__(self, normal_key, value)
-
- def __getitem__(self, key):
- """Get a header value, normalizing the key case.
-
- Args:
- key: str, the header key.
- Returns:
- String header value.
- Raises:
- KeyError if the key is not one of the accepted headers.
- """
- normal_key = self._normalize_key(key)
- if normal_key not in self.ALL_HEADERS:
- raise ValueError('Header name must be one of %s.' % self.ALL_HEADERS)
- else:
- return collections.defaultdict.__getitem__(self, normal_key)
-
- def _normalize_key(self, key):
- """Normalize a header name for use as a key."""
- return key.upper()
-
- def items(self):
- """Generator for each header."""
- for header in self.ALL_HEADERS:
- value = self[header]
- if value:
- yield header, value
-
- def write(self, headers):
- """Applies the subscription headers.
-
- Args:
- headers: dict of headers to insert values into.
- """
- for header, value in self.items():
- headers[header.lower()] = value
-
- def read(self, headers):
- """Read from headers.
-
- Args:
- headers: dict of headers to read from.
- """
- for header in self.ALL_HEADERS:
- if header.lower() in headers:
- self[header] = headers[header.lower()]
-
-
-class Subscription(object):
- """Information about a subscription."""
-
- def __init__(self):
- """Create a new Subscription."""
- self.headers = Headers()
-
- @classmethod
- def for_request(cls, request, channel, client_token=None):
- """Creates a subscription and attaches it to a request.
-
- Args:
- request: An http.HttpRequest to modify for making a subscription.
- channel: A apiclient.push.Channel describing the subscription to
- create.
- client_token: (optional) client token to verify the notification.
-
- Returns:
- New subscription object.
- """
- subscription = cls.for_channel(channel=channel, client_token=client_token)
- subscription.headers.write(request.headers)
- if request.method != 'GET':
- raise InvalidSubscriptionRequestError(
- 'Can only subscribe to requests which are GET.')
- request.method = 'POST'
-
- def _on_response(response, subscription=subscription):
- """Called with the response headers. Reads the subscription headers."""
- subscription.headers.read(response)
-
- request.add_response_callback(_on_response)
- return subscription
-
- @classmethod
- def for_channel(cls, channel, client_token=None):
- """Alternate constructor to create a subscription from a channel.
-
- Args:
- channel: A apiclient.push.Channel describing the subscription to
- create.
- client_token: (optional) client token to verify the notification.
-
- Returns:
- New subscription object.
- """
- subscription = cls()
- channel.write_header(subscription.headers)
- if client_token is None:
- client_token = new_token()
- subscription.headers[SUBSCRIPTION_ID] = new_token()
- subscription.headers[CLIENT_TOKEN] = client_token
- return subscription
-
- def verify(self, headers):
- """Verifies that a webhook notification has the correct client_token.
-
- Args:
- headers: dict of request headers for a push notification.
-
- Returns:
- Boolean value indicating whether the notification is verified.
- """
- new_subscription = Subscription()
- new_subscription.headers.read(headers)
- return new_subscription.client_token == self.client_token
-
- @property
- def subscribe(self):
- """Subscribe header value."""
- return self.headers[SUBSCRIBE]
-
- @property
- def subscription_id(self):
- """Subscription ID header value."""
- return self.headers[SUBSCRIPTION_ID]
-
- @property
- def topic_id(self):
- """Topic ID header value."""
- return self.headers[TOPIC_ID]
-
- @property
- def topic_uri(self):
- """Topic URI header value."""
- return self.headers[TOPIC_URI]
-
- @property
- def client_token(self):
- """Client Token header value."""
- return self.headers[CLIENT_TOKEN]
-
- @property
- def event_type(self):
- """Event Type header value."""
- return self.headers[EVENT_TYPE]
-
- @property
- def unsubscribe(self):
- """Unsuscribe header value."""
- return self.headers[UNSUBSCRIBE]