1. Armin Ronacher
  2. heechee


heechee / heechee / webdav.py

from lxml.etree import fromstring, tostring, Element, SubElement
from werkzeug import Request, Response, cached_property
from werkzeug.exceptions import HTTPException, MethodNotAllowed, BadRequest

from heechee.svndiff import make_cheap_diff
from heechee.repo import Repository, File, Directory

DAV_NS = "{DAV:}"
SVN_NS = "{svn:}"
SVN_DAV_NS = "{http://subversion.tigris.org/xmlns/dav/}"

class DAVRequest(Request):

    def xml_body(self):
        # would be nicer to have fromstring(self.data) but that requires
        # an unreleased version of Werkzeug.
        return fromstring(self.input_stream.read(self.content_length))

class Application(object):
    verbs = frozenset(["OPTIONS", "PROPFIND", "REPORT"])
    def __init__(self, repo):
        self.repo = repo
    def __call__(self, request):
            if request.method not in self.verbs:
                raise MethodNotAllowed(self.verbs)
            return getattr(self, request.method)(request)
        except HTTPException, e:
            return e
    def OPTIONS(self, request):
        Look, Ma, we support ALL THESE methods. Really.
        return Response(headers = {
            "Allow": ",".join(self.verbs),
    def PROPFIND(self, request):
        Responds to SVN's PROPFIND requests, which usually detail finding out
        various things about the repo.
        # Do they want the properties of the overall repo?
        tree = request.xml_body
        props = [x.tag.split("}")[-1] for x in tree.find("{DAV:}prop").getchildren()]
        if request.path == "/":
            response = self._propstat_response(request, {
                DAV_NS+"version-controlled-configuration": self._href("/!svn/vcc/default"),
                DAV_NS+"resourcetype": Element(DAV_NS+"collection"),
                SVN_DAV_NS+"baseline-relative-path": None,
                SVN_DAV_NS+"repository-uuid": self.repo.uuid,
        elif request.path == "/!svn/vcc/default":
            answers = {}
            if "checked-in" in props:
                answers[DAV_NS+"checked-in"] = self._href("/!svn/bln/1/")
            if "baseline-collection" in props:
                answers[DAV_NS+"baseline-collection"] = self._href("/!svn/bc/1/")
            if "version-name" in props:
                answers[DAV_NS+"version-name"] = "1"
            response = self._propstat_response(request, answers)
        elif request.path == "/!svn/bln/1":
            response = self._propstat_response(request, {
                DAV_NS+"baseline-collection": self._href("/!svn/bc/1/"),
                DAV_NS+"version-name": '1',
        elif request.path == "/!svn/bc/1":
            response = self._propstat_response(request, {
                DAV_NS+"version-controlled-configuration": self._href("/!svn/vcc/default"),
                DAV_NS+"resourcetype": Element(DAV_NS+"collection"),
                SVN_NS+"repository-uuid": self.repo.uuid,
        # We don't recognise it! How unlikely.
            raise BadRequest()
        return Response(response, status = 207)
    def REPORT(self, request):
        "i.e. SVN-wants-a-massive-diff time."
        up_rep = Element(SVN_NS+"update-report")
        up_rep.attrib['send-all'] = "true"
        rev = "1"
        SubElement(up_rep, SVN_NS+"target-revision", rev=rev)
        # The root directory is special.
        root_directory = self.repo.get_rev(1)
        dir = SubElement(up_rep, SVN_NS+"open-directory", rev=rev)
        dir_ch_in = SubElement(dir, DAV_NS+"checked-in")
        dir_ch_in.append(self._href("/!svn/ver/%s/" % rev))
        # The queue is a list of (parent, item) tuples.
        queue = [(dir, child) for child in root_directory.get_children()]
        while queue:
            parent, item = queue.pop()
            if isinstance(item, Directory):
                # Make the directory's entry
                dir = SubElement(parent, SVN_NS+"add-directory", name=item.name, rev="1")
                dir_ch_in = SubElement(dir, DAV_NS+"checked-in")
                dir_ch_in.append(self._href("/!svn/ver/%s%s" % (rev, item.get_path())))
                # Add its children
                for child in item.get_children():
                    queue.append((dir, child))
            elif isinstance(item, File):
                # Make the file object
                file = SubElement(parent, SVN_NS+"add-file", name=item.name)
                dir_ch_in = SubElement(file, DAV_NS+"checked-in")
                dir_ch_in.append(self._href("/!svn/ver/%s%s" % (rev, item.get_path())))
                rev_elem = SubElement(file, SVN_NS+"set-prop", name="svn:entry:committed-rev")
                rev_elem.text = str(rev)
                for window in make_cheap_diff(item.contents):
                    txdelta = SubElement(file, SVN_NS+"txdelta")
                    txdelta.text = window.encode("base64")
        return Response(tostring(up_rep))
    def _href(self, target):
        "Single-call making of DAV:href tags."
        elem = Element(DAV_NS+"href")
        elem.text = target
        return elem
    def _propstat_response(self, request, props, status="HTTP/1.1 200 OK"):
        "Creates a multi-status response."
        multi = Element(DAV_NS + "multistatus")
        response = SubElement(multi, DAV_NS + "response")
        # Path of this request
        href = SubElement(response, DAV_NS + "href")
        href.text = request.path
        # It's gonna be a propstat
        propstat = SubElement(response, DAV_NS + "propstat")
        prop = SubElement(propstat, DAV_NS + "prop")
        # Add in the props we're told about
        for tag, value in props.items():
            elem = SubElement(prop, tag)
            # Is it a string value?
            if isinstance(value, basestring):
                elem.text = value
            # Is it None? (do nothing)
            elif value is None:
            # Then it's hopefully an element.
        # Finally, add in the status
        SubElement(propstat, DAV_NS + "status").text = status
        return tostring(multi)

if __name__ == "__main__":
    from werkzeug import run_simple
    run_simple('localhost', 8080, Application(Repository()))