我们从Python开源项目中,提取了以下50个代码示例,用于说明如何使用werkzeug.exceptions.NotFound()。
def safe_join(directory, filename): """Safely join `directory` and `filename`. Example usage:: @app.route('/wiki/<path:filename>') def wiki_page(filename): filename = safe_join(app.config['WIKI_FOLDER'], filename) with open(filename, 'rb') as fd: content = fd.read() # Read and process the file content... :param directory: the base directory. :param filename: the untrusted filename relative to that directory. :raises: :class:`~werkzeug.exceptions.NotFound` if the resulting path would fall out of `directory`. """ filename = posixpath.normpath(filename) for sep in _os_alt_seps: if sep in filename: raise NotFound() if os.path.isabs(filename) or \ filename == '..' or \ filename.startswith('../'): raise NotFound() return os.path.join(directory, filename)
def index(page=1): """Show the list of published posts. Args: page (int): Listing page number to show. """ posts = ( Post.query .filter_by(is_published=True, ghost='') .order_by(Post.timestamp.desc()) .paginate(page, 10, False) ) try: return render_theme('blog/index.html', posts=posts) except NotFound: # Show a 'no posts found' notice instead of a 404 error return render_theme('blog/index.html')
def tagged(tag, page=1): """Display posts tagged with the given tag. Args: tag (str): Tag name. page (int): Listing page number to show. """ posts = ( Post.query .filter(Post.tags.any(name=tag)) .filter_by(is_published=True, ghost='') .order_by(Post.timestamp.desc()) .paginate(page, 10, False) ) try: return render_theme('blog/tagged.html', posts=posts, tag=tag) except NotFound: # Show a 'no posts found' notice instead of a 404 error return render_theme('blog/tagged.html', tag=tag)
def file_index(page=1): """Show available actions regarding files. Args: page (int): Listing page number to show. """ order_key, order_dir, ordering = _sort_uploads(request.args) files = ( FileUpload.query .order_by(ordering) .paginate(page, 20, False) ) try: return render_template( 'akamatsu/dashboard/file/index.html', files=files, order_key=order_key, order_dir=order_dir ) except NotFound: # Show a 'no files found' notice instead of a 404 error return render_template('akamatsu/dashboard/file/index.html')
def user_index(page=1): """Show list of users registered in the application. Args: page (int): Listing page number to show. """ order_key, order_dir, ordering = _sort_users(request.args) users = ( User.query .order_by(ordering) .paginate(page, 20, False) ) try: return render_template( 'akamatsu/dashboard/user/index.html', users=users, order_key=order_key, order_dir=order_dir ) except NotFound: # Show a 'no posts found' notice instead of a 404 error return render_template('akamatsu/dashboard/user/index.html')
def update(tag_id, body): """ Update category """ try: tag = Tag.objects.get(id=tag_id) except (ObjectDoesNotExist, ValueError): raise NotFound('Tag not found') try: body.pop('id', None) existing = Tag.objects.exclude(id=tag.id).values_list('name', flat=True) existing = [tg.lower() for tg in existing] if body['name'].lower().strip() in existing: raise BadRequest('Tag already exists') Tag.objects.filter(pk=tag.pk).update(**body) tag = Tag.objects.get(pk=tag.pk) except (AttributeError, KeyError, FieldError, IntegrityError, ValueError, TypeError): raise BadRequest('Invalid fields in body') return model_to_dict(tag)
def update(news_id, body, user): """ Update news """ try: if user.is_superuser: news = News.objects.get(id=news_id) else: news = News.objects.get(id=news_id, author__id=user.id) except (ObjectDoesNotExist, ValueError): return NotFound('News not found') try: body = {k: v for k, v in body.iteritems() if k not in ['author', 'date', 'tags']} News.objects.filter(pk=news.pk).update(**body) news = News.objects.get(pk=news.pk) except (KeyError, FieldError, IntegrityError): raise BadRequest('Invalid fields in body') return model_to_dict(news)
def delete_from_report(item_id, rep, user): """ Delete an item """ try: item = ReportItem.objects.get(id=item_id) ReportItem.objects.filter(report=rep, rawItem=item.rawItem).delete() report = Report.objects.get(id=rep) if report.ticket: database.log_action_on_ticket( ticket=report.ticket, action='delete_item', user=user ) return {'message': 'Item successfully removed'} except (ObjectDoesNotExist, ValueError): raise NotFound('Item not found')
def get_screenshot(item_id, report_id): """ Get screenshot for item """ try: item = ReportItem.objects.get(id=item_id, report__id=report_id) if item.itemType != 'URL': raise BadRequest('Item is not an URL') except (ObjectDoesNotExist, ValueError): raise NotFound('Item not found') try: screenshots = ImplementationFactory.instance.get_singleton_of( 'PhishingServiceBase' ).get_screenshots(item.rawItem) schema.valid_adapter_response('PhishingServiceBase', 'get_screenshots', screenshots) results = { 'rawItem': item.rawItem, 'screenshots': screenshots, } return results except (PhishingServiceException, schema.InvalidFormatError, schema.SchemaNotFound): raise InternalServerError('Error while loading screenshots')
def unblock_item(item_id, report_id=None, ticket_id=None): """ Unblock given `abuse.models.ReportItem` """ try: item = ReportItem.objects.get(id=item_id) if report_id: report = Report.objects.get(id=report_id) if item.report.id != report.id: raise BadRequest('Given item not attached to given report') if ticket_id: ticket = Ticket.objects.get(id=ticket_id) if item.report.id not in ticket.reportTicket.all().values_list('id', flat=True): raise BadRequest('Given item not attached to given ticket') except (AttributeError, ObjectDoesNotExist, TypeError, ValueError): raise NotFound('Item not found') utils.default_queue.enqueue( 'phishing.unblock_url', url=item.rawItem, ) return {'message': 'Unblocking job successfully scheduled'}
def remove_tag(defendant_id, tag_id, user): """ Remove defendant tag """ try: tag = Tag.objects.get(id=tag_id) defendant = Defendant.objects.get(id=defendant_id) for defendt in Defendant.objects.filter(customerId=defendant.customerId): defendt.tags.remove(tag) defendt.save() for ticket in defendt.ticketDefendant.all(): database.log_action_on_ticket( ticket=ticket, action='remove_tag', user=user, tag_name=tag.name ) except (ObjectDoesNotExist, FieldError, IntegrityError, ValueError): raise NotFound('Defendant or tag not found') return show(defendant_id)
def update(prov, body): """ Update provider infos """ try: provider = Provider.objects.get(email=prov) except (ObjectDoesNotExist, ValueError): raise NotFound('Provider does not exist') try: body = {k: v for k, v in body.iteritems() if k in PROVIDER_FIELDS} cat = None if body.get('defaultCategory'): cat = Category.objects.get(name=body['defaultCategory']) body.pop('defaultCategory', None) Provider.objects.filter(pk=provider.pk).update(defaultCategory=cat, **body) provider = Provider.objects.get(pk=provider.pk) except (FieldError, IntegrityError, ObjectDoesNotExist) as ex: raise BadRequest(str(ex.message)) return model_to_dict(provider)
def remove_tag(provider_email, tag_id): """ Remove defendant tag """ try: tag = Tag.objects.get(id=tag_id) provider = Provider.objects.get(email=provider_email) if provider.__class__.__name__ != tag.tagType: raise BadRequest('Invalid tag for provider') provider.tags.remove(tag) provider.save() except (ObjectDoesNotExist, FieldError, IntegrityError, ValueError): raise NotFound('Provider or tag not found') return model_to_dict(provider)
def _serve_webui(self, file_name='index.html'): # pylint: disable=redefined-builtin try: assert file_name web3 = self.flask_app.config.get('WEB3_ENDPOINT') if web3 and 'config.' in file_name and file_name.endswith('.json'): host = request.headers.get('Host') if any(h in web3 for h in ('localhost', '127.0.0.1')) and host: _, _port = split_endpoint(web3) _host, _ = split_endpoint(host) web3 = 'http://{}:{}'.format(_host, _port) response = jsonify({'raiden': self._api_prefix, 'web3': web3}) else: response = send_from_directory(self.flask_app.config['WEBUI_PATH'], file_name) except (NotFound, AssertionError): response = send_from_directory(self.flask_app.config['WEBUI_PATH'], 'index.html') return response
def get_all_service_attributes(cluster_name, service_name): """ Get all information for a service :param cluster_name: str. name of the cluster that the service is in :param service_name: str. name of service to look up :return: json """ service = {"service_name": service_name, "cluster_name": cluster_name, "tasks": []} try: for task in ecs_api.get_task_ids_from_service(service_name, cluster_name): try: task_json = get_all_container_attributes_by_task_id(task, json=False) service['tasks'].append(task_json) except NotFound as e: logger.warn('ECS API told us about task {} but unable to find in our database'. format(task)) return jsonify(service) except: abort(404, 'ECS service not found')
def show_article(aid): from model import get_article import os a = get_article(aid) def get_css(dom): path = '{}/css/{}.css'.format(app.static_folder, dom) return '{}.css'.format(dom) if os.path.isfile(path) else None if a is not None: return render_template('article.html', article=a, dom_css=get_css(a.domain)) else: raise NotFound('article not existed')
def test_app(): # app's config should be loaded from instance/config.py assert app.config['STORAGE_TYPE'] == 'file' assert app.config['THEME'] == 'test' with app.test_request_context('/'): app.send_static_file('no-use.css') app.send_static_file('no-use-2.css') with raises(NotFound): app.send_static_file('non-exists.css') origin_mode = app.config['MODE'] app.config['MODE'] = 'api-only' with app.test_request_context('/'): with raises(NotFound): app.send_static_file('no-use.css') app.config['MODE'] = origin_mode
def send_static_file(self, filename): """ Send static files from the static folder in the current selected theme prior to the global static folder. :param filename: static filename :return: response object """ if self.config['MODE'] == 'api-only': abort(404) # if 'api-only' mode is set, we should not send static files theme_static_folder = getattr(self, 'theme_static_folder', None) if theme_static_folder: try: return send_from_directory(theme_static_folder, filename) except NotFound: pass return super(CustomFlask, self).send_static_file(filename)
def entity(entity_id): entity = load_entity(entity_id, request.auth) if entity is None: raise NotFound() # load possible duplicates via fingerprint expansion # fingerprints = expand_fingerprints(entity.fingerprints, request.auth) query = Query(request.args, prefix='duplicates_', path=request.path, limit=5) # query.add_facet('countries', 'Countries', country_label) duplicates = search_duplicates(entity.id, entity.fingerprints, query, request.auth) # load links query = Query(request.args, prefix='links_', path=request.path, limit=10) query.add_facet('schemata', 'Types', link_schema_label) query.add_facet('remote.countries', 'Countries', country) links = search_links(entity, query, request.auth) return render_template("entity.html", entity=entity, links=links, duplicates=duplicates)
def test_get(self): """Test get method""" app = self._test_application() mockDBObject.return_value = app with patch.object(reqparse.RequestParser, 'parse_args'): resource = RestApplication(restapi=self.restapi, server=self.server) # Fail to find a device: raises 404 NotFound with patch.object(Application, 'find', classmethod(mockDBObject.findFail)): yield self.assertFailure(resource.get(app.appeui), e.NotFound) # Find a device success returns a dict of field values with patch.object(Application, 'find', classmethod(mockDBObject.findSuccess)): resource.getProperties = MagicMock() result = yield resource.get(app.appeui) self.assertEqual(app.appeui, result['appeui'])
def test_put(self): """Test put method""" app = self._test_application() mockDBObject.return_value = app with patch.object(reqparse.RequestParser, 'parse_args'): resource = RestApplication(restapi=self.restapi, server=self.server) # Fail to find a device: raises 404 NotFound with patch.object(Application, 'find', classmethod(mockDBObject.findFail)): yield self.assertFailure(resource.put(app.appeui), e.NotFound) # Find a device, device fails validity check: raises 400 BadRequest with patch.object(Application, 'find', classmethod(mockDBObject.findSuccess)): app.valid = MagicMock(return_value=(False, {})) yield self.assertFailure(resource.put(app.appeui), e.BadRequest) # Pass validity check, returns 200 expected = ({}, 200) app.valid = MagicMock(return_value=(True, {})) app.update = MagicMock() result = yield resource.put(app.appeui) self.assertEqual(expected, result)
def test_delete(self): """Test delete method""" app = self._test_application() mockDBObject.return_value = app with patch.object(reqparse.RequestParser, 'parse_args'): resource = RestApplication(restapi=self.restapi, server=self.server) # Device exists with AppEUI: raises 400 error with patch.object(Device, 'find', classmethod(mockDBObject.findSuccess)): yield self.assertFailure(resource.delete(app.appeui), e.BadRequest) # Fail to find the application: raises 404 NotFound with patch.object(Device, 'find', classmethod(mockDBObject.findFail)), \ patch.object(Application, 'find', classmethod(mockDBObject.findFail)): yield self.assertFailure(resource.delete(app.appeui), e.NotFound) # Find and delete, returns 200 with patch.object(Device, 'find', classmethod(mockDBObject.findFail)), \ patch.object(Application, 'find', classmethod(mockDBObject.findSuccess)): expected = ({}, 200) app.delete = MagicMock() result = yield resource.delete(app.appeui) self.assertEqual(expected, result)
def test_get(self): """Test get method""" device = self._test_device() mockDBObject.return_value = device with patch.object(reqparse.RequestParser, 'parse_args'): resource = RestDevice(restapi=self.restapi, server=self.server) # Fail to find a device: raises 404 NotFound with patch.object(Device, 'find', classmethod(mockDBObject.findFail)): yield self.assertFailure(resource.get(device.deveui), e.NotFound) # Find a device success returns a dict of field values with patch.object(Device, 'find', classmethod(mockDBObject.findSuccess)): result = yield resource.get(device.deveui) self.assertEqual(device.deveui, result['deveui'])
def test_put(self): """Test put method""" device = self._test_device() mockDBObject.return_value = device with patch.object(reqparse.RequestParser, 'parse_args'): resource = RestDevice(restapi=self.restapi, server=self.server) # Fail to find a device: raises 404 NotFound with patch.object(Device, 'find', classmethod(mockDBObject.findFail)): yield self.assertFailure(resource.put(device.deveui), e.NotFound) # Find a device, device fails validity check: raises 400 BadRequest with patch.object(Device, 'find', classmethod(mockDBObject.findSuccess)): device.valid = MagicMock(return_value=(False, {})) yield self.assertFailure(resource.put(device.deveui), e.BadRequest) # Pass validity check, returns 200 expected = ({}, 200) device.valid = MagicMock(return_value=(True, {})) device.update = MagicMock() result = yield resource.put(device.deveui) self.assertEqual(expected, result)
def test_delete(self): """Test delete method""" app = self._test_application() prop = self._test_appproperty() args = {'port': 11} with patch.object(reqparse.RequestParser, 'parse_args', MagicMock(return_value=args)): resource = RestAppProperty(restapi=self.restapi, server=self.server) mockDBObject.return_value = prop # Fail to find the application: raises 404 NotFound with patch.object(Application, 'find', classmethod(mockDBObject.findFail)): yield self.assertFailure(resource.delete(app.appeui), e.NotFound) # Find and delete, returns 200 with patch.object(Application, 'find', classmethod(mockDBObject.findSuccess)), \ patch.object(AppProperty, 'find', classmethod(mockDBObject.findSuccess)): expected = ({}, 200) prop.delete = MagicMock() result = yield resource.delete(app.appeui) self.assertEqual(expected, result)
def test_get(self): """Test get method""" gateway = self._test_gateway() mockDBObject.return_value = gateway with patch.object(reqparse.RequestParser, 'parse_args'): resource = RestGateway(restapi=self.restapi, server=self.server) # Fail to find a gateway: raises 404 NotFound with patch.object(Gateway, 'find', classmethod(mockDBObject.findFail)): yield self.assertFailure(resource.get(gateway.host), e.NotFound) # Find a device success returns a dict of field values with patch.object(Gateway, 'find', classmethod(mockDBObject.findSuccess)): result = yield resource.get(gateway.host) self.assertEqual(gateway.host, result['host'])
def test_put(self): """Test put method""" gateway = self._test_gateway() mockDBObject.return_value = gateway with patch.object(reqparse.RequestParser, 'parse_args'): resource = RestGateway(restapi=self.restapi, server=self.server) # Fail to find the gateway: raises 404 NotFound with patch.object(Gateway, 'find', classmethod(mockDBObject.findFail)): yield self.assertFailure(resource.put(gateway.host), e.NotFound) # Find a device, device fails validity check: raises 400 BadRequest with patch.object(Gateway, 'find', classmethod(mockDBObject.findSuccess)): gateway.valid = MagicMock(return_value=(False, {})) yield self.assertFailure(resource.put(gateway.host), e.BadRequest) # Pass validity check, returns 200 expected = ({}, 200) gateway.valid = MagicMock(return_value=(True, {})) gateway.update = MagicMock() result = yield resource.put(gateway.host) self.assertEqual(expected, result)
def test_get(self): """Test get method""" interface = self._test_azureiothttps() appif = self._test_appinterface() interface.appinterface = appif interfaceManager.interfaces = [interface] with patch.object(reqparse.RequestParser, 'parse_args'): resource = RestAppInterface(restapi=self.restapi, server=self.server) # Fail to find the app interface: raises 404 NotFound interfaceManager.getInterface = MagicMock(return_value=None) yield self.assertFailure(resource.get(1), e.NotFound) # Success finding the interface returns a dict of field values interfaceManager.getInterface = MagicMock(return_value=interface) result = yield resource.get(appif.id) self.assertEqual(interface.name, result['name'])
def test_delete(self): """Test delete method""" interface = self._test_azureiothttps() appif = self._test_appinterface() interface.appinterface = appif interfaceManager.interfaces = [interface] with patch.object(reqparse.RequestParser, 'parse_args'): resource = RestAppInterface(restapi=self.restapi, server=self.server) # Fail to find the interface: raises 404 NotFound interfaceManager.getInterface = MagicMock(return_value=None) yield self.assertFailure(resource.delete(1), e.NotFound) # Find and delete, returns 200 with patch.object(AzureIotHttps, 'exists', MagicMock(return_value=True)), \ patch.object(HasMany, 'get', MagicMock(return_value=[appif])), \ patch.object(AzureIotHttps, 'delete'), \ patch.object(AppInterface, 'delete'): interfaceManager.getInterface = MagicMock(return_value=interface) expected = ({}, 200) result = yield resource.delete(1) self.assertEqual(expected, result)