我们从Python开源项目中，提取了以下33个代码示例，用于说明如何使用twisted.web.resource.IResource()。
def render(self, request): """ Render a given resource. See L{IResource}'s render method. I delegate to methods of self with the form 'render_METHOD' where METHOD is the HTTP that was used to make the request. Examples: render_GET, render_HEAD, render_POST, and so on. Generally you should implement those methods instead of overriding this one. render_METHOD methods are expected to return a string which will be the rendered page, unless the return value is twisted.web.server.NOT_DONE_YET, in which case it is this class's responsibility to write the results to request.write(data), then call request.finish(). Old code that overrides render() directly is likewise expected to return a string or NOT_DONE_YET. """ if request.code not in coap.requests: raise error.UnsupportedMethod() m = getattr(self, 'render_' + coap.requests[request.code], None) if not m: raise error.UnallowedMethod() return m(request)
def test_getChildWithDefaultAuthorized(self): """ Resource traversal which encounters an L{HTTPAuthSessionWrapper} results in an L{IResource} which renders the L{IResource} avatar retrieved from the portal when the request has a valid I{Authorization} header. """ self.credentialFactories.append(BasicCredentialFactory('example.com')) request = self.makeRequest([self.childName]) child = self._authorizedBasicLogin(request) d = request.notifyFinish() def cbFinished(ignored): self.assertEqual(request.written, [self.childContent]) d.addCallback(cbFinished) request.render(child) return d
def test_renderAuthorized(self): """ Resource traversal which terminates at an L{HTTPAuthSessionWrapper} and includes correct authentication headers results in the L{IResource} avatar (not one of its children) retrieved from the portal being rendered. """ self.credentialFactories.append(BasicCredentialFactory('example.com')) # Request it exactly, not any of its children. request = self.makeRequest([]) child = self._authorizedBasicLogin(request) d = request.notifyFinish() def cbFinished(ignored): self.assertEqual(request.written, [self.avatarContent]) d.addCallback(cbFinished) request.render(child) return d
def _authorizedResource(self, request): """ Get the L{IResource} which the given request is authorized to receive. If the proper authorization headers are present, the resource will be requested from the portal. If not, an anonymous login attempt will be made. """ authheader = request.getHeader(b'authorization') if not authheader: return util.DeferredResource(self._login(Anonymous())) factory, respString = self._selectParseHeader(authheader) if factory is None: return UnauthorizedResource(self._credentialFactories) try: credentials = factory.decode(respString, request) except error.LoginFailed: return UnauthorizedResource(self._credentialFactories) except: log.err(None, "Unexpected failure from credentials factory") return ErrorPage(500, None, None) else: return util.DeferredResource(self._login(credentials))
def requestAvatar(self, avatarId, mind, *interfaces): if resource.IResource not in interfaces: raise NotImplementedError("no interface") if avatarId: return (resource.IResource, MarkAuthenticatedResource(self.resource, avatarId), lambda:None) else: return resource.IResource, self.nonauthenticated, lambda:None
def __init__(self, portal, callback=None, errback=None): """Constructs a UsernamePasswordWrapper around the given portal. @param portal: A cred portal for your web application. The checkers associated with this portal must be able to accept username/password credentials. @type portal: L{twisted.cred.portal.Portal} @param callback: Gets called after a successful login attempt. A resource that redirects to "." will display the avatar resource. If this parameter isn't provided, defaults to a standard Woven "Thank You" page. @type callback: A callable that accepts a Woven L{model<twisted.web.woven.interfaces.IModel>} and returns a L{IResource<twisted.web.resource.Resource>}. @param errback: Gets called after a failed login attempt. If this parameter is not provided, defaults to a the standard Woven form error (i.e. The original form on a page of its own, with errors noted.) @type errback: A callable that accepts a Woven L{model<twisted.web.woven.interfaces.IModel>} and returns a L{IResource<twisted.web.resource.Resource>}. """ Resource.__init__(self) self.portal = portal self.callback = callback self.errback = errback
def registerControllerForModel(controller, model): """ Registers `controller' as an adapter of `model' for IController, and optionally registers it for IResource, if it implements it. @param controller: A class that implements L{interfaces.IController}, usually a L{Controller} subclass. Optionally it can implement L{resource.IResource}. @param model: Any class, but probably a L{twisted.web.woven.model.Model} subclass. """ components.registerAdapter(controller, model, interfaces.IController) if resource.IResource.implementedBy(controller): components.registerAdapter(controller, model, resource.IResource)
def getChild(self, path, request): if path == '': path = 'index' path = path.replace(".","_") cm = getattr(self, "wchild_"+path, None) if cm: p = cm(request) if isinstance(p, Deferred): return util.DeferredResource(p) adapter = IResource(p, None) if adapter is not None: return adapter # maybe we want direct support for ModelLoader? # cl = getattr(self, "wload_"+path, None) #??? return Resource.getChild(self, path, request)
def getChild(self, path, request): """See twisted.web.Resource.getChild. """ self.restat() if not self.isdir(): return self.childNotFound if path: fpath = self.child(path) else: fpath = self.childSearchPreauth(*self.indexNames) if fpath is None: return self.directoryListing() if not fpath.exists(): fpath = fpath.siblingExtensionSearch(*self.ignoredExts) if fpath is None: return self.childNotFound if platformType == "win32": # don't want .RPY to be different than .rpy, since that would allow # source disclosure. processor = InsensitiveDict(self.processors).get(fpath.splitext()[1]) else: processor = self.processors.get(fpath.splitext()[1]) if processor: return resource.IResource(processor(fpath.path, self.registry)) return self.createSimilarFile(fpath.path) # methods to allow subclasses to e.g. decrypt files on the fly:
def requestAvatar(self, user, mind, *interfaces): # the resource is passed on regardless of user if IResource in interfaces: return (IResource, self.resource, lambda: None) raise NotImplementedError()
def getChild(self, path, request): s = request.getSession() if s is None: return request.setupSession() if path == INIT_PERSPECTIVE: def loginSuccess(result): interface, avatarAspect, logout = result s.setResourceForPortal(avatarAspect, self.portal, logout) def triggerLogin(username, password, submit=None): return self.portal.login( UsernamePassword(username, password), None, IResource ).addCallback( loginSuccess ).addErrback( self._ebFilter ) return form.FormProcessor( newLoginSignature.method( triggerLogin ), callback=self.callback, errback=self.errback ) elif path == DESTROY_PERSPECTIVE: s.portalLogout(self.portal) return Redirect(".") else: r = s.resourceForPortal(self.portal) if r: ## Delegate our getChild to the resource our portal says is the right one. return getResource(r[0], path, request) else: return DeferredResource( self.portal.login(Anonymous(), None, IResource ).addCallback( lambda (interface, avatarAspect, logout): getResource(s.setResourceForPortal(avatarAspect, self.portal, logout), path, request)))
def getChildWithDefault(self, name, request): """ Reject attempts to retrieve a child resource. All path segments beyond the one which refers to this resource are handled by the WebSocket connection. @type name: C{bytes} @param name: A single path component from a requested URL. @type request: L{twisted.web.iweb.IRequest} provider @param request: The request received. """ raise RuntimeError( "Cannot get IResource children from WebSocketsResource")
def putChild(self, path, child): """ Reject attempts to add a child resource to this resource. The WebSocket connection handles all path segments beneath this resource, so L{IResource} children can never be found. @type path: C{bytes} @param path: A single path component. @type child: L{IResource} provider @param child: A resource to put underneat this one. """ raise RuntimeError( "Cannot put IResource children under WebSocketsResource")
def requestAvatar(self, avatarId, mind, *interfaces): if IResource in interfaces: return IResource, RootResource( self.facadeClient, session=avatarId), lambda: None raise NotImplementedError()
def requestAvatar(self, avatarId, mind, *interfaces): if IResource in interfaces: if avatarId == checkers.ANONYMOUS: return (IResource, TestHTTPUser(self.template, 'anonymous'), lambda: None) return (IResource, TestHTTPUser(self.template, avatarId), lambda: None) raise NotImplementedError("Only IResource interface is supported")
def requestAvatar(self, avatarId, mind, *interfaces): if IResource in interfaces: self.loggedIn += 1 return IResource, self.avatarFactory(avatarId), self.logout raise NotImplementedError()
def _authorizedBasicLogin(self, request): """ Add an I{basic authorization} header to the given request and then dispatch it, starting from C{self.wrapper} and returning the resulting L{IResource}. """ authorization = b64encode(self.username + b':' + self.password) request.requestHeaders.addRawHeader(b'authorization', b'Basic ' + authorization) return getChildForRequest(self.wrapper, request)
def test_interface(self): """ L{UserDirectory} instances provide L{resource.IResource}. """ self.assertTrue(verifyObject(resource.IResource, self.directory))
def test_unsupported(self): """ A L{WSGIResource} cannot have L{IResource} children. Its C{getChildWithDefault} and C{putChild} methods raise L{RuntimeError}. """ self.assertRaises( RuntimeError, self.resource.getChildWithDefault, b"foo", Request(DummyChannel(), False)) self.assertRaises( RuntimeError, self.resource.putChild, b"foo", Resource())
def getChild(self, path, request): """ If this L{File}'s path refers to a directory, return a L{File} referring to the file named C{path} in that directory. If C{path} is the empty string, return a L{DirectoryLister} instead. """ self.restat(reraise=False) if not self.isdir(): return self.childNotFound if path: try: fpath = self.child(path) except filepath.InsecurePath: return self.childNotFound else: fpath = self.childSearchPreauth(*self.indexNames) if fpath is None: return self.directoryListing() if not fpath.exists(): fpath = fpath.siblingExtensionSearch(*self.ignoredExts) if fpath is None: return self.childNotFound if platformType == "win32": # don't want .RPY to be different than .rpy, since that would allow # source disclosure. processor = InsensitiveDict(self.processors).get(fpath.splitext()[1]) else: processor = self.processors.get(fpath.splitext()[1]) if processor: return resource.IResource(processor(fpath.path, self.registry)) return self.createSimilarFile(fpath.path) # methods to allow subclasses to e.g. decrypt files on the fly:
def getChildWithDefault(self, name, request): """ Reject attempts to retrieve a child resource. All path segments beyond the one which refers to this resource are handled by the WSGI application object. """ raise RuntimeError("Cannot get IResource children from WSGIResource")
def putChild(self, path, child): """ Reject attempts to add a child resource to this resource. The WSGI application object handles all path segments beneath this resource, so L{IResource} children can never be found. """ raise RuntimeError("Cannot put IResource children under WSGIResource")
def _login(self, credentials): """ Get the L{IResource} avatar for the given credentials. @return: A L{Deferred} which will be called back with an L{IResource} avatar or which will errback if authentication fails. """ d = self._portal.login(credentials, None, IResource) d.addCallbacks(self._loginSucceeded, self._loginFailed) return d
def _loginSucceeded(self, args): """ Handle login success by wrapping the resulting L{IResource} avatar so that the C{logout} callback will be invoked when rendering is complete. """ interface, avatar, logout = args class ResourceWrapper(proxyForInterface(IResource, 'resource')): """ Wrap an L{IResource} so that whenever it or a child of it completes rendering, the cred logout hook will be invoked. An assumption is made here that exactly one L{IResource} from among C{avatar} and all of its children will be rendered. If more than one is rendered, C{logout} will be invoked multiple times and probably earlier than desired. """ def getChildWithDefault(self, name, request): """ Pass through the lookup to the wrapped resource, wrapping the result in L{ResourceWrapper} to ensure C{logout} is called when rendering of the child is complete. """ return ResourceWrapper(self.resource.getChildWithDefault(name, request)) def render(self, request): """ Hook into response generation so that when rendering has finished completely (with or without error), C{logout} is called. """ request.notifyFinish().addBoth(lambda ign: logout()) return super(ResourceWrapper, self).render(request) return ResourceWrapper(avatar)
def requestAvatar(self, credentials, mind, *interfaces): if IResource in interfaces: return (IResource, self.resource, lambda: None) raise NotImplementedError()
