Why Gemfury? Push, build, and install  RubyGems npm packages Python packages Maven artifacts PHP packages Go Modules Debian packages RPM packages NuGet packages

Repository URL to install this package:

Details    
workloadmgr / usr / lib / python3 / dist-packages / workloadmgr / api / versions.py
Size: Mime:
# vim: tabstop=4 shiftwidth=4 softtabstop=4

# Copyright (c) 2013 TrilioData, Inc.
# All Rights Reserved.

import datetime
import defusedxml.ElementTree as etree
from xml.etree.ElementTree import Element, SubElement, tostring

from workloadmgr.api import wsgi
from workloadmgr.api.views import versions as views_versions
from workloadmgr.api import xmlutil


VERSIONS = {
    "v1.0": {
        "id": "v1.0",
        "status": "CURRENT",
        "updated": "2012-01-04T11:33:21Z",
        "links": [
            {
                "rel": "describedby",
                "type": "application/pdf",
                "href": "http://jorgew.github.com/block-storage-api/"
                "content/os-block-storage-1.0.pdf",
            },
            {
                "rel": "describedby",
                "type": "application/vnd.sun.wadl+xml",
                # (anthony) FIXME
                "href": "http://docs.rackspacecloud.com/"
                "servers/api/v1.1/application.wadl",
            },
        ],
        "media-types": [
            {
                "base": "application/xml",
                "type": "application/vnd.openstack.volume+xml;version=1",
            },
            {
                "base": "application/json",
                "type": "application/vnd.openstack.volume+json;version=1",
            },
        ],
    }
}


class MediaTypesTemplateElement(xmlutil.TemplateElement):
    def will_render(self, datum):
        return "media-types" in datum


def make_version(elem):
    elem.set("id")
    elem.set("status")
    elem.set("updated")

    mts = MediaTypesTemplateElement("media-types")
    elem.append(mts)

    mt = xmlutil.SubTemplateElement(mts, "media-type", selector="media-types")
    mt.set("base")
    mt.set("type")

    xmlutil.make_links(elem, "links")


version_nsmap = {None: xmlutil.XMLNS_COMMON_V10, "atom": xmlutil.XMLNS_ATOM}


class VersionTemplate(xmlutil.TemplateBuilder):
    def construct(self):
        root = xmlutil.TemplateElement("version", selector="version")
        make_version(root)
        return xmlutil.MasterTemplate(root, 1, nsmap=version_nsmap)


class VersionsTemplate(xmlutil.TemplateBuilder):
    def construct(self):
        root = xmlutil.TemplateElement("versions")
        elem = xmlutil.SubTemplateElement(root, "version", selector="versions")
        make_version(elem)
        return xmlutil.MasterTemplate(root, 1, nsmap=version_nsmap)


class ChoicesTemplate(xmlutil.TemplateBuilder):
    def construct(self):
        root = xmlutil.TemplateElement("choices")
        elem = xmlutil.SubTemplateElement(root, "version", selector="choices")
        make_version(elem)
        return xmlutil.MasterTemplate(root, 1, nsmap=version_nsmap)


class AtomSerializer(wsgi.XMLDictSerializer):

    NSMAP = {None: xmlutil.XMLNS_ATOM}

    def __init__(self, metadata=None, xmlns=None):
        self.metadata = metadata or {}
        if not xmlns:
            self.xmlns = wsgi.XMLNS_ATOM
        else:
            self.xmlns = xmlns

    def _get_most_recent_update(self, versions):
        recent = None
        for version in versions:
            updated = datetime.datetime.strptime(
                version["updated"], "%Y-%m-%dT%H:%M:%SZ"
            )
            if not recent:
                recent = updated
            elif updated > recent:
                recent = updated

        return recent.strftime("%Y-%m-%dT%H:%M:%SZ")

    def _get_base_url(self, link_href):
        # Make sure no trailing /
        link_href = link_href.rstrip("/")
        return link_href.rsplit("/", 1)[0] + "/"

    def _create_feed(self, versions, feed_title, feed_id):
        feed = Element("feed", nsmap=self.NSMAP)
        title = SubElement(feed, "title")
        title.set("type", "text")
        title.text = feed_title

        # Set this updated to the most recently updated version
        recent = self._get_most_recent_update(versions)
        SubElement(feed, "updated").text = recent

        SubElement(feed, "id").text = feed_id

        link = SubElement(feed, "link")
        link.set("rel", "self")
        link.set("href", feed_id)

        author = SubElement(feed, "author")
        SubElement(author, "name").text = "Rackspace"
        SubElement(author, "uri").text = "http://www.rackspace.com/"

        for version in versions:
            feed.append(self._create_version_entry(version))

        return feed

    def _create_version_entry(self, version):
        entry = Element("entry")
        SubElement(entry, "id").text = version["links"][0]["href"]
        title = SubElement(entry, "title")
        title.set("type", "text")
        title.text = "Version %s" % version["id"]
        SubElement(entry, "updated").text = version["updated"]

        for link in version["links"]:
            link_elem = SubElement(entry, "link")
            link_elem.set("rel", link["rel"])
            link_elem.set("href", link["href"])
            if "type" in link:
                link_elem.set("type", link["type"])

        content = SubElement(entry, "content")
        content.set("type", "text")
        content.text = "Version %s %s (%s)" % (
            version["id"],
            version["status"],
            version["updated"],
        )
        return entry


class VersionsAtomSerializer(AtomSerializer):
    def default(self, data):
        versions = data["versions"]
        feed_id = self._get_base_url(versions[0]["links"][0]["href"])
        feed = self._create_feed(versions, "Available API Versions", feed_id)
        return self._to_xml(feed)


class VersionAtomSerializer(AtomSerializer):
    def default(self, data):
        version = data["version"]
        feed_id = version["links"][0]["href"]
        feed = self._create_feed([version], "About This Version", feed_id)
        return self._to_xml(feed)


class Versions(wsgi.Resource):
    def __init__(self):
        super(Versions, self).__init__(None)

    @wsgi.serializers(xml=VersionsTemplate, atom=VersionsAtomSerializer)
    def index(self, req):
        """Return all versions."""
        builder = views_versions.get_view_builder(req)
        return builder.build_versions(VERSIONS)

    @wsgi.serializers(xml=ChoicesTemplate)
    @wsgi.response(300)
    def multi(self, req):
        """Return multiple choices."""
        builder = views_versions.get_view_builder(req)
        return builder.build_choices(VERSIONS, req)

    def get_action_args(self, request_environment):
        """Parse dictionary created by routes library."""
        args = {}
        if request_environment["PATH_INFO"] == "/":
            args["action"] = "index"
        else:
            args["action"] = "multi"

        return args


class WorkloadVersionV1(object):
    @wsgi.serializers(xml=VersionTemplate, atom=VersionAtomSerializer)
    def show(self, req):
        builder = views_versions.get_view_builder(req)
        return builder.build_version(VERSIONS["v1.0"])


def create_resource():
    return wsgi.Resource(WorkloadVersionV1())