我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用google.appengine.ext.db.delete()。
def delete(self, tablename, query): sql = self._delete(tablename, query) ### special code to handle CASCADE in SQLite & SpatiaLite db = self.db table = db[tablename] if self.dbengine in ('sqlite', 'spatialite') and table._referenced_by: deleted = [x[table._id.name] for x in db(query).select(table._id)] ### end special code to handle CASCADE in SQLite & SpatiaLite self.execute(sql) try: counter = self.cursor.rowcount except: counter = None ### special code to handle CASCADE in SQLite & SpatiaLite if self.dbengine in ('sqlite', 'spatialite') and counter: for field in table._referenced_by: if field.type=='reference '+table._tablename \ and field.ondelete=='CASCADE': db(field.belongs(deleted)).delete() ### end special code to handle CASCADE in SQLite & SpatiaLite return counter
def delete(self,tablename, query): """ This function was changed on 2010-05-04 because according to http://code.google.com/p/googleappengine/issues/detail?id=3119 GAE no longer supports deleting more than 1000 records. """ # self.db['_lastsql'] = self._delete(tablename,query) (items, tablename, fields) = self.select_raw(query) # items can be one item or a query if not isinstance(items,list): #use a keys_only query to ensure that this runs as a datastore # small operations leftitems = items.fetch(1000, keys_only=True) counter = 0 while len(leftitems): counter += len(leftitems) gae.delete(leftitems) leftitems = items.fetch(1000, keys_only=True) else: counter = len(items) gae.delete(items) return counter
def delete(self,tablename,query): if not isinstance(query,Query): raise SyntaxError("Not Supported") if query.first.type=='id' and query.op==self.EQ: id = query.second tablename = query.first.tablename assert(tablename == query.first.tablename) ctable = self.connection[tablename] try: del ctable[str(id)] return 1 except couchdb.http.ResourceNotFound: return 0 else: tablename = self.get_table(query) rows = self.select(query,[self.db[tablename]._id],{}) ctable = self.connection[tablename] for row in rows: del ctable[str(row.id)] return len(rows)
def delete(self, tablename, query): counter = 0 if query: if use_common_filters(query): query = self.common_filter(query, [tablename,]) result, data = self.connection.select(self.connection.mailbox_names[tablename]) string_query = "(%s)" % query result, data = self.connection.search(None, string_query) store_list = [item.strip() for item in data[0].split() if item.strip().isdigit()] for number in store_list: result, data = self.connection.store(number, "+FLAGS", "(\\Deleted)") if result == "OK": counter += 1 else: raise Exception("IMAP store error: %s" % data) if counter > 0: result, data = self.connection.expunge() return counter
def test_signals(self): global received_pre_delete global received_post_save received_pre_delete = False received_post_save = False def handle_pre_delete(**kwargs): global received_pre_delete received_pre_delete = True signals.pre_delete.connect(handle_pre_delete, sender=TestC) def handle_post_save(**kwargs): global received_post_save received_post_save = True signals.post_save.connect(handle_post_save, sender=TestC) a = TestC() a.put() a.delete() self.assertTrue(received_pre_delete) self.assertTrue(received_post_save)
def test_batch_signals(self): global received_pre_delete global received_post_save received_pre_delete = False received_post_save = False def handle_pre_delete(**kwargs): global received_pre_delete received_pre_delete = True signals.pre_delete.connect(handle_pre_delete, sender=TestC) def handle_post_save(**kwargs): global received_post_save received_post_save = True signals.post_save.connect(handle_post_save, sender=TestC) a = TestC() db.put([a]) db.delete([a]) self.assertTrue(received_pre_delete) self.assertTrue(received_post_save) # Test serialization
def test_cleanup(self): signals.pre_delete.connect(cleanup_relations, sender=TestC) c1 = TestC() c2 = TestC() db.put((c1, c2)) TestModelRel(modelrel=c1).put() child = SigChild(owner=c1, rel=c2) child.put() self.assertEqual(TestC.all().count(), 2) self.assertEqual(SigChild.all().count(), 1) self.assertEqual(TestModelRel.all().count(), 1) c1.delete() signals.pre_delete.disconnect(cleanup_relations, sender=TestC) self.assertEqual(SigChild.all().count(), 0) self.assertEqual(TestC.all().count(), 0) self.assertEqual(TestModelRel.all().count(), 0)
def post(self): """ Submits data to the server to delete the post """ auth_error = True if self.read_secure_cookie('usercookie'): auth_error = False else: auth_error = True username = self.read_secure_cookie('usercookie') if not self.user_exists(username): auth_error = False else: auth_error = True if not auth_error: post_id = self.request.get('post_id') key = db.Key.from_path('Post', int(post_id), parent=blog_key()) # gets the post data based upon what # is passed from post_id into key db.delete(key) self.render('/postdeleted.html') else: self.redirect('/signup')
def fetch_keys(self, keys): cache_key_mapping = dict((self.key_to_cache_key(key), key) for key in keys) object_map = {} max_in_queries = datastore.MAX_ALLOWABLE_QUERIES cache_keys = cache_key_mapping.keys() for i in range(0, len(cache_keys), max_in_queries): objects = FacebookCachedObject.get_by_key_name(cache_keys[i:i + max_in_queries]) for o in objects: if o: # Sometimes objects get created without json_data, so we need to manually purge those from our DB and not pass them on to clients if o.json_data: if o.date_cached > self.oldest_allowed: object_map[cache_key_mapping[o.key().name()]] = o.decode_data() else: o.delete() logging.info("BatchLookup: db lookup objects found: %s", object_map.keys()) return object_map
def delete_backup_files(filesystem, backup_files): if backup_files: if filesystem == FILES_API_BLOBSTORE_FILESYSTEM: blob_keys = [] for fname in backup_files: blob_key = get_blob_key(fname) if blob_key: blob_keys.append(blob_key) if len(blob_keys) == MAX_BLOBS_PER_DELETE: blobstore_api.delete(blob_keys) blob_keys = [] if blob_keys: blobstore_api.delete(blob_keys)
def post(self): """Handler for post requests to datastore_admin/backup_delete.do. Deletes are executed and user is redirected to the base-path handler. """ backup_ids = self.request.get_all('backup_id') token = self.request.get('xsrf_token') params = () if backup_ids and utils.ValidateXsrfToken(token, XSRF_ACTION): try: for backup_info in db.get(backup_ids): if backup_info: delete_backup_info(backup_info) except Exception, e: logging.exception('Failed to delete datastore backup.') params = [('error', e.message)] self.SendRedirect(params=params)
def _cleanup_old_file(self, gcs_file): """Clean up the old version of a file. The old version may or may not be finalized yet. Either way, when user tries to create a file that already exists, we delete the old version first. Args: gcs_file: an instance of _AE_GCSFileInfo_. """ if gcs_file: if gcs_file.finalized: blobkey = gcs_file.key().name() self.blob_storage.DeleteBlob(blobkey) else: db.delete(_AE_GCSPartialFile_.all().ancestor(gcs_file)) gcs_file.delete()
def post(self): """Handle modifying actions and redirect to a GET page.""" if self.request.get('action:flush_memcache'): if memcache.flush_all(): message = 'Cache flushed, all keys dropped.' else: message = 'Flushing the cache failed. Please try again.' self.redirect(self._construct_url(remove=['action:flush_memcache'], add={'message': message})) elif self.request.get('action:delete_entities'): entity_keys = self.request.params.getall('entity_key') db.delete(entity_keys) self.redirect(self._construct_url( remove=['action:delete_entities'], add={'message': '%d entities deleted' % len(entity_keys)})) else: self.error(404)
def get(self, post_id): key = db.Key.from_path('Post', int(post_id), parent=blog_key()) post = db.get(key) post.subject error = '' info='' user_id = self.read_secure_cookie('user_id') if self.read_secure_cookie('user_id') == '': self.redirect("/login") if not post: self.redirect("/") return if post.user_id != user_id: error = 'You are not allowed to delete this post.' else: db.delete(key) info='Post Deleted!' self.render("infopost.html", error = error, info=info)
def post(self, post_id): ''' This block first if user is logged in or not, if logged in then checks the request made for deletion of post is in database, if found it deletes that post. ''' if self.user: key = db.Key.from_path('Post', int(post_id)) self.post = db.get(key) if self.post: comments = Comment.get_comments(post_id) if self.post.created_by_user == self.user.username: db.delete(key) for comment in comments: db.delete(comment) self.redirect("/home") else: self.redirect("/home") else: self.redirect("/home") else: self.redirect("/login")
def _delete_entity(self): """Delete entity from datastore. Attempts to delete using the key_name stored on the object, whether or not the given key is in the datastore. """ if self._is_ndb(): ndb.Key(self._model, self._key_name).delete() else: entity_key = db.Key.from_path(self._model.kind(), self._key_name) db.delete(entity_key)
def locked_delete(self): """Delete Credential from datastore.""" if self._cache: self._cache.delete(self._key_name) self._delete_entity()
def _delete_entity(self): """Delete entity from datastore. Attempts to delete using the key_name stored on the object, whether or not the given key is in the datastore. """ if self._is_ndb(): _NDB_KEY(self._model, self._key_name).delete() else: entity_key = db.Key.from_path(self._model.kind(), self._key_name) db.delete(entity_key)
def _delete(self,tablename, query): return 'delete %s where %s' % (repr(tablename),repr(query))
def truncate(self,table,mode): self.db(self.db._adapter.id_query(table)).delete()
def __delitem__(self, key): if isinstance(key, dict): query = self._build_query(key) if not self._db(query).delete(): raise SyntaxError('No such record: %s' % key) elif not str(key).isdigit() or \ not self._db(self._id == key).delete(): raise SyntaxError('No such record: %s' % key)
def delete(self): db = self.db tablename = db._adapter.get_table(self.query) table = db[tablename] if any(f(self) for f in table._before_delete): return 0 ret = db._adapter.delete(tablename,self.query) ret and [f(self) for f in table._after_delete] return ret
def __call__(self): return self.db(self.db[self.tablename]._id==self.id).delete()
def delete(self): return self._getset().delete()
def test_serializer(self, format='json'): from django.core import serializers created = datetime.now() x = SerializeModel(key_name='blue_key', name='blue', count=4) x.put() SerializeModel(name='green', count=1, created=created).put() data = serializers.serialize(format, SerializeModel.all()) db.delete(SerializeModel.all().fetch(100)) for obj in serializers.deserialize(format, data): obj.save() self.validate_state( ('key.name', 'name', 'count', 'created'), (None, 'green', 1, created), ('blue_key', 'blue', 4, None), )
def cleanup_relations(instance, **kwargs): if getattr(instance, '__handling_delete', False): return rels_seen, to_delete, to_put = get_cleanup_entities(instance) _get_included_cleanup_entities((instance,), rels_seen, to_delete, to_put) for entity in [instance] + to_delete: entity.__handling_delete = True if to_delete: db.delete(to_delete) for entity in [instance] + to_delete: del entity.__handling_delete if to_put: db.put(to_put)
def post(self): """ Submits data to the server to delete the post """ auth_error = True if self.read_secure_cookie('usercookie'): auth_error = False else: auth_error = True username = self.read_secure_cookie('usercookie') if not self.user_exists(username): auth_error = False else: auth_error = True if not auth_error: comment_id = self.request.get('comment_id') post_id = self.request.get('post_id') key = db.Key.from_path('Comment', int(comment_id), parent=post_key(post_id)) # gets the post data based upon what # is passed from post_id into key db.delete(key) self.render('/commentdeleted.html') else: self.redirect('/signup') # GAE APPLICATION VARIABLE # This variable sets the atributes of the individual HTML # files that will be served using google app engine. # '/([0-9]+)' recieves post_id from NewPostHandler class passing it # into PermaPost class via the url using regular expression
