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

comparison libervia/server/pages.py @ 1509:106bae41f5c8

massive refactoring from camelCase -> snake_case. See backend commit log for more details

author	Goffi <goffi@goffi.org>
date	Sat, 08 Apr 2023 13:44:11 +0200
parents	ce879da7fcf7
children	16228994ca3b

comparison

equal deleted inserted replaced

-:ec3ad9abf9f9
+:106bae41f5c8
 from twisted.words.protocols.jabber import jid
 from twisted.python import failure
 from sat.core.i18n import _
 from sat.core import exceptions
-from sat.tools.utils import asDeferred
+from sat.tools.utils import as_deferred
 from sat.tools.common import date_utils
 from sat.tools.common import utils
 from sat.tools.common import data_format
 from sat.core.log import getLogger
 from sat_frontends.bridge.bridge_frontend import BridgeException
 this URL may not be valid, as pages may require path arguments
 @param name(unicode, None): if not None, a unique name to identify the page
 can then be used for e.g. redirection
 "/" is not allowed in names (as it can be used to construct URL paths)
 @param redirect(unicode, None): if not None, this page will be redirected.
-A redirected parameter is used as in self.pageRedirect.
+A redirected parameter is used as in self.page_redirect.
 parse_url will not be skipped
 using this redirect parameter is called "full redirection"
-using self.pageRedirect is called "partial redirection" (because some
+using self.page_redirect is called "partial redirection" (because some
 rendering method can still be used, e.g. parse_url)
 @param access(unicode, None): permission needed to access the page
 None means public access.
 Pages inherit from parent pages: e.g. if a "settings" page is restricted
 to admins, and if "settings/blog" is public, it still can only be accessed by
 lambda x: x is not None
 for x in (parse_url, prepare_render, render, template)
 ):
 raise ValueError(
 _("you can't use full page redirection with other rendering"
-"method, check self.pageRedirect if you need to use them"))
+"method, check self.page_redirect if you need to use them"))
 self.redirect = redirect
 else:
 self.redirect = None
 self.parse_url = parse_url
 self.add_breadcrumb = add_breadcrumb
 @property
 def site_themes(self):
 return self.vhost_root.site_themes
 @staticmethod
-def createPage(host, meta_path, vhost_root, url_elts, replace_on_conflict=False):
+def create_page(host, meta_path, vhost_root, url_elts, replace_on_conflict=False):
 """Create a LiberviaPage instance
 @param meta_path(Path): path to the page_meta.py file
 @param vhost_root(resource.Resource): root resource of the virtual host
 @param url_elts(list[unicode]): list of path element from root site to this page
 url_cache=page_data.get("url_cache", False),
 replace_on_conflict=replace_on_conflict
 )
 @staticmethod
-def createBrowserData(
+def create_browser_data(
 vhost_root,
 resource: Optional[LiberviaPage],
 browser_path: Path,
 path_elts: Optional[List[str]],
 engine: str = "brython"
 vhost_root.browser_modules.setdefault(
 engine, []).append(dyn_data)
 @classmethod
-def importPages(cls, host, vhost_root, root_path=None, _parent=None, _path=None,
+def import_pages(cls, host, vhost_root, root_path=None, _parent=None, _path=None,
 _extra_pages=False):
 """Recursively import Libervia pages
 @param host(Libervia): Libervia instance
 @param vhost_root(LiberviaRootResource): root of this VirtualHost
 root_dir = root_path / C.PAGES_DIR
 _extra_pages = True
 _parent = vhost_root
 root_browser_path = root_dir / C.PAGES_BROWSER_DIR
 if root_browser_path.is_dir():
-cls.createBrowserData(vhost_root, None, root_browser_path, None)
+cls.create_browser_data(vhost_root, None, root_browser_path, None)
 else:
 root_dir = _parent.root_dir
 for d in os.listdir(root_dir):
 dir_path = root_dir / d
 continue
 meta_path = dir_path / C.PAGES_META_FILE
 if meta_path.is_file():
 new_path = _path + [d]
 try:
-page_data, resource = cls.createPage(
+page_data, resource = cls.create_page(
 host, meta_path, vhost_root, new_path)
 except exceptions.ConflictError as e:
 if _extra_pages:
 # extra pages are discarded if there is already an existing page
 continue
 except KeyError:
 log.error(_("missing {name} method to handle {1}/{2}")
 .format(name=cb_name, *uri_tuple))
 continue
 else:
-resource.registerURI(uri_tuple, cb)
+resource.register_uri(uri_tuple, cb)
-LiberviaPage.importPages(
+LiberviaPage.import_pages(
 host, vhost_root, _parent=resource, _path=new_path,
 _extra_pages=_extra_pages)
 # now we check if there is some code for browser
 browser_path = dir_path / C.PAGES_BROWSER_DIR
 if browser_path.is_dir():
-cls.createBrowserData(vhost_root, resource, browser_path, new_path)
+cls.create_browser_data(vhost_root, resource, browser_path, new_path)
 @classmethod
-def onFileChange(cls, host, file_path, flags, site_root, site_path):
+def on_file_change(cls, host, file_path, flags, site_root, site_path):
 """Method triggered by file_watcher when something is changed in files
 This method is used in dev mode to reload pages when needed
 @param file_path(filepath.FilePath): path of the file which triggered the event
 @param flags[list[unicode]): human readable flags of the event (from
 except AttributeError:
 parent = page
 try:
 # we (re)create a page with the new/modified code
-__, resource = cls.createPage(host, path, site_root, path_elts,
+__, resource = cls.create_page(host, path, site_root, path_elts,
 replace_on_conflict=True)
 if not new_page:
 try:
 resource.children = page.original.children
 except AttributeError:
 parent.putChild(url_elt, resource)
 # is there any browser data to create?
 browser_path = resource.root_dir / C.PAGES_BROWSER_DIR
 if browser_path.is_dir():
-cls.createBrowserData(
+cls.create_browser_data(
 resource.vhost_root,
 resource,
 browser_path,
 resource.url.split('/')
 )
 if new_page:
 log.info(_("{page} created").format(page=resource))
 else:
 log.info(_("{page} reloaded").format(page=resource))
-def checkCSRF(self, request):
+def check_csrf(self, request):
-session = self.host.getSessionData(
+session = self.host.get_session_data(
 request, session_iface.IWebSession
 )
 if session.profile is None:
 # CSRF doesn't make sense when no user is logged
 log.debug("disabling CSRF check because service profile is used")
 return
 csrf_token = session.csrf_token
 given_csrf = request.getHeader("X-Csrf-Token")
 if given_csrf is None:
 try:
-given_csrf = self.getPostedData(request, "csrf_token")
+given_csrf = self.get_posted_data(request, "csrf_token")
 except KeyError:
 pass
 if given_csrf is None or given_csrf != csrf_token:
 log.warning(
 _("invalid CSRF token, hack attempt? URL: {url}, IP: {ip}").format(
 url=request.uri, ip=request.getClientIP()
 )
 )
-self.pageError(request, C.HTTP_FORBIDDEN)
+self.page_error(request, C.HTTP_FORBIDDEN)
-def exposeToScripts(
+def expose_to_scripts(
 self,
 request: server.Request,
 **kwargs: str
 ) -> None:
 """Make a local variable available to page script as a global variable
 value = repr(str(value))
 else:
 value = repr(value)
 scripts.add(Script(content=f"var {name}={value};"))
-def registerURI(self, uri_tuple, get_uri_cb):
+def register_uri(self, uri_tuple, get_uri_cb):
 """Register a URI handler
 @param uri_tuple(tuple[unicode, unicode]): type or URIs handler
-type/subtype as returned by tools/common/parseXMPPUri
+type/subtype as returned by tools/common/parse_xmpp_uri
 or type/None to handle all subtypes
 @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(_("{}/{} URIs are already handled, replacing by the new handler")
 .format( *uri_tuple))
 self.uri_callbacks[uri_tuple] = (self, get_uri_cb)
-def getConfig(self, key, default=None, value_type=None):
+def config_get(self, key, default=None, value_type=None):
-return self.host.getConfig(self.vhost_root, key=key, default=default,
+return self.host.config_get(self.vhost_root, key=key, default=default,
 value_type=value_type)
-def getBuildPath(self, session_data):
+def get_build_path(self, session_data):
 return session_data.cache_dir + self.vhost.site_name
-def getPageByName(self, name):
+def get_page_by_name(self, name):
-return self.vhost_root.getPageByName(name)
+return self.vhost_root.get_page_by_name(name)
-def getPagePathFromURI(self, uri):
+def get_page_path_from_uri(self, uri):
-return self.vhost_root.getPagePathFromURI(uri)
+return self.vhost_root.get_page_path_from_uri(uri)
-def getPageRedirectURL(self, request, page_name="login", url=None):
+def get_page_redirect_url(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 "{root_url}?redirect_url={redirect_url}".format(
-root_url=self.getPageByName(page_name).url,
+root_url=self.get_page_by_name(page_name).url,
 redirect_url=urllib.parse.quote_plus(request.uri)
 if url is None
 else url.encode("utf-8"),
 )
-def getURL(self, *args: str, **kwargs: str) -> str:
+def get_url(self, *args: str, **kwargs: str) -> str:
 """retrieve URL of the page set arguments
 @param *args: arguments to add to the URL as path elements empty or None
 arguments will be ignored
 @param **kwargs: query parameters
 encoded = urllib.parse.urlencode(
 {k: v for k, v in kwargs.items()}
 )
 url +=  f"?{encoded}"
-return self.host.checkRedirection(
+return self.host.check_redirection(
 self.vhost_root,
 url
 )
-def getCurrentURL(self, request):
+def get_current_url(self, request):
 """retrieve URL used to access this page
 @return(unicode): current URL
 """
 # we get url in the following way (splitting request.path instead of using
 if request.postpath:
 if not request.postpath[-1]:
 #  we remove trailing slash
 request.postpath = request.postpath[:-1]
 if request.postpath:
-#  getSubPageURL must return subpage from the point where
+#  get_sub_page_url 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 "/" + "/".join(path_elts)
-def getParamURL(self, request, **kwargs):
+def get_param_url(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)
+current_url = self.get_current_url(request)
 if kwargs:
 encoded = urllib.parse.urlencode(
 {k: v for k, v in kwargs.items()}
 )
 current_url = current_url + "?" + encoded
 return current_url
-def getSubPageByName(self, subpage_name, parent=None):
+def get_sub_page_by_name(self, subpage_name, parent=None):
 """retrieve a subpage and its path using its name
 @param subpage_name(unicode): name of the sub page
 it must be a direct children of parent page
 @param parent(LiberviaPage, None): parent page
 return path.decode('utf-8'), child
 raise exceptions.NotFound(
 _("requested sub page has not been found ({subpage_name})").format(
 subpage_name=subpage_name))
-def getSubPageURL(self, request, page_name, *args):
+def get_sub_page_url(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,
 i.e. it's more prepath than path).
 Requested page is checked in children and an absolute URL is then built
 by the resulting combination.
 This method is useful to construct absolute URLs for children instead of
 using relative path, which may not work in subpages, and are linked to the
 names of directories (i.e. relative URL will break if subdirectory is renamed
-while getSubPageURL won't as long as page_name is consistent).
+while get_sub_page_url won't as long as page_name is consistent).
 Also, request.path is used, keeping real path used by user,
 and potential redirections.
 @param request(server.Request): current HTTP request
 @param page_name(unicode): name of the page to retrieve
 it must be a direct children of current page
 @param *args(list[unicode]): arguments to add as path elements
 if an arg is None, it will be ignored
 @return (unicode): absolute URL to the sub page
 """
-current_url = self.getCurrentURL(request)
+current_url = self.get_current_url(request)
-path, child = self.getSubPageByName(page_name)
+path, child = self.get_sub_page_by_name(page_name)
 return os.path.join(
 "/", current_url, path, *[quote(a) for a in args if a is not None]
 )
-def getURLByNames(self, named_path):
+def get_url_by_names(self, named_path):
 """Retrieve URL from pages names and arguments
 @param named_path(list[tuple[unicode, list[unicode]]]): path to the page as a list
 of tuples of 2 items:
 - first item is page name
 """
 current_page = None
 path = []
 for page_name, page_args in named_path:
 if current_page is None:
-current_page = self.getPageByName(page_name)
+current_page = self.get_page_by_name(page_name)
-path.append(current_page.getURL(*page_args))
+path.append(current_page.get_url(*page_args))
 else:
-sub_path, current_page = self.getSubPageByName(
+sub_path, current_page = self.get_sub_page_by_name(
 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, "/".join(path))
+return self.host.check_redirection(self.vhost_root, "/".join(path))
-def getURLByPath(self, *args):
+def get_url_by_path(self, *args):
 """Generate URL by path
-this method as a similar effect as getURLByNames, but it is more readable
+this method as a similar effect as get_url_by_names, but it is more readable
 by using SubPage to get pages instead of using tuples
 @param *args: path element:
 - if unicode, will be used as argument
 - if util.SubPage instance, must be the name of a subpage
 @return (unicode): generated path
 # root page is the one needed to construct the base of the URL
 # if first arg is not a SubPage instance, we use current page
 if not isinstance(args[0], SubPage):
 root = self
 else:
-root = self.getPageByName(args.pop(0))
+root = self.get_page_by_name(args.pop(0))
 # we keep track of current page to check subpage
 current_page = root
 url_elts = []
 arguments = []
 while True:
 while args and not isinstance(args[0], SubPage):
 arguments.append(quote(args.pop(0)))
 if not url_elts:
-url_elts.append(root.getURL(*arguments))
+url_elts.append(root.get_url(*arguments))
 else:
 url_elts.extend(arguments)
 if not args:
 break
 else:
-path, current_page = current_page.getSubPageByName(args.pop(0))
+path, current_page = current_page.get_sub_page_by_name(args.pop(0))
 arguments = [path]
-return self.host.checkRedirection(self.vhost_root, "/".join(url_elts))
+return self.host.check_redirection(self.vhost_root, "/".join(url_elts))
 def getChildWithDefault(self, path, request):
 # we handle children ourselves
 raise exceptions.InternalError(
 "this method should not be used with LiberviaPage"
 )
-def nextPath(self, request):
+def next_path(self, request):
 """get next URL path segment, and update request accordingly
 will move first segment of postpath in prepath
 @param request(server.Request): current HTTP request
 @return (unicode): unquoted segment
 """
 pathElement = request.postpath.pop(0)
 request.prepath.append(pathElement)
 return urllib.parse.unquote(pathElement.decode('utf-8'))
-def _filterPathValue(self, value, handler, name, request):
+def _filter_path_value(self, value, handler, name, request):
-"""Modify a path value according to handler (see [getPathArgs])"""
+"""Modify a path value according to handler (see [get_path_args])"""
 if handler in ("@", "@jid") and value == "@":
 value = None
 if handler in ("", "@"):
 if value is None:
 if value:
 try:
 return jid.JID(value)
 except (RuntimeError, jid.InvalidFormat):
 log.warning(_("invalid jid argument: {value}").format(value=value))
-self.pageError(request, C.HTTP_BAD_REQUEST)
+self.page_error(request, C.HTTP_BAD_REQUEST)
 else:
 return ""
 else:
 return handler(self, value, name, request)
 return value
-def getPathArgs(self, request, names, min_args=0, **kwargs):
+def get_path_args(self, request, names, min_args=0, **kwargs):
 """get several path arguments at once
 Arguments will be put in request data.
 Missing arguments will have None value
 @param names(list[unicode]): list of arguments to get
 - 'jid': value must be converted to jid.JID if it exists, else empty
 string is used
 - '@jid': if value of arguments is empty or '@', empty string will be
 used, else it will be converted to jid
 """
-data = self.getRData(request)
+data = self.get_r_data(request)
 for idx, name in enumerate(names):
 if name[0] == "*":
 value = data[name[1:]] = []
 while True:
 try:
-value.append(self.nextPath(request))
+value.append(self.next_path(request))
 except IndexError:
 idx -= 1
 break
 else:
 idx += 1
 else:
 try:
-value = data[name] = self.nextPath(request)
+value = data[name] = self.next_path(request)
 except IndexError:
 data[name] = None
 idx -= 1
 break
 values_count = idx + 1
 if values_count < min_args:
 log.warning(_("Missing arguments in URL (got {count}, expected at least "
 "{min_args})").format(count=values_count, min_args=min_args))
-self.pageError(request, C.HTTP_BAD_REQUEST)
+self.page_error(request, C.HTTP_BAD_REQUEST)
 for name in names[values_count:]:
 data[name] = None
 for name, handler in kwargs.items():
 if name[0] == "*":
 data[name] = [
-self._filterPathValue(v, handler, name, request) for v in data[name]
+self._filter_path_value(v, handler, name, request) for v in data[name]
 ]
 else:
-data[name] = self._filterPathValue(data[name], handler, name, request)
+data[name] = self._filter_path_value(data[name], handler, name, request)
 ## Pagination/Filtering ##
-def getPubsubExtra(self, request, page_max=10, params=None, extra=None,
+def get_pubsub_extra(self, request, page_max=10, params=None, extra=None,
 order_by=C.ORDER_BY_CREATION):
 """Set extra dict to retrieve PubSub items corresponding to URL parameters
 Following parameters are used:
 - after: set rsm_after with ID of item
 - before: set rsm_before with ID of item
 @param request(server.Request): current HTTP request
 @param page_max(int): required number of items per page
 @param params(None, dict[unicode, list[unicode]]): params as returned by
-self.getAllPostedData.
+self.get_all_posted_data.
 None to parse URL automatically
 @param extra(None, dict): extra dict to use, or None to use a new one
 @param order_by(unicode, None): key to order by
 None to not specify order
 @return (dict): fill extra data
 """
 if params is None:
-params = self.getAllPostedData(request, multiple=False)
+params = self.get_all_posted_data(request, multiple=False)
 if extra is None:
 extra = {}
 else:
 assert not {"rsm_max", "rsm_after", "rsm_before",
 C.KEY_ORDER_BY}.intersection(list(extra.keys()))
 # RSM returns list in order (oldest first), but we want most recent first
 # so we start by the end
 extra['rsm_before'] = ""
 return extra
-def setPagination(self, request: server.Request, pubsub_data: dict) -> None:
+def set_pagination(self, request: server.Request, pubsub_data: dict) -> None:
 """Add to template_data if suitable
 "previous_page_url" and "next_page_url" will be added using respectively
 "before" and "after" URL parameters
 @param request: current HTTP request
 except KeyError:
 # no pagination available
 return
 # if we have a search query, we must keep it
-search = self.getPostedData(request, 'search', raise_on_missing=False)
+search = self.get_posted_data(request, 'search', raise_on_missing=False)
 if search is not None:
 extra['search'] = search.strip()
 # same for page_max
-page_max = self.getPostedData(request, 'page_max', raise_on_missing=False)
+page_max = self.get_posted_data(request, 'page_max', raise_on_missing=False)
 if page_max is not None:
 extra['page_max'] = page_max
 if rsm.get("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 = rsm["first"]
-template_data['previous_page_url'] = self.getParamURL(
+template_data['previous_page_url'] = self.get_param_url(
 request, before=first_id, **extra)
 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(
+template_data['next_page_url'] = self.get_param_url(
 request, after=last_id, **extra)
 ## Cache handling ##
-def _setCacheHeaders(self, request, cache):
+def _set_cache_headers(self, request, cache):
 """Set ETag and Last-Modified HTTP headers, used for caching"""
 request.setHeader("ETag", cache.hash)
-last_modified = self.host.getHTTPDate(cache.created)
+last_modified = self.host.get_http_date(cache.created)
 request.setHeader("Last-Modified", last_modified)
-def _checkCacheHeaders(self, request, cache):
+def _check_cache_headers(self, request, cache):
 """Check if a cache condition is set on the request
 if condition is valid, C.HTTP_NOT_MODIFIED is returned
 """
 etag_match = request.getHeader("If-None-Match")
 if etag_match is not None:
 if cache.hash == etag_match:
-self.pageError(request, C.HTTP_NOT_MODIFIED, no_body=True)
+self.page_error(request, C.HTTP_NOT_MODIFIED, no_body=True)
 else:
 modified_match = request.getHeader("If-Modified-Since")
 if modified_match is not None:
 modified = date_utils.date_parse(modified_match)
 if modified >= int(cache.created):
-self.pageError(request, C.HTTP_NOT_MODIFIED, no_body=True)
+self.page_error(request, C.HTTP_NOT_MODIFIED, no_body=True)
-def checkCacheSubscribeCb(self, sub_id, service, node):
+def check_cache_subscribe_cb(self, sub_id, service, node):
 self.cache_pubsub_sub.add((service, node, sub_id))
-def checkCacheSubscribeEb(self, failure_, service, node):
+def check_cache_subscribe_eb(self, failure_, service, node):
 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):
+def ps_node_watch_add_eb(self, failure_, service, node):
 log.warning(_("Can't add node watched: {msg}").format(msg=failure_))
-def useCache(self, request: server.Request) -> bool:
+def use_cache(self, request: server.Request) -> bool:
 """Indicate if the cache should be used
 test request header to see if it is requested to skip the cache
 @return: True if cache should be used
 """
 return request.getHeader('cache-control') != 'no-cache'
-def checkCache(self, request, cache_type, **kwargs):
+def check_cache(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
 pubsub node)
 @param request(server.Request): current HTTP request
 if request.uri != request.path:
 # we don't cache page with query arguments as there can be a lot of variants
 # influencing page results (e.g. search terms)
 log.debug("ignoring cache due to query arguments")
-no_cache = not self.useCache(request)
+no_cache = not self.use_cache(request)
-profile = self.getProfile(request) or C.SERVICE_PROFILE
+profile = self.get_profile(request) or C.SERVICE_PROFILE
 if cache_type == C.CACHE_PUBSUB:
 service, node = kwargs["service"], kwargs["node"]
 if not node:
 try:
 'registered'))
 return
 if profile != C.SERVICE_PROFILE:
 #  only service profile is cached for now
 return
-session_data = self.host.getSessionData(request, session_iface.IWebSession)
+session_data = self.host.get_session_data(request, session_iface.IWebSession)
 locale = session_data.locale
 if locale == C.DEFAULT_LOCALE:
 # no need to duplicate cache here
 locale = None
 try:
 cache = (self.cache[profile][cache_type][service][node]
 [self.vhost_root][request.uri][locale][self])
 except KeyError:
 # no cache yet, let's subscribe to the pubsub node
-d1 = self.host.bridgeCall(
+d1 = self.host.bridge_call(
-"psSubscribe", service.full(), node, "", profile
+"ps_subscribe", service.full(), node, "", profile
 )
-d1.addCallback(self.checkCacheSubscribeCb, service, node)
+d1.addCallback(self.check_cache_subscribe_cb, service, node)
-d1.addErrback(self.checkCacheSubscribeEb, service, node)
+d1.addErrback(self.check_cache_subscribe_eb, service, node)
-d2 = self.host.bridgeCall("psNodeWatchAdd", service.full(), node, profile)
+d2 = self.host.bridge_call("ps_node_watch_add", service.full(), node, profile)
-d2.addErrback(self.psNodeWatchAddEb, service, node)
+d2.addErrback(self.ps_node_watch_add_eb, service, node)
 self._do_cache = [self, profile, cache_type, service, node,
 self.vhost_root, request.uri, locale]
 #  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("Unknown cache_type")
 log.debug("using cache for {page}".format(page=self))
 cache.last_access = time.time()
-self._setCacheHeaders(request, cache)
+self._set_cache_headers(request, cache)
-self._checkCacheHeaders(request, cache)
+self._check_cache_headers(request, cache)
 request.write(cache.rendered)
 request.finish()
 raise failure.Failure(exceptions.CancelError("cache is used"))
-def _cacheURL(self, request, profile):
+def _cache_url(self, request, profile):
 self.cached_urls.setdefault(profile, {})[request.uri] = CacheURL(request)
 @classmethod
-def onNodeEvent(cls, host, service, node, event_type, items, profile):
+def on_node_event(cls, host, service, node, event_type, items, profile):
 """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(_(
 "Removing subscription for {service}/{node}: "
 "the page is not cached").format(service=service, node=node))
-d1 = host.bridgeCall("psUnsubscribe", service, node, profile)
+d1 = host.bridge_call("ps_unsubscribe", service, node, profile)
 d1.addErrback(
 lambda failure_: log.warning(
 _("Can't unsubscribe from {service}/{node}: {msg}").format(
 service=service, node=node, msg=failure_)))
-d2 = host.bridgeCall("psNodeWatchAdd", service, node, profile)
+d2 = host.bridge_call("ps_node_watch_add", service, node, profile)
 # TODO: check why the page is not in cache, remove subscription?
 d2.addErrback(
 lambda failure_: log.warning(
 _("Can't remove watch for {service}/{node}: {msg}").format(
 service=service, node=node, msg=failure_)))
 else:
 cache.clear()
 # identities
-async def fillMissingIdentities(
+async def fill_missing_identities(
 self,
 request: server.Request,
 entities: List[Union[str, jid.JID, None]],
 ) -> None:
 """Check if all entities have an identity cache, get missing ones from backend
 @param request: request with a plugged profile
 @param entities: entities to check, None or empty strings will be filtered
 """
 entities = {str(e) for e in entities if e}
-profile = self.getProfile(request) or C.SERVICE_PROFILE
+profile = self.get_profile(request) or C.SERVICE_PROFILE
-identities = self.host.getSessionData(
+identities = self.host.get_session_data(
 request,
 session_iface.IWebSession
 ).identities
 for e in entities:
 if e not in identities:
-id_raw = await self.host.bridgeCall(
+id_raw = await self.host.bridge_call(
-'identityGet', e, [], True, profile)
+'identity_get', e, [], True, profile)
 identities[e] = data_format.deserialise(id_raw)
 # signals, server => browser communication
-def delegateToResource(self, request, resource):
+def delegate_to_resource(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("resource delegation"))
-def HTTPRedirect(self, request, url):
+def http_redirect(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("HTTP redirection is used"))
-def redirectOrContinue(self, request, redirect_arg="redirect_url"):
+def redirect_or_continue(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
+@interrupt page_error(C.HTTP_BAD_REQUEST): empty or non local URL is used
 """
 redirect_arg = redirect_arg.encode('utf-8')
 try:
 url = request.args[redirect_arg][0].decode('utf-8')
 except (KeyError, IndexError):
 pass
 else:
 #  a redirection is requested
 if not url or url[0] != "/":
 # we only want local urls
-self.pageError(request, C.HTTP_BAD_REQUEST)
+self.page_error(request, C.HTTP_BAD_REQUEST)
 else:
-self.HTTPRedirect(request, url)
+self.http_redirect(request, url)
-def pageRedirect(self, page_path, request, skip_parse_url=True, path_args=None):
+def page_redirect(self, page_path, request, skip_parse_url=True, path_args=None):
 """redirect a page to a named page
 the workflow will continue with the workflow of the named page,
 skipping named page's parse_url method if it exist.
-If you want to do a HTTP redirection, use HTTPRedirect
+If you want to do a HTTP redirection, use http_redirect
 @param page_path(unicode): path to page (elements are separated by "/"):
 if path starts with a "/":
 path is a full path starting from root
 else:
 - first element is name as registered in name variable
 # if cache is needed, it will be handled by final page
 redirect_page._do_cache = self._do_cache
 self._do_cache = None
 defer.ensureDeferred(
-redirect_page.renderPage(request, skip_parse_url=skip_parse_url)
+redirect_page.render_page(request, skip_parse_url=skip_parse_url)
 )
 raise failure.Failure(exceptions.CancelError("page redirection is used"))
-def pageError(self, request, code=C.HTTP_NOT_FOUND, no_body=False):
+def page_error(self, request, code=C.HTTP_NOT_FOUND, no_body=False):
 """generate an error page and terminate the request
 @param request(server.Request): HTTP request
 @param core(int): error code to use
 @param no_body: don't write body if True
 if no_body:
 request.finish()
 else:
 template = "error/" + str(code) + ".html"
 template_data = request.template_data
-session_data = self.host.getSessionData(request, session_iface.IWebSession)
+session_data = self.host.get_session_data(request, session_iface.IWebSession)
 if session_data.locale is not None:
 template_data['locale'] = session_data.locale
 if self.vhost_root.site_name:
 template_data['site'] = self.vhost_root.site_name
 site_themes=self.site_themes,
 error_code=code,
 **template_data
 )
-self.writeData(rendered, request)
+self.write_data(rendered, request)
 raise failure.Failure(exceptions.CancelError("error page is used"))
-def writeData(self, data, request):
+def write_data(self, data, request):
 """write data to transport and finish the request"""
 if data is None:
-self.pageError(request)
+self.page_error(request)
 data_encoded = data.encode("utf-8")
 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)
+self._set_cache_headers(request, page_cache)
 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)
+self._check_cache_headers(request, page_cache)
 try:
 request.write(data_encoded)
 except AttributeError:
 log.warning(_("Can't write page, the request has probably been cancelled "
 "(browser tab closed or reloaded)"))
 return
 request.finish()
-def _subpagesHandler(self, request):
+def _subpages_handler(self, request):
 """render subpage if suitable
 this method checks if there is still an unmanaged part of the path
 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).encode('utf-8')
+subpage = self.next_path(request).encode('utf-8')
 try:
 child = self.children[subpage]
 except KeyError:
-self.pageError(request)
+self.page_error(request)
 else:
 child.render(request)
 raise failure.Failure(exceptions.CancelError("subpage page is used"))
 def _prepare_dynamic(self, request):
-session_data = self.host.getSessionData(request, session_iface.IWebSession)
+session_data = self.host.get_session_data(request, session_iface.IWebSession)
 # we need to activate dynamic page
 # we set data for template, and create/register token
 # socket_token = str(uuid.uuid4())
 socket_url = self.host.get_websocket_url(request)
 # as for CSRF, it is important to not let the socket token if we use the service
 # profile, as those pages can be cached, and then the token leaked.
 socket_token = '' if session_data.profile is None else session_data.ws_token
-socket_debug = C.boolConst(self.host.debug)
+socket_debug = C.bool_const(self.host.debug)
 request.template_data["websocket"] = WebsocketMeta(
 socket_url, socket_token, socket_debug
 )
 # we will keep track of handlers to remove
 request._signals_registered = []
 def _render_template(self, request):
 template_data = request.template_data
 # if confirm variable is set in case of successfuly data post
-session_data = self.host.getSessionData(request, session_iface.IWebSession)
+session_data = self.host.get_session_data(request, session_iface.IWebSession)
 template_data['identities'] = session_data.identities
-if session_data.popPageFlag(self, C.FLAG_CONFIRM):
+if session_data.pop_page_flag(self, C.FLAG_CONFIRM):
 template_data["confirm"] = True
-notifs = session_data.popPageNotifications(self)
+notifs = session_data.pop_page_notifications(self)
 if notifs:
 template_data["notifications"] = notifs
 if session_data.jid is not None:
 template_data["own_jid"] = session_data.jid
 if session_data.locale is not None:
 for key, value in data_common["template"].items():
 if key not in template_data:
 template_data[key] = value
 theme = session_data.theme or self.default_theme
-self.exposeToScripts(
+self.expose_to_scripts(
 request,
 cache_path=session_data.cache_dir,
-templates_root_url=str(self.vhost_root.getFrontURL(theme)),
+templates_root_url=str(self.vhost_root.get_front_url(theme)),
 profile=session_data.profile)
 uri = request.uri.decode()
 try:
 template_data["current_page"] = next(
 return self.host.renderer.render(
 self.template,
 theme=theme,
 site_themes=self.site_themes,
-page_url=self.getURL(),
+page_url=self.get_url(),
 media_path=f"/{C.MEDIA_DIR}",
 build_path=f"/{C.BUILD_DIR}/",
 cache_path=session_data.cache_dir,
 main_menu=self.main_menu,
 **template_data)
 else:
 ret = tuple(ret)
 raise NotImplementedError(
 _("iterable in on_data_post return value is not used yet")
 )
-session_data = self.host.getSessionData(request, session_iface.IWebSession)
+session_data = self.host.get_session_data(request, session_iface.IWebSession)
-request_data = self.getRData(request)
+request_data = self.get_r_data(request)
 if "post_redirect_page" in request_data:
 redirect_page_data = request_data["post_redirect_page"]
 if isinstance(redirect_page_data, tuple):
 redirect_page = redirect_page_data[0]
 redirect_page_args = redirect_page_data[1:]
-redirect_uri = redirect_page.getURL(*redirect_page_args)
+redirect_uri = redirect_page.get_url(*redirect_page_args)
 else:
 redirect_page = redirect_page_data
 redirect_uri = redirect_page.url
 else:
 redirect_page = self
 redirect_uri = request.uri
 if not C.POST_NO_CONFIRM in ret:
-session_data.setPageFlag(redirect_page, C.FLAG_CONFIRM)
+session_data.set_page_flag(redirect_page, C.FLAG_CONFIRM)
 request.setResponseCode(C.HTTP_SEE_OTHER)
 request.setHeader(b"location", redirect_uri)
 request.finish()
 raise failure.Failure(exceptions.CancelError("Post/Redirect/Get is used"))
 async def _on_data_post(self, request):
-self.checkCSRF(request)
+self.check_csrf(request)
 try:
-ret = await asDeferred(self.on_data_post, self, request)
+ret = await as_deferred(self.on_data_post, self, request)
 except exceptions.DataError as e:
 # something is wrong with the posted data, we re-display the page with a
 # warning notification
-session_data = self.host.getSessionData(request, session_iface.IWebSession)
+session_data = self.host.get_session_data(request, session_iface.IWebSession)
-session_data.setPageNotification(self, str(e), C.LVL_WARNING)
+session_data.set_page_notification(self, str(e), C.LVL_WARNING)
 request.setResponseCode(C.HTTP_SEE_OTHER)
 request.setHeader("location", request.uri)
 request.finish()
 raise failure.Failure(exceptions.CancelError("Post/Redirect/Get is used"))
 else:
 if ret != "continue":
 self._on_data_post_redirect(ret, request)
-def getPostedData(
+def get_posted_data(
 self,
 request: server.Request,
 keys,
 multiple: bool = False,
 raise_on_missing: bool = True,
 if len(keys) == 1:
 return ret[0]
 else:
 return ret
-def getAllPostedData(self, request, except_=(), multiple=True):
+def get_all_posted_data(self, request, except_=(), multiple=True):
 """get all posted data
 @param request(server.Request): request linked to the session
 @param except_(iterable[unicode]): key of values to ignore
 csrf_token will always be ignored
 ret[key] = urllib.parse.unquote(values[0])
 else:
 ret[key] = [urllib.parse.unquote(v) for v in values]
 return ret
-def getProfile(self, request):
+def get_profile(self, request):
 """Helper method to easily get current profile
 @return (unicode, None): current profile
 None if no profile session is started
 """
-web_session = self.host.getSessionData(request, session_iface.IWebSession)
+web_session = self.host.get_session_data(request, session_iface.IWebSession)
 return web_session.profile
-def getJid(self, request):
+def get_jid(self, request):
 """Helper method to easily get current jid
 @return: current jid
 """
-web_session = self.host.getSessionData(request, session_iface.IWebSession)
+web_session = self.host.get_session_data(request, session_iface.IWebSession)
 return web_session.jid
-def getRData(self, request):
+def get_r_data(self, request):
 """Helper method to get request data dict
 this dictionnary if for the request only, it is not saved in session
 It is mainly used to pass data between pages/methods called during request
 workflow
 return request.data
 except AttributeError:
 request.data = {}
 return request.data
-def getPageData(self, request, key):
+def get_page_data(self, request, key):
 """Helper method to retrieve reload resistant data"""
-web_session = self.host.getSessionData(request, session_iface.IWebSession)
+web_session = self.host.get_session_data(request, session_iface.IWebSession)
-return web_session.getPageData(self, key)
+return web_session.get_page_data(self, key)
-def setPageData(self, request, key, value):
+def set_page_data(self, request, key, value):
 """Helper method to set reload resistant data"""
-web_session = self.host.getSessionData(request, session_iface.IWebSession)
+web_session = self.host.get_session_data(request, session_iface.IWebSession)
-return web_session.setPageData(self, key, value)
+return web_session.set_page_data(self, key, value)
-def handleSearch(self, request, extra):
+def handle_search(self, request, extra):
 """Manage Full-Text Search
 Check if "search" query argument is present, and add MAM filter for it if
 necessary.
 If used, the "search" variable will also be available in template data, thus
 frontend can display some information about it.
 """
-search = self.getPostedData(request, 'search', raise_on_missing=False)
+search = self.get_posted_data(request, 'search', raise_on_missing=False)
 if search is not None:
 search = search.strip()
 if search:
 try:
 extra[f'mam_filter_{self.host.ns_map["fulltextmam"]}'] = search
 except KeyError:
 log.warning(_("Full-text search is not available"))
 else:
 request.template_data['search'] = search
-def _checkAccess(self, request):
+def _check_access(self, request):
 """Check access according to self.access
-if access is not granted, show a HTTP_FORBIDDEN pageError and stop request,
+if access is not granted, show a HTTP_FORBIDDEN page_error and stop request,
 else return data (so it can be inserted in deferred chain
 """
 if self.access == C.PAGES_ACCESS_PUBLIC:
 pass
 elif self.access == C.PAGES_ACCESS_PROFILE:
-profile = self.getProfile(request)
+profile = self.get_profile(request)
 if not profile:
 # registration allowed, we redirect to login page
-login_url = self.getPageRedirectURL(request)
+login_url = self.get_page_redirect_url(request)
-self.HTTPRedirect(request, login_url)
+self.http_redirect(request, login_url)
-def setBestLocale(self, request):
+def set_best_locale(self, request):
 """Guess the best locale when it is not specified explicitly by user
 This method will check "accept-language" header, and set locale to first
 matching value with available translations.
 """
 lang = lang.split(';')[0].strip().lower()
 if not lang:
 continue
 for a in available:
 if a.lower().startswith(lang):
-session_data = self.host.getSessionData(request,
+session_data = self.host.get_session_data(request,
 session_iface.IWebSession)
 session_data.locale = a
 return
-async def renderPage(self, request, skip_parse_url=False):
+async def render_page(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"):
 # if template_data doesn't exist, it's the beginning of the request workflow
 # so we fill essential data
-session_data = self.host.getSessionData(request, session_iface.IWebSession)
+session_data = self.host.get_session_data(request, session_iface.IWebSession)
 profile = session_data.profile
 request.template_data = {
 "profile": profile,
 # it's important to not add CSRF token and session uuid if service profile
 # is used because the page may be cached, and the token then leaked
 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:
-self.setBestLocale(request)
+self.set_best_locale(request)
 # theme
 key_theme = C.KEY_THEME.encode()
 if key_theme in request.args:
 theme = request.args.pop(key_theme)[0].decode()
 else:
 session_data.theme = theme
 try:
 try:
-self._checkAccess(request)
+self._check_access(request)
 if self.redirect is not None:
-self.pageRedirect(self.redirect, request, skip_parse_url=False)
+self.page_redirect(self.redirect, request, skip_parse_url=False)
 if self.parse_url is not None and not skip_parse_url:
 if self.url_cache:
-profile = self.getProfile(request)
+profile = self.get_profile(request)
 try:
 cache_url = self.cached_urls[profile][request.uri]
 except KeyError:
 # no cache for this URI yet
 #  we do normal URL parsing, and then the cache
-await asDeferred(self.parse_url, self, request)
+await as_deferred(self.parse_url, self, request)
-self._cacheURL(request, profile)
+self._cache_url(request, profile)
 else:
 log.debug(f"using URI cache for {self}")
 cache_url.use(request)
 else:
-await asDeferred(self.parse_url, self, request)
+await as_deferred(self.parse_url, self, request)
 if self.add_breadcrumb is None:
 label = (
 self.label
 or self.name
 "url": self.url,
 "label": label.title(),
 }
 request.template_data["breadcrumbs"].append(breadcrumb)
 else:
-await asDeferred(
+await as_deferred(
 self.add_breadcrumb,
 self,
 request,
 request.template_data["breadcrumbs"]
 )
-self._subpagesHandler(request)
+self._subpages_handler(request)
 if request.method not in (C.HTTP_METHOD_GET, C.HTTP_METHOD_POST):
 # only HTTP GET and POST are handled so far
-self.pageError(request, C.HTTP_BAD_REQUEST)
+self.page_error(request, C.HTTP_BAD_REQUEST)
 if request.method == C.HTTP_METHOD_POST:
 if self.on_data_post == 'continue':
 pass
 elif self.on_data_post is None:
 # if we don't have on_data_post, the page was not expecting POST
 # so we return an error
-self.pageError(request, C.HTTP_BAD_REQUEST)
+self.page_error(request, C.HTTP_BAD_REQUEST)
 else:
 await self._on_data_post(request)
 # by default, POST follow normal behaviour after on_data_post is called
 # this can be changed by a redirection or other method call in on_data_post
 if self.dynamic:
 self._prepare_dynamic(request)
 if self.prepare_render:
-await asDeferred(self.prepare_render, self, request)
+await as_deferred(self.prepare_render, self, request)
 if self.template:
 rendered = self._render_template(request)
 elif self.render_method:
-rendered = await asDeferred(self.render_method, self, request)
+rendered = await as_deferred(self.render_method, self, request)
 else:
 raise exceptions.InternalError(
 "No method set to render page, please set a template or use a "
 "render method"
 )
-self.writeData(rendered, request)
+self.write_data(rendered, request)
 except failure.Failure as f:
 # we have to unpack the Failure to catch the right Exception
 raise f.value
 except exceptions.CancelError:
 pass
 except BridgeException as e:
 if e.condition == 'not-allowed':
 log.warning("not allowed exception catched")
-self.pageError(request, C.HTTP_FORBIDDEN)
+self.page_error(request, C.HTTP_FORBIDDEN)
 elif e.condition == 'item-not-found' or e.classname == 'NotFound':
-self.pageError(request, C.HTTP_NOT_FOUND)
+self.page_error(request, C.HTTP_NOT_FOUND)
 elif e.condition == 'remote-server-not-found':
-self.pageError(request, C.HTTP_NOT_FOUND)
+self.page_error(request, C.HTTP_NOT_FOUND)
 elif e.condition == 'forbidden':
-if self.getProfile(request) is None:
+if self.get_profile(request) is None:
 log.debug("access forbidden, we're redirecting to log-in page")
-self.HTTPRedirect(request, self.getPageRedirectURL(request))
+self.http_redirect(request, self.get_page_redirect_url(request))
 else:
-self.pageError(request, C.HTTP_FORBIDDEN)
+self.page_error(request, C.HTTP_FORBIDDEN)
 else:
 log.error(
 _("Uncatched bridge exception for HTTP request on {url}: {e}\n"
 "page name: {name}\npath: {path}\nURL: {full_url}\n{tb}")
 .format(
 full_url=request.URLPath(),
 tb=traceback.format_exc(),
 )
 )
 try:
-self.pageError(request, C.HTTP_INTERNAL_ERROR)
+self.page_error(request, C.HTTP_INTERNAL_ERROR)
 except exceptions.CancelError:
 pass
 except Exception as e:
 log.error(
 _("Uncatched error for HTTP request on {url}: {e}\npage name: "
 full_url=request.URLPath(),
 tb=traceback.format_exc(),
 )
 )
 try:
-self.pageError(request, C.HTTP_INTERNAL_ERROR)
+self.page_error(request, C.HTTP_INTERNAL_ERROR)
 except exceptions.CancelError:
 pass
 def render_GET(self, request):
-defer.ensureDeferred(self.renderPage(request))
+defer.ensureDeferred(self.render_page(request))
 return server.NOT_DONE_YET
 def render_POST(self, request):
-defer.ensureDeferred(self.renderPage(request))
+defer.ensureDeferred(self.render_page(request))
 return server.NOT_DONE_YET

Mercurial > libervia-web

comparison libervia/server/pages.py @ 1509:106bae41f5c8