我们从Python开源项目中,提取了以下17个代码示例,用于说明如何使用twisted.web.resource.NoResource()。
def getChild(self, name, request): """ Dispatch a specific incoming request to an appropriate resource """ # First try: static resource static = self.static.handle_static(request) if static: return static # If that doesn't serve the request, try the plugin dynamic path if request.path in self.path_map: print 'using plugin %s for %s' % (self.path_map[request.path], request.path) cfg = self.cfg.copy() cfg['tab_map'] = self.tab_map for arg in request.args: if arg not in cfg: cfg[arg] = request.args[arg][0] # Augment the request with our own session data request.sdata = session.get_data(request) return self.path_map[request.path](cfg) # Nothing else to try print 'Failed to match path', request.path, 'to any plugins', self.path_map return NoResource()
def getChild(self, name, request): out = Resource.getChild(self, name, request) LOGGER.warning("%s - %s", request.path, out.__class__.__name__) if isinstance(out, NoResource): # this matches the pre-existing behavior # of accepting any post as a message forward components = request.path.split('/') while components and components[0] == '': components.pop(0) LOGGER.warning("%s - %d", components, len(components)) if len(components) > 0: out = ForwardPage(self.validator, components[0]) return out
def getChildWithDefault(self, name, request): """ This resource cannot have children, hence this will always fail. """ return NoResource("No such child resource.")
def getChild(self, path, request): fn = os.path.join(self.path, path) if os.path.isdir(fn): return ResourceScriptDirectory(fn, self.registry) if os.path.exists(fn): return ResourceScript(fn, self.registry) return resource.NoResource()
def render(self, request): return resource.NoResource().render(request)
def render(self, request): """ Render me to a web client. Load my file, execute it in a special namespace (with 'request' and '__file__' global vars) and finish the request. Output to the web-page will NOT be handled with print - standard output goes to the log - but with request.write. """ request.setHeader(b"x-powered-by", networkString("Twisted/%s" % copyright.version)) namespace = {'request': request, '__file__': _coerceToFilesystemEncoding("", self.filename), 'registry': self.registry} try: execfile(self.filename, namespace, namespace) except IOError as e: if e.errno == 2: #file not found request.setResponseCode(http.NOT_FOUND) request.write(resource.NoResource("File not found.").render(request)) except: io = NativeStringIO() traceback.print_exc(file=io) output = util._PRE(io.getvalue()) if _PY3: output = output.encode("utf8") request.write(output) request.finish() return server.NOT_DONE_YET
def _getResourceForRequest(self, request): """(Internal) Get the appropriate resource for the given host. """ hostHeader = request.getHeader(b'host') if hostHeader == None: return self.default or resource.NoResource() else: host = hostHeader.lower().split(b':', 1)[0] return (self.hosts.get(host, self.default) or resource.NoResource("host %s not in vhost map" % repr(host)))
def test_getChild(self): """ L{_HostResource.getChild} returns the proper I{Resource} for the vhost embedded in the URL. Verify that returning the proper I{Resource} required changing the I{Host} in the header. """ bazroot = Data(b'root data', "") bazuri = Data(b'uri data', "") baztest = Data(b'test data', "") bazuri.putChild(b'test', baztest) bazroot.putChild(b'uri', bazuri) hr = _HostResource() root = NameVirtualHost() root.default = Data(b'default data', "") root.addHost(b'baz.com', bazroot) request = DummyRequest([b'uri', b'test']) request.prepath = [b'bar', b'http', b'baz.com'] request.site = Site(root) request.isSecure = lambda: False request.host = b'' step = hr.getChild(b'baz.com', request) # Consumes rest of path self.assertIsInstance(step, Data) request = DummyRequest([b'uri', b'test']) step = root.getChild(b'uri', request) self.assertIsInstance(step, NoResource)
def getChild(self, name, request): if name == '': return self td = '.twistd' if name[-len(td):] == td: username = name[:-len(td)] sub = 1 else: username = name sub = 0 try: pw_name, pw_passwd, pw_uid, pw_gid, pw_gecos, pw_dir, pw_shell \ = self._pwd.getpwnam(username) except KeyError: return resource.NoResource() if sub: twistdsock = os.path.join(pw_dir, self.userSocketName) rs = ResourceSubscription('unix',twistdsock) self.putChild(name, rs) return rs else: path = os.path.join(pw_dir, self.userDirName) if not os.path.exists(path): return resource.NoResource() return static.File(path)
def getChild(self, path, request): fnp = self.child(path) if not fnp.exists(): return static.File.childNotFound elif fnp.isdir(): return CGIDirectory(fnp.path) else: return CGIScript(fnp.path) return resource.NoResource()
def render(self, request): notFound = resource.NoResource( "CGI directories do not support directory listing.") return notFound.render(request)
def test_getChild_unknown_transport(self): '''A request with a non-iframe path and an unknown transport results in a 404. ''' resrc = self.txdarn.getChild(b'unknown-transport', self.request) self.assertIsInstance(resrc, NoResource)
def getChild(self, path, request): if self.MATCH_IFRAME.match(path): return self.iframe try: request.postpath.insert(0, path) return self.transports[request.postpath[-1]] except KeyError: return resource.NoResource()
def getChild(self, name, request): if name == b'rpc': return self.rpc else: return NoResource()
def __init__(self, resource=None, *args, **kwargs): # type: (Resource, *Any, **Any) -> None """ :param resource: Root resource for Twisted's standard Site implementation. Pass a resource if you want to fall back to Twisted's default resource lookup mechanism in case no route matches. If None is passed, defaults to a NoResource() instance. """ resource = resource or NoResource() Site.__init__(self, resource, *args, **kwargs) self.routes = {} # type: Dict[Pattern, Resource]
def render(self, request): if 'dir' in request.args: dir = unquote(request.args['dir'][0]) elif 'root' in request.args: dir = unquote(request.args['root'][0]) else: dir = '' if 'file' in request.args: filename = unquote(request.args["file"][0]) path = dir + filename #dirty backwards compatibility hack if not os_path.exists(path): path = resolveFilename(SCOPE_HDD, filename) print "[WebChilds.FileStreamer] path is %s" %path if os_path.exists(path): basename = filename.decode('utf-8', 'ignore').encode('ascii', 'ignore') if '/' in basename: basename = basename.split('/')[-1] request.setHeader("content-disposition", "attachment;filename=\"%s\"" % (basename)) file = static.File(path, defaultType = "application/octet-stream") return file.render_GET(request) else: return resource.NoResource(message="file '%s' was not found" %(dir + filename)).render(request) else: return resource.NoResource(message="no file given with file={filename}").render(request) return server.NOT_DONE_YET
def test_route_site(): root = Resource() root.putChild(b"default", HotelPage()) path_site = retwist.RouteSite(root) # Add a path with named parameters (get stored as dictionary) path_site.addRoute(r"/hotels/(?P<hotel_id>.*)/info", HotelPage()) # Test that we get the correct request object request = MyDummyRequest([b"hotels", b"1234", b"info"]) resource = path_site.getResourceFor(request) assert isinstance(resource, HotelPage) assert request.path_args == { "hotel_id": "1234" } # Test that request rendering receives tha arguments correctly yield _render(resource, request) response_str = b"".join(request.written) assert response_str == b"1234" # ... now let's add a path with unnamed parameters, which are passed as a tuple path_site.addRoute(r"/restaurants/(.*)/info", RestaurantPage()) request = MyDummyRequest([b"restaurants", b"5678", b"info"]) resource = path_site.getResourceFor(request) assert isinstance(resource, RestaurantPage) assert request.path_args == ("5678",) # Again, test that rendering works as expected yield _render(resource, request) response_str = b"".join(request.written) assert response_str == b"5678" # Test the fallback to regular Twisted path resolution request = MyDummyRequest([b"default"]) resource = path_site.getResourceFor(request) assert isinstance(resource, HotelPage) # Test 404 request = MyDummyRequest([b"some", b"nonexistent", b"path"]) resource = path_site.getResourceFor(request) assert isinstance(resource, NoResource)