#!/usr/bin/env python # # Copyright 2007 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. # """ProspectiveSearch API. A service that enables AppEngine apps to match queries to documents. Functions defined in this module: subscribe: Add a query to set of matching queries. unsubscribe: Remove query from set of matching queries. get_subscription: Retrieves subscription with particular id. list_subscriptions: Lists subscriptions on a particular topic. list_topics: Lists topics that have subscriptions. match: Match all subscribed queries to document. """ __all__ = ['get_document', 'get_subscription', 'list_subscriptions', 'list_topics', 'match', 'unsubscribe', 'subscribe', 'subscription_state_name', 'DEFAULT_RESULT_BATCH_SIZE', 'DEFAULT_LEASE_DURATION_SEC', 'DEFAULT_LIST_SUBSCRIPTIONS_MAX_RESULTS', 'DEFAULT_LIST_TOPICS_MAX_RESULTS', 'DocumentTypeError', 'Error', 'QuerySyntaxError', 'SchemaError', 'SubscriptionDoesNotExist', 'SubscriptionState', 'TopicNotSpecified'] import base64 import sys from google.appengine.api import apiproxy_stub_map from google.appengine.api import datastore from google.appengine.api import datastore_types from google.appengine.api.prospective_search import error_pb from google.appengine.api.prospective_search import prospective_search_admin from google.appengine.api.prospective_search import prospective_search_pb from google.appengine.runtime import apiproxy_errors from google.appengine.datastore import entity_pb DEFAULT_RESULT_BATCH_SIZE = 100 DEFAULT_LEASE_DURATION_SEC = 0 DEFAULT_LIST_SUBSCRIPTIONS_MAX_RESULTS = 1000 DEFAULT_LIST_TOPICS_MAX_RESULTS = 1000 _doc_class = prospective_search_pb.MatchRequest _schema_type = prospective_search_pb.SchemaEntry _entity_meaning = entity_pb.Property class SubscriptionState: OK = 0 PENDING = 1 ERROR = 2 _State_NAMES = { 0: "OK", 1: "PENDING", 2: "ERROR", } def subscription_state_name(x): return SubscriptionState._State_NAMES.get(x, "") def _GetSchemaEntryForPropertyType(property_type): """Converts db.Model type to internal schema type.""" from google.appengine.ext import db _MODEL_TYPE_TO_SCHEMA_ENTRY = { db.StringProperty: (_schema_type.STRING, None), db.IntegerProperty: (_schema_type.INT32, None), db.BooleanProperty: (_schema_type.BOOLEAN, None), db.FloatProperty: (_schema_type.DOUBLE, None), db.TextProperty: (_schema_type.STRING, None) } return _MODEL_TYPE_TO_SCHEMA_ENTRY.get(property_type, (None, None)) def _GetModelTypeForListPropertyType(property_type): """Converts (supported) db.ListProperty type to db.Model type.""" from google.appengine.ext import db _LISTPROPERTY_TYPE_TO_SCHEMA_ENTRY = { basestring: db.StringProperty, str: db.StringProperty, unicode: db.StringProperty, bool: db.BooleanProperty, int: db.IntegerProperty, float: db.FloatProperty, db.Text: db.TextProperty, } return _LISTPROPERTY_TYPE_TO_SCHEMA_ENTRY.get(property_type, None) def _GetModelTypeForEntityType(python_type): """Converts python built in type to db.Model type.""" from google.appengine.ext import db _PYTHON_TYPE_TO_MODEL_TYPE = { basestring: db.StringProperty, str: db.StringProperty, unicode: db.StringProperty, int: db.IntegerProperty, bool: db.BooleanProperty, float: db.FloatProperty, db.Text: db.TextProperty, } return _PYTHON_TYPE_TO_MODEL_TYPE.get(python_type, None) class Error(Exception): """Base error class for this module.""" class TopicNotSpecified(Error): def __str__(self): return 'Topic must be specified.' class SubscriptionDoesNotExist(Error): """Subscription does not exist.""" def __init__(self, topic, sub_id): Exception.__init__(self) self.topic = topic self.sub_id = sub_id def __str__(self): return "Subscription '%s' on topic '%s' does not exist." % (self.sub_id, self.topic) class DocumentTypeError(Error): """Document type is not supported.""" def __str__(self): return 'Document type is not supported.' class SchemaError(Error): """Schema error.""" def __init__(self, detail): Exception.__init__(self) self.detail = detail def __str__(self): return 'SchemaError: %s' % self.detail class QuerySyntaxError(Error): """Query syntax not valid error.""" def __init__(self, topic, sub_id, query, detail): Exception.__init__(self) self.topic = topic self.sub_id = sub_id self.query = query self.detail = detail def __str__(self): return "QuerySyntaxError: topic:'%s' sub_id:'%s' query:'%s' detail:'%s'" % ( self.topic, self.sub_id, self.query, self.detail) def _get_document_topic(document, topic): if topic: return topic return document.kind() def _make_sync_call(service, call, request, response): """The APIProxy entry point for a synchronous API call. Args: service: string representing which service to call call: string representing which function to call request: protocol buffer for the request response: protocol buffer for the response Returns: Response protocol buffer. Caller should always use returned value which may or may not be same as passed in 'response'. Raises: apiproxy_errors.Error: """ resp = apiproxy_stub_map.MakeSyncCall(service, call, request, response) if resp is not None: return resp return response def _add_schema_entry(model_type, name, add_entry): """Add single entry to SchemaEntries by invoking add_entry.""" schema_type, entity_meaning = _GetSchemaEntryForPropertyType(model_type) if not schema_type: return entry = add_entry() entry.set_name(name) entry.set_type(schema_type) if entity_meaning: entry.set_meaning(entity_meaning) def _entity_schema_to_prospective_search_schema(schema, add_entry): """Produce SchemaEntries from python schema. Args: schema: dictionary mapping python type to field names. add_entry: sink function for prospective search schema entries. Raises: SchemaError: schema is invalid. """ all_names = [] for python_type, names in schema.items(): all_names.extend(names) for name in names: model_type = _GetModelTypeForEntityType(python_type) if not model_type: continue _add_schema_entry(model_type, name, add_entry) if len(all_names) != len(set(all_names)): duplicate_names = all_names for name in set(all_names): duplicate_names.remove(name) raise SchemaError('Duplicate names in schema: %s' % duplicate_names) def _model_to_prospective_search_schema(model, add_entry): """Produce SchemaEntries from db.Model class.""" from google.appengine.ext import db for name, model_property in model.properties().iteritems(): model_class = model_property.__class__ if issubclass(model_class, db.ListProperty): model_class = _GetModelTypeForListPropertyType(model_property.item_type) _add_schema_entry(model_class, name, add_entry) def subscribe(document_class, query, sub_id, schema=None, topic=None, lease_duration_sec=DEFAULT_LEASE_DURATION_SEC): """Subscribe a query. If the document_class is a datastore.Entity, a schema must be specified to map document_class member names to the prospective_search supported types. For example, the datastore.Entity 'person' has the following schema: person = datastore.Entity('person') person['first_name'] = 'Andrew' person['surname'] = 'Smith' person['height'] = 150 person_schema = { str : ['first_name', 'surname'], int : ['height'], } Args: document_class: datastore.Entity or db.Model class. query: str or unicode query for documents of type document_class. sub_id: subscription id returned when this subscription is matched. schema: required for datastore.Entity document_class. topic: required for datastore.Entity document_class, optional for db.Model. If not specified for db.Model, topic is assumed to be the class name. Only documents of same topic will be matched against this subscription. lease_duration_sec: minimum number of seconds subscription should exist. 0 indicates subscription never expires (default). Raises: DocumentTypeError: document type is unsupported. TopicNotSpecified: raised for datastore.Entity if topic is not specified. QuerySyntaxError: raised when query is invalid or does not match schema. SchemaError: schema is invalid. apiproxy_errors.Error: subscribe call failed. """ from google.appengine.ext import db request = prospective_search_pb.SubscribeRequest() request.set_sub_id(sub_id) request.set_vanilla_query(unicode(query).encode('utf-8')) request.set_lease_duration_sec(lease_duration_sec) if issubclass(document_class, db.Model): topic = _get_document_topic(document_class, topic) _model_to_prospective_search_schema(document_class, request.add_schema_entry) elif issubclass(document_class, datastore.Entity): if not topic: raise TopicNotSpecified() _entity_schema_to_prospective_search_schema(schema, request.add_schema_entry) else: raise DocumentTypeError() request.set_topic(topic) response = prospective_search_pb.SubscribeResponse() try: _make_sync_call('matcher', 'Subscribe', request, response) except apiproxy_errors.ApplicationError, e: if e.application_error is error_pb.Error.BAD_REQUEST: raise QuerySyntaxError(sub_id, topic, query, e.error_detail) raise e def unsubscribe(document_class, sub_id, topic=None): """Unsubscribe a query. Args: document_class: datastore.Entity or db.Model class. sub_id: subscription id to remove. topic: required for datastore.Entity document_class, optional for db.Model. Topic must be same as used in the subscribe call for this subscription. Raises: DocumentTypeError: document type is unsupported. TopicNotSpecified: raised for datastore.Entity if topic is not specified. apiproxy_errors.Error: unsubscribe call failed. """ from google.appengine.ext import db request = prospective_search_pb.UnsubscribeRequest() if issubclass(document_class, db.Model): topic = _get_document_topic(document_class, topic) elif issubclass(document_class, datastore.Entity): if not topic: raise TopicNotSpecified() else: raise DocumentTypeError() request.set_topic(topic) request.set_sub_id(sub_id) response = prospective_search_pb.UnsubscribeResponse() _make_sync_call('matcher', 'Unsubscribe', request, response) def get_subscription(document_class, sub_id, topic=None): """Get subscription information. Args: document_class: datastore.Entity or db.Model class. sub_id: subscription id to lookup. topic: required for datastore.Entity document_class, optional for db.Model. Returns: Tuple containing: subscription id query expiration time (secs after the Unix epoch when subscription expires) state (SubscriptionState = OK/PENDING/ERROR) error_message (if state is ERROR) Raises: DocumentTypeError: document type is unsupported. SubscriptionDoesNotExist: if subscription of specified id does not exist. TopicNotSpecified: raised for datastore.Entity if topic is not specified. apiproxy_errors.Error: call failed. """ subscriptions = list_subscriptions(document_class, sub_id, topic=topic, max_results=1) if len(subscriptions) and subscriptions[0][0] == sub_id: return subscriptions[0] raise SubscriptionDoesNotExist(topic, sub_id) def list_subscriptions(document_class, sub_id_start='', topic=None, max_results=DEFAULT_LIST_SUBSCRIPTIONS_MAX_RESULTS, expires_before=None): """List subscriptions on a topic. Args: document_class: datastore.Entity or db.Model class. sub_id_start: return only subscriptions that are lexicographically equal or greater than the specified value. topic: required for datastore.Entity document_class, optional for db.Model. max_results: maximum number of subscriptions to return. expires_before: when set, limits list to subscriptions which will expire no later than expires_before (epoch time). Returns: List of subscription tuples. The subscription tuple contains: subscription id query expiration time (secs after the Unix epoch when subscription expires) state (SubscriptionState = OK/PENDING/ERROR) error_message (if state is ERROR) Raises: DocumentTypeError: document type is unsupported. TopicNotSpecified: raised for datastore.Entity if topic is not specified. apiproxy_errors.Error: list call failed. """ from google.appengine.ext import db if issubclass(document_class, db.Model): topic = _get_document_topic(document_class, topic) elif issubclass(document_class, datastore.Entity): if not topic: raise TopicNotSpecified() else: raise DocumentTypeError() return prospective_search_admin.list_subscriptions(topic, max_results, None, sub_id_start, expires_before) def list_topics(max_results=DEFAULT_LIST_TOPICS_MAX_RESULTS, topic_start=None): """List topics. Args: max_results: maximum number of topics to return. topic_start: if not None, start listing topics from this one. Returns: List of topics (strings) """ return prospective_search_admin.list_topics(max_results, topic_start=topic_start) def match(document, topic=None, result_key=None, result_relative_url='/_ah/prospective_search', result_task_queue='default', result_batch_size=DEFAULT_RESULT_BATCH_SIZE, result_return_document=True): """Match document with all subscribed queries on specified topic. Args: document: instance of datastore.Entity or db.Model document. topic: required for datastore.Entity, optional for db.Model. Only subscriptions of this topic will be matched against this document. result_key: key to return in result, potentially to identify document. result_relative_url: url of taskqueue event handler for results. result_task_queue: name of taskqueue queue to put batched results on. result_batch_size: number of subscription ids per taskqueue task batch. result_return_document: returns document with match results if true. Raises: DocumentTypeError: document type is unsupported. TopicNotSpecified: raised for datastore.Entity if topic is not specified. apiproxy_errors.Error: match call failed. """ from google.appengine.ext import db request = prospective_search_pb.MatchRequest() if isinstance(document, db.Model): topic = _get_document_topic(document, topic) doc_pb = db.model_to_protobuf(document) if result_return_document: request.set_result_python_document_class(_doc_class.MODEL) elif isinstance(document, datastore.Entity): topic = _get_document_topic(document, topic) doc_pb = document.ToPb() if result_return_document: request.set_result_python_document_class(_doc_class.ENTITY) else: raise DocumentTypeError() request.set_topic(topic) request.mutable_document().CopyFrom(doc_pb) if result_key: request.set_result_key(result_key) request.set_result_relative_url(result_relative_url) request.set_result_task_queue(result_task_queue) request.set_result_batch_size(result_batch_size) response = prospective_search_pb.MatchResponse() _make_sync_call('matcher', 'Match', request, response) def get_document(request): """Decodes document from prospective_search result POST request. Args: request: received POST request Returns: document: original datastore.Entity or db.Model document from match call. Raises: DocumentTypeError: """ from google.appengine.ext import db doc_class = request.get('python_document_class') if not doc_class: return None entity = entity_pb.EntityProto() entity.ParseFromString(base64.urlsafe_b64decode( request.get('document').encode('utf-8'))) doc_class = int(doc_class) if doc_class is _doc_class.ENTITY: return datastore.Entity('temp-kind').FromPb(entity) elif doc_class is _doc_class.MODEL: return db.model_from_protobuf(entity) else: raise DocumentTypeError()