libervia-web: libervia/server/pages.py comparison

comparison libervia/server/pages.py @ 1216:b2d067339de3

python 3 port: /!\ Python 3.6+ is now needed to use libervia /!\ instability may occur and features may not be working anymore, this will improve with time /!\ TxJSONRPC dependency has been removed The same procedure as in backend has been applied (check backend commit ab2696e34d29 logs for details). Removed now deprecated code (Pyjamas compiled browser part, legacy blog, JSON RPC related code). Adapted code to work without `html` and `themes` dirs.

author	Goffi <goffi@goffi.org>
date	Tue, 13 Aug 2019 19:12:31 +0200
parents	584e29d9510a
children	62bf4f87c249

comparison

equal deleted inserted replaced

-:f14ab8a25e8b
+:b2d067339de3
 from libervia.server.utils import quote, SubPage
 from libervia.server.classes import WebsocketMeta
 import uuid
 import os.path
-import urllib
+import urllib.request, urllib.parse, urllib.error
 import time
 import hashlib
 import copy
+from functools import reduce
 log = getLogger(__name__)
 class CacheBase(object):
 self.name = name
 if name is not None:
 if (name in self.named_pages
 and not (replace_on_conflict and self.named_pages[name].url == url)):
 raise exceptions.ConflictError(
-_(u'a Libervia page named "{}" already exists'.format(name)))
+_('a Libervia page named "{}" already exists'.format(name)))
-if u"/" in name:
+if "/" in name:
-raise ValueError(_(u'"/" is not allowed in page names'))
+raise ValueError(_('"/" is not allowed in page names'))
 if not name:
-raise ValueError(_(u"a page name can't be empty"))
+raise ValueError(_("a page name can't be empty"))
 self.named_pages[name] = self
 if access is None:
 access = C.PAGES_ACCESS_PUBLIC
 if access not in (
 C.PAGES_ACCESS_PUBLIC,
 C.PAGES_ACCESS_PROFILE,
 C.PAGES_ACCESS_NONE,
 ):
 raise NotImplementedError(
-_(u"{} access is not implemented yet").format(access)
+_("{} access is not implemented yet").format(access)
 )
 self.access = access
 self.dynamic = dynamic
 if redirect is not None:
 # only page access and name make sense in case of full redirection
 if not all(
 lambda x: x is not None
 for x in (parse_url, prepare_render, render, template)
 ):
 raise ValueError(
-_(u"you can't use full page redirection with other rendering"
+_("you can't use full page redirection with other rendering"
-u"method, check self.pageRedirect if you need to use them"))
+"method, check self.pageRedirect if you need to use them"))
 self.redirect = redirect
 else:
 self.redirect = None
 self.parse_url = parse_url
 self.prepare_render = prepare_render
 self.url_cache = url_cache
 if access == C.PAGES_ACCESS_NONE:
 # none pages just return a 404, no further check is needed
 return
 if template is not None and render is not None:
-log.error(_(u"render and template methods can't be used at the same time"))
+log.error(_("render and template methods can't be used at the same time"))
 if parse_url is not None and not callable(parse_url):
-log.error(_(u"parse_url must be a callable"))
+log.error(_("parse_url must be a callable"))
 # if not None, next rendering will be cached
 #  it must then contain a list of the the keys to use (without the page instance)
 # e.g. [C.SERVICE_PROFILE, "pubsub", server@example.tld, pubsub_node]
 self._do_cache = None
-def __unicode__(self):
-return u"LiberviaPage {name} at {url} (vhost: {vhost_root})".format(
-name=self.name or u"<anonymous>", url=self.url, vhost_root=self.vhost_root)
 def __str__(self):
-return self.__unicode__().encode("utf-8")
+return "LiberviaPage {name} at {url} (vhost: {vhost_root})".format(
+name=self.name or "<anonymous>", url=self.url, vhost_root=self.vhost_root)
 @property
 def named_pages(self):
 return self.vhost_root.named_pages
 @return (tuple[dict, LiberviaPage]): tuple with:
 - page_data: dict containing data of the page
 - libervia_page: created resource
 """
 dir_path = os.path.dirname(meta_path)
-page_data = {"__name__": u".".join([u"page"] + url_elts)}
+page_data = {"__name__": ".".join(["page"] + url_elts)}
 # we don't want to force the presence of __init__.py
 # so we use execfile instead of import.
 # TODO: when moved to Python 3, __init__.py is not mandatory anymore
 #       so we can switch to import
-execfile(meta_path, page_data)
+exec(compile(open(meta_path, "rb").read(), meta_path, 'exec'), page_data)
 return page_data, LiberviaPage(
 host=host,
 vhost_root=vhost_root,
 root_dir=dir_path,
-url=u"/" + u"/".join(url_elts),
+url="/" + "/".join(url_elts),
-name=page_data.get(u"name"),
+name=page_data.get("name"),
-redirect=page_data.get(u"redirect"),
+redirect=page_data.get("redirect"),
-access=page_data.get(u"access"),
+access=page_data.get("access"),
-dynamic=page_data.get(u"dynamic", False),
+dynamic=page_data.get("dynamic", False),
-parse_url=page_data.get(u"parse_url"),
+parse_url=page_data.get("parse_url"),
-prepare_render=page_data.get(u"prepare_render"),
+prepare_render=page_data.get("prepare_render"),
-render=page_data.get(u"render"),
+render=page_data.get("render"),
-template=page_data.get(u"template"),
+template=page_data.get("template"),
-on_data_post=page_data.get(u"on_data_post"),
+on_data_post=page_data.get("on_data_post"),
-on_data=page_data.get(u"on_data"),
+on_data=page_data.get("on_data"),
-on_signal=page_data.get(u"on_signal"),
+on_signal=page_data.get("on_signal"),
-url_cache=page_data.get(u"url_cache", False),
+url_cache=page_data.get("url_cache", False),
 replace_on_conflict=replace_on_conflict
 )
 @classmethod
 def importPages(cls, host, vhost_root, root_path=None, _parent=None, _path=None,
 for d in os.listdir(root_dir):
 dir_path = os.path.join(root_dir, d)
 if not os.path.isdir(dir_path):
 continue
 if _extra_pages and d in _parent.children:
-log.debug(_(u"[{host_name}] {path} is already present, ignoring it")
+log.debug(_("[{host_name}] {path} is already present, ignoring it")
-.format(host_name=vhost_root.host_name, path=u'/'.join(_path+[d])))
+.format(host_name=vhost_root.host_name, path='/'.join(_path+[d])))
 continue
 meta_path = os.path.join(dir_path, C.PAGES_META_FILE)
 if os.path.isfile(meta_path):
 new_path = _path + [d]
 try:
 if _extra_pages:
 # extra pages are discarded if there is already an existing page
 continue
 else:
 raise e
-_parent.putChild(d, resource)
+_parent.putChild(d.encode('utf-8'), resource)
-log_msg = (u"[{host_name}] Added /{path} page".format(
+log_msg = ("[{host_name}] Added /{path} page".format(
 host_name=vhost_root.host_name,
-path=u"[…]/".join(new_path)))
+path="[…]/".join(new_path)))
 if _extra_pages:
 log.debug(log_msg)
 else:
 log.info(log_msg)
 if "uri_handlers" in page_data:
 if not isinstance(page_data, dict):
-log.error(_(u"uri_handlers must be a dict"))
+log.error(_("uri_handlers must be a dict"))
 else:
-for uri_tuple, cb_name in page_data["uri_handlers"].iteritems():
+for uri_tuple, cb_name in page_data["uri_handlers"].items():
-if len(uri_tuple) != 2 or not isinstance(cb_name, basestring):
+if len(uri_tuple) != 2 or not isinstance(cb_name, str):
-log.error(_(u"invalid uri_tuple"))
+log.error(_("invalid uri_tuple"))
 continue
 if not _extra_pages:
-log.info(_(u"setting {}/{} URIs handler")
+log.info(_("setting {}/{} URIs handler")
 .format(*uri_tuple))
 try:
 cb = page_data[cb_name]
 except KeyError:
-log.error(_(u"missing {name} method to handle {1}/{2}")
+log.error(_("missing {name} method to handle {1}/{2}")
 .format(name=cb_name, *uri_tuple))
 continue
 else:
 resource.registerURI(uri_tuple, cb)
 """
 if flags == ['create']:
 return
 path = file_path.path.decode('utf-8')
 base_name = os.path.basename(path)
-if base_name != u"page_meta.py":
+if base_name != "page_meta.py":
 # we only handle libervia pages
 return
-log.debug(u"{flags} event(s) received for {file_path}".format(
+log.debug("{flags} event(s) received for {file_path}".format(
-flags=u", ".join(flags), file_path=file_path))
+flags=", ".join(flags), file_path=file_path))
 dir_path = os.path.dirname(path)
 if not dir_path.startswith(site_path):
-raise exceptions.InternalError(u"watched file should start with site path")
+raise exceptions.InternalError("watched file should start with site path")
 path_elts = [p for p in dir_path[len(site_path):].split('/') if p]
 if not path_elts:
 return
 page = page.children[child_name]
 except KeyError:
 if idx != len(path_elts)-1:
 # a page has been created in a subdir when one or more
 # page_meta.py are missing on the way
-log.warning(_(u"Can't create a page at {path}, missing parents")
+log.warning(_("Can't create a page at {path}, missing parents")
 .format(path=path))
 return
 new_page = True
 else:
 if idx<len(path_elts)-1:
 except AttributeError:
 # FIXME: this .original handling madness is due to EncodingResourceWrapper
 #        EncodingResourceWrapper should probably be removed
 resource.children = page.children
 except Exception as e:
-log.warning(_(u"Can't create page: {reason}").format(reason=e))
+log.warning(_("Can't create page: {reason}").format(reason=e))
 else:
 url_elt = path_elts[-1]
 if not new_page:
 # the page was already existing, we remove it
 del parent.children[url_elt]
 # we can now add the new page
 parent.putChild(url_elt, resource)
 if new_page:
-log.info(_(u"{page} created").format(page=resource))
+log.info(_("{page} created").format(page=resource))
 else:
-log.info(_(u"{page} reloaded").format(page=resource))
+log.info(_("{page} reloaded").format(page=resource))
 def registerURI(self, uri_tuple, get_uri_cb):
 """Register a URI handler
 @param uri_tuple(tuple[unicode, unicode]): type or URIs handler
 @param get_uri_cb(callable): method which take uri_data dict as only argument
 and return absolute path with correct arguments or None if the page
 can't handle this URL
 """
 if uri_tuple in self.uri_callbacks:
-log.info(_(u"{}/{} URIs are already handled, replacing by the new handler")
+log.info(_("{}/{} URIs are already handled, replacing by the new handler")
 .format( *uri_tuple))
 self.uri_callbacks[uri_tuple] = (self, get_uri_cb)
 def getSignalId(self, request):
 """Retrieve signal_id for a request
 are doing if you unset this option /!\
 """
 # FIXME: add a timeout; if socket is not opened before it, signal handler
 #        must be removed
 if not self.dynamic:
-log.error(_(u"You can't register signal if page is not dynamic"))
+log.error(_("You can't register signal if page is not dynamic"))
 return
 signal_id = self.getSignalId(request)
 LiberviaPage.signals_handlers.setdefault(signal, {})[signal_id] = [
 self,
 request,
 return self.vhost_root.getPageByName(name)
 def getPagePathFromURI(self, uri):
 return self.vhost_root.getPagePathFromURI(uri)
-def getPageRedirectURL(self, request, page_name=u"login", url=None):
+def getPageRedirectURL(self, request, page_name="login", url=None):
 """generate URL for a page with redirect_url parameter set
 mainly used for login page with redirection to current page
 @param request(server.Request): current HTTP request
 @param page_name(unicode): name of the page to go
 @param url(None, unicode): url to redirect to
 None to use request path (i.e. current page)
 @return (unicode): URL to use
 """
-return u"{root_url}?redirect_url={redirect_url}".format(
+return "{root_url}?redirect_url={redirect_url}".format(
 root_url=self.getPageByName(page_name).url,
-redirect_url=urllib.quote_plus(request.uri)
+redirect_url=urllib.parse.quote_plus(request.uri)
 if url is None
 else url.encode("utf-8"),
 )
 def getURL(self, *args):
 if self.name is not None and self.name in self.pages_redirects:
 #  we check for redirection
 redirect_data = self.pages_redirects[self.name]
 args_hash = tuple(args)
-for limit in xrange(len(args) + 1):
+for limit in range(len(args) + 1):
 current_hash = args_hash[:limit]
 if current_hash in redirect_data:
 url_base = redirect_data[current_hash]
 remaining = args[limit:]
 remaining_url = "/".join(remaining)
 # request.prepath) because request.prepath may have been modified by
 # redirection (if redirection args have been specified), while path reflect
 # the real request
 # we ignore empty path elements (i.e. double '/' or '/' at the end)
-path_elts = [p for p in request.path.split("/") if p]
+path_elts = [p for p in request.path.decode('utf-8').split("/") if p]
 if request.postpath:
 if not request.postpath[-1]:
 #  we remove trailing slash
 request.postpath = request.postpath[:-1]
 #  getSubPageURL must return subpage from the point where
 # the it is called, so we have to remove remanining
 # path elements
 path_elts = path_elts[: -len(request.postpath)]
-return u"/" + "/".join(path_elts).decode("utf-8")
+return "/" + "/".join(path_elts)
 def getParamURL(self, request, **kwargs):
 """use URL of current request but modify the parameters in query part
 **kwargs(dict[str, unicode]): argument to use as query parameters
 @return (unicode): constructed URL
 """
 current_url = self.getCurrentURL(request)
 if kwargs:
-encoded = urllib.urlencode(
+encoded = urllib.parse.urlencode(
-{k: v.encode("utf-8") for k, v in kwargs.iteritems()}
+{k: v for k, v in kwargs.items()}
-).decode("utf-8")
+)
-current_url = current_url + u"?" + encoded
+current_url = current_url + "?" + encoded
 return current_url
 def getSubPageByName(self, subpage_name, parent=None):
 """retrieve a subpage and its path using its name
 @return (tuple[str, LiberviaPage]): page subpath and instance
 @raise exceptions.NotFound: no page has been found
 """
 if parent is None:
 parent = self
-for path, child in parent.children.iteritems():
+for path, child in parent.children.items():
 try:
 child_name = child.name
 except AttributeError:
 #  LiberviaPages have a name, but maybe this is an other Resource
 continue
 if child_name == subpage_name:
-return path, child
+return path.decode('utf-8'), child
-raise exceptions.NotFound(_(u"requested sub page has not been found"))
+raise exceptions.NotFound(_("requested sub page has not been found"))
 def getSubPageURL(self, request, page_name, *args):
 """retrieve a page in direct children and build its URL according to request
 request's current path is used as base (at current parsing point,
 @return (unicode): absolute URL to the sub page
 """
 current_url = self.getCurrentURL(request)
 path, child = self.getSubPageByName(page_name)
 return os.path.join(
-u"/", current_url, path, *[quote(a) for a in args if a is not None]
+"/", current_url, path, *[quote(a) for a in args if a is not None]
 )
 def getURLByNames(self, named_path):
 """Retrieve URL from pages names and arguments
 page_name, parent=current_page
 )
 path.append(sub_path)
 if page_args:
 path.extend([quote(a) for a in page_args])
-return self.host.checkRedirection(self.vhost_root, u"/".join(path))
+return self.host.checkRedirection(self.vhost_root, "/".join(path))
 def getURLByPath(self, *args):
 """Generate URL by path
 this method as a similar effect as getURLByNames, but it is more readable
 if not args:
 break
 else:
 path, current_page = current_page.getSubPageByName(args.pop(0))
 arguments = [path]
-return self.host.checkRedirection(self.vhost_root, u"/".join(url_elts))
+return self.host.checkRedirection(self.vhost_root, "/".join(url_elts))
 def getChildWithDefault(self, path, request):
 # we handle children ourselves
 raise exceptions.InternalError(
-u"this method should not be used with LiberviaPage"
+"this method should not be used with LiberviaPage"
 )
 def nextPath(self, request):
 """get next URL path segment, and update request accordingly
 @return (unicode): unquoted segment
 @raise IndexError: there is no segment left
 """
 pathElement = request.postpath.pop(0)
 request.prepath.append(pathElement)
-return urllib.unquote(pathElement).decode("utf-8")
+return urllib.parse.unquote(pathElement.decode('utf-8'))
 def _filterPathValue(self, value, handler, name, request):
 """Modify a path value according to handler (see [getPathArgs])"""
-if handler in (u"@", u"@jid") and value == u"@":
+if handler in ("@", "@jid") and value == "@":
 value = None
-if handler in (u"", u"@"):
+if handler in ("", "@"):
 if value is None:
-return u""
+return ""
-elif handler in (u"jid", u"@jid"):
+elif handler in ("jid", "@jid"):
 if value:
 try:
 return jid.JID(value)
 except RuntimeError:
-log.warning(_(u"invalid jid argument: {value}").format(value=value))
+log.warning(_("invalid jid argument: {value}").format(value=value))
 self.pageError(request, C.HTTP_BAD_REQUEST)
 else:
-return u""
+return ""
 else:
 return handler(self, value, name, request)
 return value
 used, else it will be converted to jid
 """
 data = self.getRData(request)
 for idx, name in enumerate(names):
-if name[0] == u"*":
+if name[0] == "*":
 value = data[name[1:]] = []
 while True:
 try:
 value.append(self.nextPath(request))
 except IndexError:
 idx -= 1
 break
 values_count = idx + 1
 if values_count < min_args:
-log.warning(_(u"Missing arguments in URL (got {count}, expected at least "
+log.warning(_("Missing arguments in URL (got {count}, expected at least "
-u"{min_args})").format(count=values_count, min_args=min_args))
+"{min_args})").format(count=values_count, min_args=min_args))
 self.pageError(request, C.HTTP_BAD_REQUEST)
 for name in names[values_count:]:
 data[name] = None
-for name, handler in kwargs.iteritems():
+for name, handler in kwargs.items():
 if name[0] == "*":
 data[name] = [
 self._filterPathValue(v, handler, name, request) for v in data[name]
 ]
 else:
 if params is None:
 params = self.getAllPostedData(request, multiple=False)
 if extra is None:
 extra = {}
 else:
-assert not {u"rsm_max", u"rsm_after", u"rsm_before",
+assert not {"rsm_max", "rsm_after", "rsm_before",
-C.KEY_ORDER_BY}.intersection(extra.keys())
+C.KEY_ORDER_BY}.intersection(list(extra.keys()))
-extra[u"rsm_max"] = unicode(page_max)
+extra["rsm_max"] = str(page_max)
 if order_by is not None:
 extra[C.KEY_ORDER_BY] = order_by
-if u'after' in params:
+if 'after' in params:
-extra[u'rsm_after'] = params[u'after']
+extra['rsm_after'] = params['after']
-elif u'before' in params:
+elif 'before' in params:
-extra[u'rsm_before'] = params[u'before']
+extra['rsm_before'] = params['before']
 return extra
 def setPagination(self, request, pubsub_data):
 """Add  to template_data if suitable
 @param pubsub_data(dict): pubsub metadata parsed with
 data_objects.parsePubSubMetadata
 """
 template_data = request.template_data
 try:
-last_id = pubsub_data[u"rsm_last"]
+last_id = pubsub_data["rsm_last"]
 except KeyError:
 # no pagination available
 return
 if pubsub_data.get("rsm_index", 1) > 0:
 # We only show previous button if it's not the first page already.
 # If we have no index, we default to display the button anyway
 # as we can't know if we are on the first page or not.
-first_id = pubsub_data[u"rsm_first"]
+first_id = pubsub_data["rsm_first"]
 template_data['previous_page_url'] = self.getParamURL(request,
 before=first_id)
-if not pubsub_data[u"complete"]:
+if not pubsub_data["complete"]:
 # we also show the page next button if complete is None because we
 # can't know where we are in the feed in this case.
 template_data['next_page_url'] = self.getParamURL(request, after=last_id)
 def checkCacheSubscribeCb(self, sub_id, service, node):
 self.cache_pubsub_sub.add((service, node, sub_id))
 def checkCacheSubscribeEb(self, failure_, service, node):
-log.warning(_(u"Can't subscribe to node: {msg}").format(msg=failure_))
+log.warning(_("Can't subscribe to node: {msg}").format(msg=failure_))
 # FIXME: cache must be marked as unusable here
 def psNodeWatchAddEb(self, failure_, service, node):
-log.warning(_(u"Can't add node watched: {msg}").format(msg=failure_))
+log.warning(_("Can't add node watched: {msg}").format(msg=failure_))
 def checkCache(self, request, cache_type, **kwargs):
 """check if a page is in cache and return cached version if suitable
 this method may perform extra operation to handle cache (e.g. subscribing to a
 if not node:
 try:
 short = kwargs["short"]
 node = self.host.ns_map[short]
 except KeyError:
-log.warning(_(u'Can\'t use cache for empty node without namespace '
+log.warning(_('Can\'t use cache for empty node without namespace '
-u'set, please ensure to set "short" and that it is '
+'set, please ensure to set "short" and that it is '
-u'registered'))
+'registered'))
 return
 if profile != C.SERVICE_PROFILE:
 #  only service profile is cache for now
 return
 session_data = self.host.getSessionData(request, session_iface.ISATSession)
 #  we don't return the Deferreds as it is not needed to wait for
 # the subscription to continue with page rendering
 return
 else:
-raise exceptions.InternalError(u"Unknown cache_type")
+raise exceptions.InternalError("Unknown cache_type")
-log.debug(u"using cache for {page}".format(page=self))
+log.debug("using cache for {page}".format(page=self))
 cache.last_access = time.time()
 self._setCacheHeaders(request, cache)
 self._checkCacheHeaders(request, cache)
 request.write(cache.rendered)
 request.finish()
-raise failure.Failure(exceptions.CancelError(u"cache is used"))
+raise failure.Failure(exceptions.CancelError("cache is used"))
 def _cacheURL(self, __, request, profile):
 self.cached_urls.setdefault(profile, {})[request.uri] = CacheURL(request)
 @classmethod
 """Invalidate cache for all pages linked to this node"""
 try:
 cache = cls.cache[profile][C.CACHE_PUBSUB][jid.JID(service)][node]
 except KeyError:
 log.info(_(
-u"Removing subscription for {service}/{node}: "
+"Removing subscription for {service}/{node}: "
-u"the page is not cached").format(service=service, node=node))
+"the page is not cached").format(service=service, node=node))
 d1 = host.bridgeCall("psUnsubscribe", service, node, profile)
 d1.addErrback(
 lambda failure_: log.warning(
-_(u"Can't unsubscribe from {service}/{node}: {msg}").format(
+_("Can't unsubscribe from {service}/{node}: {msg}").format(
 service=service, node=node, msg=failure_)))
 d2 = host.bridgeCall("psNodeWatchAdd", service, node, profile)
 # TODO: check why the page is not in cache, remove subscription?
 d2.addErrback(
 lambda failure_: log.warning(
-_(u"Can't remove watch for {service}/{node}: {msg}").format(
+_("Can't remove watch for {service}/{node}: {msg}").format(
 service=service, node=node, msg=failure_)))
 else:
 cache.clear()
 @classmethod
 @param signal(unicode): name of the signal
 @param *args: args of the signals
 """
 for page, request, check_profile in cls.signals_handlers.get(
 signal, {}
-).itervalues():
+).values():
 if check_profile:
 signal_profile = args[-1]
 request_profile = page.getProfile(request)
 if not request_profile:
 # if you want to use signal without session, unset check_profile
 # (be sure to know what you are doing)
-log.error(_(u"no session started, signal can't be checked"))
+log.error(_("no session started, signal can't be checked"))
 continue
 if signal_profile != request_profile:
 #  we ignore the signal, it's not for our profile
 continue
 if request._signals_cache is not None:
 # socket is not yet opened, we cache the signal
 request._signals_cache.append((request, signal, args))
 log.debug(
-u"signal [{signal}] cached: {args}".format(signal=signal, args=args)
+"signal [{signal}] cached: {args}".format(signal=signal, args=args)
 )
 else:
 page.on_signal(page, request, signal, *args)
 def onSocketOpen(self, request):
 """
 assert request._signals_cache is not None
 # we need to replace corresponding original requests by this websocket request
 # in signals_handlers
 signal_id = request.signal_id
-for signal_handlers_map in self.__class__.signals_handlers.itervalues():
+for signal_handlers_map in self.__class__.signals_handlers.values():
 if signal_id in signal_handlers_map:
 signal_handlers_map[signal_id][1] = request
 cache = request._signals_cache
 request._signals_cache = None
 for signal in request._signals_registered:
 signal_id = self.getSignalId(request)
 try:
 del LiberviaPage.signals_handlers[signal][signal_id]
 except KeyError:
-log.error(_(u"Can't find signal handler for [{signal}], this should not "
+log.error(_("Can't find signal handler for [{signal}], this should not "
-u"happen").format(signal=signal))
+"happen").format(signal=signal))
 else:
-log.debug(_(u"Removed signal handler"))
+log.debug(_("Removed signal handler"))
 def delegateToResource(self, request, resource):
 """continue workflow with Twisted Resource"""
 buf = resource.render(request)
 if buf == server.NOT_DONE_YET:
 pass
 else:
 request.write(buf)
 request.finish()
-raise failure.Failure(exceptions.CancelError(u"resource delegation"))
+raise failure.Failure(exceptions.CancelError("resource delegation"))
 def HTTPRedirect(self, request, url):
 """redirect to an URL using HTTP redirection
 @param request(server.Request): current HTTP request
 @param url(unicode): url to redirect to
 """
 web_util.redirectTo(url.encode("utf-8"), request)
 request.finish()
-raise failure.Failure(exceptions.CancelError(u"HTTP redirection is used"))
+raise failure.Failure(exceptions.CancelError("HTTP redirection is used"))
-def redirectOrContinue(self, request, redirect_arg=u"redirect_url"):
+def redirectOrContinue(self, request, redirect_arg="redirect_url"):
 """helper method to redirect a page to an url given as arg
 if the arg is not present, the page will continue normal workflow
 @param request(server.Request): current HTTP request
 @param redirect_arg(unicode): argument to use to get redirection URL
 @interrupt: redirect the page to requested URL
 @interrupt pageError(C.HTTP_BAD_REQUEST): empty or non local URL is used
 """
+redirect_arg = redirect_arg.encode('utf-8')
 try:
-url = request.args["redirect_url"][0]
+url = request.args[redirect_arg][0].decode('utf-8')
 except (KeyError, IndexError):
 pass
 else:
 #  a redirection is requested
-if not url or url[0] != u"/":
+if not url or url[0] != "/":
 # we only want local urls
 self.pageError(request, C.HTTP_BAD_REQUEST)
 else:
 self.HTTPRedirect(request, url)
 skipped
 @param path_args(list[unicode], None): path arguments to use in redirected page
 @raise KeyError: there is no known page with this name
 """
 # FIXME: render non LiberviaPage resources
-path = page_path.rstrip(u"/").split(u"/")
+path = page_path.rstrip("/").split("/")
 if not path[0]:
 redirect_page = self.vhost_root
 else:
 redirect_page = self.named_pages[path[0]]
 for subpage in path[1:]:
+subpage = subpage.encode('utf-8')
 if redirect_page is self.vhost_root:
 redirect_page = redirect_page.children[subpage]
 else:
 redirect_page = redirect_page.original.children[subpage]
 # if cache is needed, it will be handled by final page
 redirect_page._do_cache = self._do_cache
 self._do_cache = None
 redirect_page.renderPage(request, skip_parse_url=skip_parse_url)
-raise failure.Failure(exceptions.CancelError(u"page redirection is used"))
+raise failure.Failure(exceptions.CancelError("page redirection is used"))
 def pageError(self, request, code=C.HTTP_NOT_FOUND, no_body=False):
 """generate an error page and terminate the request
 @param request(server.Request): HTTP request
 self._do_cache = None
 request.setResponseCode(code)
 if no_body:
 request.finish()
 else:
-template = u"error/" + unicode(code) + ".html"
+template = "error/" + str(code) + ".html"
 template_data = request.template_data
 session_data = self.host.getSessionData(request, session_iface.ISATSession)
 if session_data.locale is not None:
-template_data[u'locale'] = session_data.locale
+template_data['locale'] = session_data.locale
 if self.vhost_root.site_name:
-template_data[u'site'] = self.vhost_root.site_name
+template_data['site'] = self.vhost_root.site_name
 rendered = self.host.renderer.render(
 template,
 error_code=code,
 **template_data
 )
 self.writeData(rendered, request)
-raise failure.Failure(exceptions.CancelError(u"error page is used"))
+raise failure.Failure(exceptions.CancelError("error page is used"))
 def writeData(self, data, request):
 """write data to transport and finish the request"""
 if data is None:
 self.pageError(request)
 if self._do_cache is not None:
 redirected_page = self._do_cache.pop(0)
 cache = reduce(lambda d, k: d.setdefault(k, {}), self._do_cache, self.cache)
 page_cache = cache[redirected_page] = CachePage(data_encoded)
 self._setCacheHeaders(request, page_cache)
-log.debug(_(u"{page} put in cache for [{profile}]")
+log.debug(_("{page} put in cache for [{profile}]")
 .format( page=self, profile=self._do_cache[0]))
 self._do_cache = None
 self._checkCacheHeaders(request, page_cache)
 try:
 request.write(data_encoded)
 except AttributeError:
-log.warning(_(u"Can't write page, the request has probably been cancelled "
+log.warning(_("Can't write page, the request has probably been cancelled "
-u"(browser tab closed or reloaded)"))
+"(browser tab closed or reloaded)"))
 return
 request.finish()
 def _subpagesHandler(self, __, request):
 """render subpage if suitable
 and check if it corresponds to a subpage. If so, it render the subpage
 else it render a NoResource.
 If there is no unmanaged part of the segment, current page workflow is pursued
 """
 if request.postpath:
-subpage = self.nextPath(request)
+subpage = self.nextPath(request).encode('utf-8')
 try:
 child = self.children[subpage]
 except KeyError:
 self.pageError(request)
 else:
 child.render(request)
-raise failure.Failure(exceptions.CancelError(u"subpage page is used"))
+raise failure.Failure(exceptions.CancelError("subpage page is used"))
 def _prepare_dynamic(self, __, request):
 # we need to activate dynamic page
 # we set data for template, and create/register token
-socket_token = unicode(uuid.uuid4())
+socket_token = str(uuid.uuid4())
 socket_url = self.host.getWebsocketURL(request)
 socket_debug = C.boolConst(self.host.debug)
 request.template_data["websocket"] = WebsocketMeta(
 socket_url, socket_token, socket_debug
 )
 template_data = request.template_data
 # if confirm variable is set in case of successfuly data post
 session_data = self.host.getSessionData(request, session_iface.ISATSession)
 if session_data.popPageFlag(self, C.FLAG_CONFIRM):
-template_data[u"confirm"] = True
+template_data["confirm"] = True
 notifs = session_data.popPageNotifications(self)
 if notifs:
-template_data[u"notifications"] = notifs
+template_data["notifications"] = notifs
 if session_data.locale is not None:
-template_data[u'locale'] = session_data.locale
+template_data['locale'] = session_data.locale
 if self.vhost_root.site_name:
-template_data[u'site'] = self.vhost_root.site_name
+template_data['site'] = self.vhost_root.site_name
 return self.host.renderer.render(
 self.template,
 page_url=self.getURL(),
-media_path=u"/" + C.MEDIA_DIR,
+media_path="/" + C.MEDIA_DIR,
 cache_path=session_data.cache_dir,
-build_path=u"/" + C.BUILD_DIR + u"/",
+build_path="/" + C.BUILD_DIR + "/",
 main_menu=self.main_menu,
 **template_data)
 def _renderEb(self, failure_, request):
 """don't raise error on CancelError"""
 failure_.trap(exceptions.CancelError)
 def _internalError(self, failure_, request):
 """called if an error is not catched"""
-if failure_.check(BridgeException) and failure_.value.condition == u'not-allowed':
+if failure_.check(BridgeException) and failure_.value.condition == 'not-allowed':
-log.warning(u"not allowed exception catched")
+log.warning("not allowed exception catched")
 self.pageError(request, C.HTTP_FORBIDDEN)
-log.error(_(u"Uncatched error for HTTP request on {url}: {msg}")
+log.error(_("Uncatched error for HTTP request on {url}: {msg}")
 .format( url=request.URLPath(), msg=failure_))
 self.pageError(request, C.HTTP_INTERNAL_ERROR)
 def _on_data_post_error(self, failure_, request):
 failure_.trap(exceptions.DataError)
 session_data = self.host.getSessionData(request, session_iface.ISATSession)
 session_data.setPageNotification(self, failure_.value.message, C.LVL_WARNING)
 request.setResponseCode(C.HTTP_SEE_OTHER)
 request.setHeader("location", request.uri)
 request.finish()
-raise failure.Failure(exceptions.CancelError(u"Post/Redirect/Get is used"))
+raise failure.Failure(exceptions.CancelError("Post/Redirect/Get is used"))
 def _on_data_post_redirect(self, ret, request):
 """called when page's on_data_post has been done successfuly
 This will do a Post/Redirect/Get pattern.
 @param ret(None, unicode, iterable): on_data_post return value
 see LiberviaPage.__init__ on_data_post docstring
 """
 if ret is None:
 ret = ()
-elif isinstance(ret, basestring):
+elif isinstance(ret, str):
 ret = (ret,)
 else:
 ret = tuple(ret)
 raise NotImplementedError(
-_(u"iterable in on_data_post return value is not used yet")
+_("iterable in on_data_post return value is not used yet")
 )
 session_data = self.host.getSessionData(request, session_iface.ISATSession)
 request_data = self.getRData(request)
 if "post_redirect_page" in request_data:
 redirect_page_data = request_data["post_redirect_page"]
 redirect_uri = request.uri
 if not C.POST_NO_CONFIRM in ret:
 session_data.setPageFlag(redirect_page, C.FLAG_CONFIRM)
 request.setResponseCode(C.HTTP_SEE_OTHER)
-request.setHeader("location", redirect_uri)
+request.setHeader(b"location", redirect_uri)
 request.finish()
-raise failure.Failure(exceptions.CancelError(u"Post/Redirect/Get is used"))
+raise failure.Failure(exceptions.CancelError("Post/Redirect/Get is used"))
 def _on_data_post(self, __, request):
 csrf_token = self.host.getSessionData(
 request, session_iface.ISATSession
 ).csrf_token
 try:
-given_csrf = self.getPostedData(request, u"csrf_token")
+given_csrf = self.getPostedData(request, "csrf_token")
 except KeyError:
 given_csrf = None
 if given_csrf is None or given_csrf != csrf_token:
 log.warning(
-_(u"invalid CSRF token, hack attempt? URL: {url}, IP: {ip}").format(
+_("invalid CSRF token, hack attempt? URL: {url}, IP: {ip}").format(
 url=request.uri, ip=request.getClientIP()
 )
 )
 self.pageError(request, C.HTTP_FORBIDDEN)
 d = defer.maybeDeferred(self.on_data_post, self, request)
 values received for this(these) key(s)
 @raise KeyError: one specific key has been requested, and it is missing
 """
 #  FIXME: request.args is already unquoting the value, it seems we are doing
 #         double unquote
-if isinstance(keys, basestring):
+if isinstance(keys, str):
 keys = [keys]
 get_first = True
 else:
 get_first = False
+keys = [k.encode('utf-8') for k in keys]
 ret = []
 for key in keys:
-gen = (urllib.unquote(v).decode("utf-8") for v in request.args.get(key, []))
+gen = (urllib.parse.unquote(v.decode("utf-8"))
+for v in request.args.get(key, []))
 if multiple:
 ret.append(gen)
 else:
 try:
 ret.append(next(gen))
 @param except_(iterable[unicode]): key of values to ignore
 csrf_token will always be ignored
 @param multiple(bool): if False, only the first values are returned
 @return (dict[unicode, list[unicode]]): post values
 """
-except_ = tuple(except_) + (u"csrf_token",)
+except_ = tuple(except_) + ("csrf_token",)
 ret = {}
-for key, values in request.args.iteritems():
+for key, values in request.args.items():
-key = urllib.unquote(key).decode("utf-8")
+key = key.decode('utf-8')
+key = urllib.parse.unquote(key)
 if key in except_:
 continue
+values = [v.decode('utf-8') for v in values]
 if not multiple:
-ret[key] = urllib.unquote(values[0]).decode("utf-8")
+ret[key] = urllib.parse.unquote(values[0])
 else:
-ret[key] = [urllib.unquote(v).decode("utf-8") for v in values]
+ret[key] = [urllib.parse.unquote(v) for v in values]
 return ret
 def getProfile(self, request):
 """helper method to easily get current profile
 """
 accept_language = request.getHeader("accept-language")
 if not accept_language:
 return
 accepted = {a.strip() for a in accept_language.split(',')}
-available = [unicode(l) for l in self.host.renderer.translations]
+available = [str(l) for l in self.host.renderer.translations]
 for lang in accepted:
 lang = lang.split(';')[0].strip().lower()
 if not lang:
 continue
 for a in available:
 @param template(unicode): path of the template to render
 @param template_data(dict): template_data to use
 """
 if not self.dynamic:
 raise exceptions.InternalError(
-_(u"renderPartial must only be used with dynamic pages")
+_("renderPartial must only be used with dynamic pages")
 )
 session_data = self.host.getSessionData(request, session_iface.ISATSession)
 if session_data.locale is not None:
-template_data[u'locale'] = session_data.locale
+template_data['locale'] = session_data.locale
 if self.vhost_root.site_name:
-template_data[u'site'] = self.vhost_root.site_name
+template_data['site'] = self.vhost_root.site_name
 return self.host.renderer.render(
 template,
 page_url=self.getURL(),
-media_path=u"/" + C.MEDIA_DIR,
+media_path="/" + C.MEDIA_DIR,
 cache_path=session_data.cache_dir,
-build_path=u"/" + C.BUILD_DIR + u"/",
+build_path="/" + C.BUILD_DIR + "/",
 main_menu=self.main_menu,
 **template_data
 )
 def renderAndUpdate(
 template_data = request.template_data.copy()
 template_data.update(template_data_update)
 html = self.renderPartial(request, template, template_data)
 try:
 request.sendData(
-u"dom", selectors=selectors, update_type=update_type, html=html)
+"dom", selectors=selectors, update_type=update_type, html=html)
 except Exception as e:
-log.error(u"Can't renderAndUpdate, html was: {html}".format(html=html))
+log.error("Can't renderAndUpdate, html was: {html}".format(html=html))
 raise e
 def renderPage(self, request, skip_parse_url=False):
 """Main method to handle the workflow of a LiberviaPage"""
 # template_data are the variables passed to template
 if not hasattr(request, "template_data"):
 session_data = self.host.getSessionData(request, session_iface.ISATSession)
 csrf_token = session_data.csrf_token
 request.template_data = {
-u"profile": session_data.profile,
+"profile": session_data.profile,
-u"csrf_token": csrf_token,
+"csrf_token": csrf_token,
 }
 # XXX: here is the code which need to be executed once
 #      at the beginning of the request hanling
 if request.postpath and not request.postpath[-1]:
 # i18n
 if C.KEY_LANG in request.args:
 try:
 locale = request.args.pop(C.KEY_LANG)[0]
 except IndexError:
-log.warning(u"empty lang received")
+log.warning("empty lang received")
 else:
-if u"/" in locale:
+if "/" in locale:
 # "/" is refused because locale may sometime be used to access
 # path, if localised documents are available for instance
-log.warning(_(u'illegal char found in locale ("/"), hack '
+log.warning(_('illegal char found in locale ("/"), hack '
-u'attempt? locale={locale}').format(locale=locale))
+'attempt? locale={locale}').format(locale=locale))
 locale = None
 session_data.locale = locale
 # if locale is not specified, we try to find one requested by browser
 if session_data.locale is None:
 # no cache for this URI yet
 #  we do normal URL parsing, and then the cache
 d.addCallback(self.parse_url, request)
 d.addCallback(self._cacheURL, request, profile)
 else:
-log.debug(_(u"using URI cache for {page}").format(page=self))
+log.debug(_("using URI cache for {page}").format(page=self))
 cache_url.use(request)
 else:
 d.addCallback(self.parse_url, request)
 d.addCallback(self._subpagesHandler, request)

Mercurial > libervia-web

comparison libervia/server/pages.py @ 1216:b2d067339de3