Repository URL to install this package:
Version:
5.0.6.dev17 ▾
|
# 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())