web/views/urlpublishing.py
author Sylvain Thénault <sylvain.thenault@logilab.fr>
Fri, 19 Feb 2010 09:34:14 +0100
branchstable
changeset 4643 921737d2e3a8
parent 4252 6c4f109c2b03
child 5174 78438ad513ca
child 5301 f4219a6e62e3
permissions -rw-r--r--
fix optimisation with super session that may lead to integrity loss at some point I've decided to stop ensuring ?1 cardinality was respected when adding a new relation using a super session, to avoid the cost of the delete query. That was yet discussable because it introduced unexpected difference between execute and unsafe_execute, which is imo not worth it. Also, now that rql() in migration script default to unsafe_execute, we definitly don't want that implicit behaviour change (which already cause bug when for instance adding another default workflow for an entity type: without that fix we end up with *two* default workflows while the schema tells we can have only one. IMO we should go to the direction that super session skip all security check, but nothing else, unless explicitly asked.

"""associate url's path to view identifier / rql queries

It currently handle url's path with the forms

* <publishing_method>

* minimal REST publishing:
  * <eid>
  * <etype>[/<attribute name>/<attribute value>]*

* folder navigation


You can actually control URL (more exactly path) resolution using URL path
evaluator.

XXX actionpath and folderpath execute a query whose results is lost
because of redirecting instead of direct traversal

:organization: Logilab
:copyright: 2001-2010 LOGILAB S.A. (Paris, FRANCE), license is LGPL v2.
:contact: http://www.logilab.fr/ -- mailto:contact@logilab.fr
:license: GNU Lesser General Public License, v2.1 - http://www.gnu.org/licenses
"""

__docformat__ = "restructuredtext en"

from rql import TypeResolverException

from cubicweb import RegistryException, typed_eid
from cubicweb.web import NotFound, Redirect, component


class PathDontMatch(Exception):
    """exception used by url evaluators to notify they can't evaluate
    a path
    """

class URLPublisherComponent(component.Component):
    """associate url's path to view identifier / rql queries,
    by applying a chain of urlpathevaluator components.

    An evaluator is a URLPathEvaluator subclass with a .evaluate_path
    method taking the request object and the path to publish as
    argument.  It will either returns a publishing method identifier
    and a rql query on success or raises a `PathDontMatch` exception
    on failure. URL evaluators are called according to their `priority`
    attribute, with 0 as the greatest priority and greater values as
    lower priority.  The first evaluator returning a result or raising
    something else than `PathDontMatch` will stop the handlers chain.
    """
    __regid__ = 'urlpublisher'
    vreg = None # XXX necessary until property for deprecation warning is on appobject

    def __init__(self, vreg, default_method='view'):
        super(URLPublisherComponent, self).__init__()
        self.vreg = vreg
        self.default_method = default_method
        evaluators = []
        for evaluatorcls in vreg['components']['urlpathevaluator']:
            # instantiation needed
            evaluator = evaluatorcls(self)
            evaluators.append(evaluator)
        self.evaluators = sorted(evaluators, key=lambda x: x.priority)

    def process(self, req, path):
        """given an url (essentialy caracterized by a path on the server,
        but additional information may be found in the request object), return
        a publishing method identifier (eg controller) and an optional result
        set

        :type req: `cubicweb.web.Request`
        :param req: the request object

        :type path: str
        :param path: the path of the resource to publish

        :rtype: tuple(str, `cubicweb.utils.ResultSet` or None)
        :return: the publishing method identifier and an optional result set

        :raise NotFound: if no handler is able to decode the given path
        """
        parts = [part for part in path.split('/')
                 if part != ''] or (self.default_method,)
        if req.form.get('rql'):
            if parts[0] in self.vreg['controllers']:
                return parts[0], None
            return 'view', None
        for evaluator in self.evaluators:
            try:
                pmid, rset = evaluator.evaluate_path(req, parts[:])
                break
            except PathDontMatch:
                continue
        else:
            raise NotFound(path)
        if pmid is None:
            pmid = self.default_method
        return pmid, rset


class URLPathEvaluator(component.Component):
    __abstract__ = True
    __regid__ = 'urlpathevaluator'
    vreg = None # XXX necessary until property for deprecation warning is on appobject

    def __init__(self, urlpublisher):
        self.urlpublisher = urlpublisher
        self.vreg = urlpublisher.vreg

class RawPathEvaluator(URLPathEvaluator):
    """handle path of the form::

        <publishing_method>?parameters...
    """
    priority = 0
    def evaluate_path(self, req, parts):
        if len(parts) == 1 and parts[0] in self.vreg['controllers']:
            return parts[0], None
        raise PathDontMatch()


class EidPathEvaluator(URLPathEvaluator):
    """handle path with the form::

        <eid>
    """
    priority = 1
    def evaluate_path(self, req, parts):
        if len(parts) != 1:
            raise PathDontMatch()
        try:
            rset = req.execute('Any X WHERE X eid %(x)s',
                               {'x': typed_eid(parts[0])}, 'x')
        except ValueError:
            raise PathDontMatch()
        if rset.rowcount == 0:
            raise NotFound()
        return None, rset


class RestPathEvaluator(URLPathEvaluator):
    """handle path with the form::

        <etype>[[/<attribute name>]/<attribute value>]*
    """
    priority = 2

    def evaluate_path(self, req, parts):
        if not (0 < len(parts) < 4):
            raise PathDontMatch()
        try:
            etype = self.vreg.case_insensitive_etypes[parts.pop(0).lower()]
        except KeyError:
            raise PathDontMatch()
        cls = self.vreg['etypes'].etype_class(etype)
        if parts:
            if len(parts) == 2:
                attrname = parts.pop(0).lower()
                try:
                    cls.e_schema.subjrels[attrname]
                except KeyError:
                    raise PathDontMatch()
            else:
                attrname = cls._rest_attr_info()[0]
            value = req.url_unquote(parts.pop(0))
            rset = self.attr_rset(req, etype, attrname, value)
        else:
            rset = self.cls_rset(req, cls)
        if rset.rowcount == 0:
            raise NotFound()
        return None, rset

    def cls_rset(self, req, cls):
        return req.execute(cls.fetch_rql(req.user))

    def attr_rset(self, req, etype, attrname, value):
        rql = u'Any X WHERE X is %s, X %s %%(x)s' % (etype, attrname)
        if attrname == 'eid':
            try:
                rset = req.execute(rql, {'x': typed_eid(value)}, 'x')
            except (ValueError, TypeResolverException):
                # conflicting eid/type
                raise PathDontMatch()
        else:
            rset = req.execute(rql, {'x': value})
        return rset


class URLRewriteEvaluator(URLPathEvaluator):
    """tries to find a rewrite rule to apply

    URL rewrite rule definitions are stored in URLRewriter objects
    """
    priority = 3
    def evaluate_path(self, req, parts):
        # uri <=> req._twreq.path or req._twreq.uri
        uri = req.url_unquote('/' + '/'.join(parts))
        evaluators = sorted(self.vreg['urlrewriting'].all_objects(),
                            key=lambda x: x.priority, reverse=True)
        for rewritercls in evaluators:
            rewriter = rewritercls(req)
            try:
                # XXX we might want to chain url rewrites
                return rewriter.rewrite(req, uri)
            except KeyError:
                continue
        raise PathDontMatch()


class ActionPathEvaluator(URLPathEvaluator):
    """handle path with the form::

    <any evaluator path>/<action>
    """
    priority = 4
    def evaluate_path(self, req, parts):
        if len(parts) < 2:
            raise PathDontMatch()
        # remove last part and see if this is something like an actions
        # if so, call
        # XXX bad smell: refactor to simpler code
        try:
            actionsreg = self.vreg['actions']
            requested = parts.pop(-1)
            actions = actionsreg[requested]
        except RegistryException:
            raise PathDontMatch()
        for evaluator in self.urlpublisher.evaluators:
            if evaluator is self or evaluator.priority == 0:
                continue
            try:
                pmid, rset = evaluator.evaluate_path(req, parts[:])
            except PathDontMatch:
                continue
            else:
                try:
                    action = actionsreg._select_best(actions, req, rset=rset)
                except RegistryException:
                    continue
                else:
                    # XXX avoid redirect
                    raise Redirect(action.url())
        raise PathDontMatch()