Source code for gavo.formal.nevowc

Some classes adapting twisted.web (t.w) to make it "basically" work with
twisted templates and rend.Pages.

See porting guide in the README.

#c Copyright 2008-2023, the GAVO project <>
#c This program is free software, covered by the GNU GPL.  See the
#c COPYING file in the source distribution.

import io
import itertools
import sys

from xml.sax import make_parser, handler

from twisted.internet import defer
from twisted.python.filepath import FilePath
from twisted.web import resource
from twisted.web import server
from twisted.web import template

from .twistedpatch import Raw, _ToStan, _NSContext, Tag



# ELEMENT_MAPPING is only consulted if the namespace is in
# NAMESPACE_MAPPING, so we can be a bit cavalier here.
    "invisible": "transparent",

[docs]class NoDataError(Exception): """is raised when no data can be found for a tag during flattening. """ pass
[docs]class MyToStan(_ToStan): """A SAX parser unifying nevow and twisted.web namespaces. We also map invisible (nevow) to transparent (t.w). """ def __init__(self, sourceFilename): _ToStan.__init__(self, sourceFilename) self.prefixMap = _NSContext() self.prefixMap[NEVOW_NS] = "nevow" self.prefixMap[TWISTED_NS] = "nevow"
[docs] def startPrefixMapping(self, prefix, uri): # overridden to swallow attempts to map prefixes for our # standard namespaces if uri==NEVOW_NS or uri==TWISTED_NS: self.prefixMap = _NSContext(self.prefixMap) return else: if prefix=="nevow": raise Exception("You may not bind the nevow prefix" " to a non-twisted or non-nevow URI") _ToStan.startPrefixMapping(self, prefix, uri)
[docs] def startElementNS(self, namespaceAndName, qName, attrs): # regrettably, we also need to map attribute namespaces; # alternative: replace parent's startElementNS entirely. Hm. for attrNS, attrName in list(attrs.keys()): if attrNS in NAMESPACE_MAPPING: attrs._attrs[(NAMESPACE_MAPPING[attrNS], attrName) ] = attrs._attrs.pop((attrNS, attrName)) ns, name = namespaceAndName if ns in NAMESPACE_MAPPING: ns = NAMESPACE_MAPPING[ns] name = ELEMENT_MAPPING.get(name, name) # twisted web blindly discards attributes when constructing # n:attr; we need n:data, at least, so I need to copy and fix # the code if ns==TWISTED_NS and name=='attr': if not self.stack or (None, 'name') not in attrs: raise AssertionError( '<attr> usage error') fixedAttrs = {} for (ns, attName), val in list(attrs.items()): if ns: fixedAttrs["%s:%s"%( self.prefixMap[ns], attName)] = val else: fixedAttrs[attName] = val el = Tag('', render=attrs.get((TWISTED_NS, 'render')), attributes=fixedAttrs, filename=self.sourceFilename, lineNumber=self.locator.getLineNumber()) self.stack[-1].attributes[attrs[None, 'name']] = el self.stack.append(el) self.current = el.children return if ns==TWISTED_NS and name=='invisible': name = 'transparent' return _ToStan.startElementNS(self, (ns, name), qName, attrs)
def _flatsaxParse(fl): """A copy of t.w.template._flatsaxParse that lets me use my own _ToStan """ parser = make_parser() parser.setFeature(handler.feature_validation, 0) parser.setFeature(handler.feature_namespaces, 1) parser.setFeature(handler.feature_external_ges, 0) parser.setFeature(handler.feature_external_pes, 0) s = MyToStan(getattr(fl, "name", None)) parser.setContentHandler(s) parser.setEntityResolver(s) parser.setProperty(handler.property_lexical_handler, s) parser.parse(fl) return s.document
[docs]class XMLFile(template.XMLFile): """a t.w.template.XMLFile able to load nevow templates some extent; we accept both nevow and twisted.web namespace We also unify namespaces prefix on attributes for both to nevow: for simpler handling in later processing. Yes, this will break if someone binds nevow: to some other namespace. Don't do that, then. """ def _loadDoc(self): # overridden to inject my _flatsaxParse; otherwise, it's a copy # of template.XMLFile.loadDoc if not isinstance(self._path, FilePath): return _flatsaxParse(self._path) else: with'r') as f: return _flatsaxParse(f)
[docs]class XMLString(template.XMLString): """as XMLFile, just for strings. Again, we override to let us pass in our own parser. """ def __init__(self, s): if isinstance(s, str): s = s.encode('utf8') self._loadedTemplate = _flatsaxParse(io.BytesIO(s))
[docs]class Passthrough(Raw): """a stan-like tag that inserts its content literally into the target document. No escaping will we done, so if what you pass in is not XML, you'll have a malformed result. This is tags.xml from nevow born again; hence, use it as T.xml from template.tags. """ def __init__(self, content): if isinstance(content, str): content = content.encode("utf-8") if not isinstance(content, bytes): raise Exception("xml content must be a byte string, not %s"% content) self.content = content
[docs] def getContent(self): return self.content
template.tags.xml = Passthrough
[docs]def iterChildren(tag, stopRenders=frozenset(["sequence", "mapping"]), stopAtData=False): """yields the Tag-typed descendents of tag preorder. stopRenderer is a set of renderer names at which traversal does not recurse; this is generally where data "changes" externally for most of our use cases here. """ # somewhat sadly, n:attr ends up in attributes. So, we need to # iterate over the attribute values, too. Sigh. for t in itertools.chain( tag.children, iter(tag.attributes.values())): if isinstance(t, template.Tag): yield t if (t.render is None or t.render not in stopRenders ) and not (stopAtData and "nevow:data" in t.attributes): for c in iterChildren(t, stopRenders, stopAtData): yield c
[docs]def locatePatterns(tag, searchPatterns=[], stopRenders=["sequence", "mapping"]): """returns all descendents of tags for which nevow:pattern is in searchPatterns. The return value is a dictionary mapping each item in searchPatterns to the pattern name. This recursively traverses the children of tag, but recursion will stop when tags have nevow:render attributes in stopRenders. """ searchPatterns, stopRenders = set(searchPatterns), set(stopRenders) res = dict((p, []) for p in searchPatterns) for t in iterChildren(tag, stopRenders): attrs = t.attributes if attrs.get("nevow:pattern") in searchPatterns: pat = t.clone() res[attrs["nevow:pattern"]].append(pat) del pat.attributes["nevow:pattern"] return res
[docs]def addNevowAttributes(tag, **kwargs): """adds kwargs as n:key to a clone of tag's attrs. This is so you can add n:pattern, n:data and friends even within a stan DOM. """ res = tag.clone() for k, v in list(kwargs.items()): res.attributes["nevow:"+k] = v return res
[docs]class CommonRenderers(object): """A container for some basic renderers we want on all active elements. This is basically what nevow had. """
[docs] @template.renderer def sequence(self, request, tag): toProcess = tag.slotData patterns = locatePatterns(tag, ["item", "empty", "separator", "header", "footer"], ["sequence"]) if not patterns["item"]: patterns["item"] = [template.tags.transparent] tagIterator = iter(patterns["item"]) newTag = tag.clone(True).clear() newTag.render = None if patterns["header"]: newTag[patterns["header"]] for item in toProcess: try: nextTag = next(tagIterator) except StopIteration: newTag(patterns["separator"]) tagIterator = iter(patterns["item"]) nextTag = next(tagIterator) newTag(nextTag.clone(True, item)) if newTag.children or not patterns["empty"]: if patterns["footer"]: newTag[patterns["footer"]] return newTag else: return patterns["empty"]
[docs] @template.renderer def mapping(self, request, tag): return tag.fillSlots(**tag.slotData)
[docs] def data_key(self, keyName): """returns data[keyName]; of course, this only works if the current data is a dict. """ def _(request, tag): return tag.slotData[keyName] return _
[docs] @template.renderer def string(self, request, tag): return tag[str(tag.slotData)]
[docs] @template.renderer def xml(self, request, tag): return tag[template.tags.xml(tag.slotData)]
[docs] @template.renderer def passthrough(self, request, tag): """inserts the current data into the stan tree. That's nevow's "data" renderer; but it's more limited in that t.w's flattener is more limited. """ return tag[tag.slotData]
[docs] def lookupRenderMethod(self, name): if callable(name): return name if " " in name: parts = name.split() method = template.renderer.get(self, parts[0], None) if method is None: raise Exception("Missing render method on %s: %s"%( repr(self), name)) method = method(" ".join(parts[1:])) else: method = template.renderer.get(self, name, None) if method is None: raise Exception("Missing render method on %s: %s"%( repr(self), name)) return method
[docs] def lookupDataMethod(self, name): """returns a callable (request, tag) -> something. If name is a number, this will be tag.slotData[number]. If name contains a blank, name will be split, and data_name will be called with the parts[1:] as arguments to obtain the callable. Else this will just return data_name. """ try: ct = int(name) def getter(request, tag): data = "<unset>" try: return tag.slotData[ct] except Exception as ex: raise NoDataError("While trying to get item %s in %s:" " %s"%(ct, data, ex)) return getter except (ValueError, TypeError): # it's not a number, so keep trying pass if " " in name: parts = name.split() return getattr(self, "data_"+parts[0])(" ".join(parts[1:])) return getattr(self, "data_"+name)
[docs]class NevowcElement(CommonRenderers, template.Element): """a template.Element that has all our basic renderer functions. This is probably what you want to base your own elements on. """ pass
[docs]def elementFromTag(tag, rendererFactory): """returns a t.w.template element rendering tag but using a rendererFactory (a class). This is used here to furnish the template element with externally defined renderers. You probably won't need this outside of testing code, as TemplatedPage already arranges everything if you use loader. """ class _Root(rendererFactory, template.Element): def __init__(self, child): self.child = child def render(self, request): return self.child return _Root(tag)
[docs]class TemplatedPage(resource.Resource, CommonRenderers): """A t.w Resource rendering from its loader attribute. To actually have the full feature set, be sure to use the XMLFile loader from this module as the loader. It does *not* restore ``renderHTTP`` or ``locateChild``, as there's no sane way to keep the interface. Port to t.w-style getChild; if you override ``render_GET``, you probably want to end it with ``return TemplatedPage.render_GET(self, request)``. This will look for a ``gavo_useDoctype`` attribute on the template and use an XHTML doctype if it's not present. You can override ``handleError(failure, request)`` to override the error document (this will finish the request, too). If you need extra callbacks, define your own render method and use the renderDocument method to obtain a "naked" deferred. """ loader = None # override in derived classes defaultDoctype = ('<!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.0 Strict//EN"' ' "">') def _getDoc(self, request): if self.loader is None: raise Exception("You must override loader on TemplatedPage") # define an artificial root element so renderers see the resource's # attributes (like render methods or data attributes) class _Root(template.Element): loader = self.loader # make the element see our methods and attributes def __getattr__(el_self, name): if name.startswith("__"): raise AttributeError(name) return getattr(self, name) # we need to redirect lookupRenderMethod so Element # can deal with render="foo arg". def lookupRenderMethod(el_self, *args): return self.lookupRenderMethod(*args) return _Root()
[docs] def handleError(self, failure, request): failure.printTraceback() request.write("Uncaught error:\n<pre><![CDATA[\n{}\n]]></pre>\n" .format(failure.getErrorMessage()) .encode("utf-8")) try: request.finish() except defer.AlreadyCalledError: # I can't say what state I'm in at this state. If the request # was already finished nobody saw the message, so at least # gargle it out on stderr and hope for the best sys.stderr.write("Error on request that was already finished:\n") sys.stderr.write(failure.getErrorMessage()) pass
[docs] def renderDocument(self, request): """returns a deferred firing when the document is written. Use this method rather than render if you need to introduce extra callbacks. This does *not* finish the request by itself. """ doc = self._getDoc(request) doctype = getattr(doc, "gavo_useDoctype", self.defaultDoctype) if isinstance(doctype, str): doctype = doctype.encode("utf-8") if doctype is not None: request.write(doctype) request.write(b'\n') return template.flatten(request, doc, request.write)
[docs] def finishRequest(self, ignored, request): request.finish()
[docs] def render_GET(self, request): self.renderDocument(request ).addCallback(self.finishRequest, request ).addErrback(self.handleError, request) return server.NOT_DONE_YET
# default: handle GET and POST equivalently, don't # care about HEAD (should we?) render_POST = render_GET
[docs] def getChild(self, name, request): if not name and not request.postpath: return self return resource.NoResource()
[docs]class Redirect(resource.Resource): # todo: Anyone using this? Why not t.w.util.Redirect? def __init__(self, destURL, code=302): self.destURL = str(destURL) self.code = 302
[docs] def render(self, request): # todo: see if destURL is relative and fix based on request? request.setHeader("location", self.destURL) request.setResponseCode(self.code) request.setHeader("Content-Type", "text/plain") return ("You are not supposed to see this, but if you do:" " Your browser was supposed to go to %s"%self.destURL)
[docs]def flattenSync(element, request=None): """returns a string representation of element synchronously. This, of course, only works if there are no deferreds within element. """ result = [flattenSync] def _(res): result[0] = res template.flattenString(request, element).addBoth(_) result = result[0] if result is flattenSync: raise NotImplementedError("flattenSync cannot deal with elements" " containing deferreds.") elif hasattr(result, "raiseException"): result.raiseException() return result
[docs]def flattenSyncToString(element, request=None): """returns elemented flattened to a string (as opposed to bytes) """ return flattenSync(element, request).decode("utf-8")
# vim:et:sta:sw=4