Change an ATOM entry on the catalog¶
In this scenario, we download an ATOM entry that exists on the catalog, change it locally and and re-upload the changed entry. This can be useful when an automatically generated entry (see post-atom.ipynb) needs to be amended.
1. Set the necessary variables¶
The following section defines all the necessary information as variables so the code below can be easily reused.
[ ]:
import getpass
# Set the credentials (Ellip username and API key)
username = raw_input("What is your Ellip username? ")
api_key = getpass.getpass("What is your Ellip API key? ")
# Set the name of the destination index on the Terradue catalog
index_name = raw_input("What is the destination index name? (press Enter to confirm default [{0}]) ".format(username))
if not index_name:
index_name = username
# Set the catalog endpoint URL
endpoint = "https://catalog.terradue.com/{0}".format(index_name)
# Set the identifier of the entry to be changed
uid = 'data-publication-sample'
2. Define a class for ATOM manipulation¶
[ ]:
import lxml.etree as etree
import sys
import os
import string
import hashlib
import urllib2
import base64
import time
class Atom:
tree = None
root = None
entry = None
def __init__(self, root):
self.root = root
self.tree = root
self.links = self.root.xpath('/a:feed/a:entry/a:link', namespaces={'a':'http://www.w3.org/2005/Atom'})
entries = self.root.xpath('/a:feed/a:entry', namespaces={'a':'http://www.w3.org/2005/Atom'})
if len(entries) > 0:
self.entry = entries[0]
@staticmethod
def from_template():
template = """<?xml version="1.0"?>
<feed xmlns="http://www.w3.org/2005/Atom">
<entry>
<title type="text"></title>
<summary type="html"></summary>
<link rel="enclosure" type="application/octet-stream" href=""/>
<date xmlns="http://purl.org/dc/elements/1.1/"></date>
<published></published>
<identifier xmlns="http://purl.org/dc/elements/1.1/"></identifier>
</entry>
</feed>"""
tree = etree.fromstring(template)
return Atom(tree)
@staticmethod
def load(url, username=None, api_key=None):
"""Load and return the atom file at the location url
"""
request = urllib2.Request(url)
if ( username != None ):
base64string = base64.b64encode('%s:%s' % (username, api_key))
request.add_header("Authorization", "Basic %s" % base64string)
fp = urllib2.urlopen(request)
tree = etree.parse(fp)
fp.close()
if ( tree.getroot().tag != "{http://www.w3.org/2005/Atom}feed" ):
raise ValueError('not an Atom feed')
return Atom(tree)
def set_identifier(self, identifier):
"""Set first atom entry identifier
"""
el_identifier = self.root.xpath('/a:feed/a:entry/d:identifier',
namespaces={'a':'http://www.w3.org/2005/Atom',
'd':'http://purl.org/dc/elements/1.1/'})
el_identifier[0].text = identifier
def get_identifier(self):
el_identifier = self.root.xpath('/a:feed/a:entry/d:identifier',
namespaces={'a':'http://www.w3.org/2005/Atom',
'd':'http://purl.org/dc/elements/1.1/'})
if (len(el_identifier) == 0):
return None
return el_identifier[0].text;
def get_total_results(self, create=False):
# get OS total results in feed
totalResults = self.root.xpath('/a:feed/os:totalResults', namespaces={'a':'http://www.w3.org/2005/Atom', 'os':'http://a9.com/-/spec/opensearch/1.1/'})
if (len(totalResults) == 0):
return None
return int(totalResults[0].text)
def get_title(self, create=False):
# get or create title
titles = self.root.xpath('/a:feed/a:entry/a:title', namespaces={'a':'http://www.w3.org/2005/Atom'})
if (len(titles) == 0):
if (create):
titles = [etree.SubElement(self.entry, "{http://www.w3.org/2005/Atom}title")]
return titles[0]
return None
return titles[0]
def set_title_text(self, text):
"""Set first atom entry title
"""
el_title = self.root.xpath('/a:feed/a:entry/a:title',
namespaces={'a':'http://www.w3.org/2005/Atom'})
el_title[0].text = text
def get_summary(self, create=False):
# get or create summary
summaries = self.root.xpath('/a:feed/a:entry/a:summary', namespaces={'a':'http://www.w3.org/2005/Atom'})
if (len(summaries) == 0):
if (create):
summaries = [etree.SubElement(self.entry, "{http://www.w3.org/2005/Atom}summary")]
return summaries[0]
return None
return summaries[0]
def set_summary_text(self, text):
# get or create summary
summary = self.get_summary(True)
summary.text = text
def get_links(self, rel_type):
# get links
return self.root.xpath('/a:feed/a:entry/a:link[@rel = "{0}"]'.format(rel_type), namespaces={'a':'http://www.w3.org/2005/Atom'})
def set_enclosure_link(self, href, title):
el_enclosure_link = self.root.xpath('/a:feed/a:entry/a:link[@rel="enclosure" and (@href="" or @href="{0}")]'.format(href),
namespaces={'a':'http://www.w3.org/2005/Atom'})
if (len(el_enclosure_link) > 0):
link = el_enclosure_link[0]
link.attrib['href'] = href
else:
link = self.add_enclosure_link(href, title)
def add_enclosure_link(self, href, title):
xml_string = '<link rel="enclosure" type="application/octet-stream" title="%s" href="%s"/>' % (title, href.replace('&', '&'))
print(xml_string)
link = etree.fromstring(xml_string)
self.entry.append(link)
return link
def add_extension(self, xml_ext):
el_entry = self.root.xpath('/a:feed/a:entry/a:link',
namespaces={'a':'http://www.w3.org/2005/Atom'})
el_entry[0].addnext(xml_ext)
def add_link(self, href, rel, title=None, type=None):
link = etree.SubElement(self.root.xpath('/a:feed/a:entry',
namespaces={'a':'http://www.w3.org/2005/Atom'})[0], "{http://www.w3.org/2005/Atom}link")
link.attrib['href'] = href
link.attrib['rel'] = rel
if title:
link.attrib['title'] = title
if type:
link.attrib['type'] = type
def remove_link(self, rel, link_title=None, link_type=None, link_url=None):
links = self.get_links(rel)
filter = None
value = None
if link_title:
filter = 'title'
value = link_title
elif link_type:
filter = 'type'
value = link_type
elif link_url:
filter = 'url'
value = link_url
else:
raise Exception("Required parameter link_title, link_type or link_url")
for link in links:
if link.attrib[filter] == value:
link.getparent().remove(link)
def get_offering_elements(self, offering_code):
return self.root.xpath('/a:feed/a:entry/b:offering[@code="{0}"]'.format(offering_code),
namespaces={'a':'http://www.w3.org/2005/Atom',
'b':'http://www.opengis.net/owc/1.0'})
@staticmethod
def get_operation_elements(offering_element, operation_code=None):
xpath = 'b:operation'
if (operation_code):
xpath += '[@code="{0}"]'.format(operation_code)
return offering_element.xpath(xpath, namespaces={'b':'http://www.opengis.net/owc/1.0'})
def add_offering(self, offering):
self.root.xpath('/a:feed/a:entry', namespaces={'a':'http://www.w3.org/2005/Atom'})[0].append(offering)
def add_offerings(self, offerings):
for offering in offerings:
self.add_offering(offering)
def get_dctspatial(self, create=False):
# get or create summary
spatials = self.root.xpath('/a:feed/a:entry/c:spatial',
namespaces={'a':'http://www.w3.org/2005/Atom',
'c':'http://purl.org/dc/terms/'})
if (len(spatials) == 0):
if (create):
spatials = [etree.SubElement(self.entry, "{http://purl.org/dc/terms/}spatial")]
return spatials[0]
return None
return spatials[0]
def set_dctspatial(self, wkt):
el_spatial = self.get_dctspatial(True)
el_spatial.text = wkt
def get_dcdate(self, create):
# get or create dcdate
el_dates = self.root.xpath('/a:feed/a:entry/d:date',
namespaces={'a':'http://www.w3.org/2005/Atom',
'd':'http://purl.org/dc/elements/1.1/'})
if (len(el_dates) == 0):
if (create):
el_dates = [etree.SubElement(self.entry, "{http://purl.org/dc/elements/1.1/}date")]
return el_dates[0]
return None
return el_dates[0]
def set_dcdate(self, date):
# get or create dcdate
dcdate = self.get_dcdate(True)
dcdate.text = date
def set_published(self, published):
el_published = self.root.xpath('/a:feed/a:entry/a:published',
namespaces={'a':'http://www.w3.org/2005/Atom'})
el_published[0].text = published
def get_category_by_scheme(self, scheme):
categories = self.root.xpath('/a:feed/a:entry/a:category[@scheme="{0}"]'.format(scheme), namespaces={'a':'http://www.w3.org/2005/Atom'})
if (len(categories) == 0):
return None
return categories[0]
def get_categories(self, term, scheme=None):
# get categories
filter = '@term="{0}"'.format(term)
if scheme != None:
filter = '{0} and @scheme="{1}"'.format(filter, scheme)
return self.root.xpath('/a:feed/a:entry/a:category[{0}]'.format(filter), namespaces={'a':'http://www.w3.org/2005/Atom'})
def remove_category(self, term, scheme=None):
# get and remove category
for category in self.get_categories(term, scheme):
category.getparent().remove(category)
def remove_category_by_scheme(self, scheme):
# get categories
filter = '@scheme="{0}"'.format(scheme)
categories = self.root.xpath('/a:feed/a:entry/a:category[{0}]'.format(filter), namespaces={'a':'http://www.w3.org/2005/Atom'})
for category in categories:
category.getparent().remove(category)
def set_category(self, term, label=None, scheme=None):
categories = self.get_categories(term, scheme)
if (len(categories) == 0):
categories = [etree.SubElement(self.entry, "{http://www.w3.org/2005/Atom}category")]
categories[0].attrib['term'] = term
if label != None:
categories[0].attrib['label'] = label
if scheme != None:
categories[0].attrib['scheme'] = scheme
def set_generator(self, uri, version, text):
# get or create generator
el_generator = self.root.xpath('/a:feed/a:entry/a:generator', namespaces={'a':'http://www.w3.org/2005/Atom'})
if (len(el_generator) == 0):
el_generator = [etree.SubElement(self.root.xpath('/a:feed/a:entry',
namespaces={'a':'http://www.w3.org/2005/Atom'})[0], "{http://www.w3.org/2005/Atom}generator")]
el_generator[0].attrib['uri'] = uri
el_generator[0].attrib['version'] = version
el_generator[0].text = text
def append_summary_html(self, text):
"""Append atom summary with text
"""
html_summary = self.get_summary(True).text
html_summary += "<p>%s</p>" % text
self.set_summary_text(html_summary)
def to_string(self, pretty_print = True):
return etree.tostring(self.tree, pretty_print=pretty_print)
def clear_enclosures(self):
links = self.get_links("enclosure")
for link in links:
link.getparent().remove(link)
def get_extensions(self, name, namespace):
return self.root.xpath('/a:feed/a:entry/e:{0}'.format(name),
namespaces={'a':'http://www.w3.org/2005/Atom',
'e':namespace})
3. Load the ATOM feed of a product and change its entry¶
We download the feed containing one entry and change the title of the entry.
[ ]:
import datetime
atom = Atom.load("{0}/search?q={1}".format(endpoint, uid), username, api_key);
# Change a property (the title)
atom.set_title_text("---- CHANGED ----- Title for data-publication-sample")
4. Post ATOM feed¶
We post the ATOM feed back to the same index on the catalog.
[ ]:
import requests
request = requests.post(endpoint,
data=atom.to_string(),
headers={"Content-Type": "application/atom+xml", "Accept": "application/xml"},
auth=(username, api_key))
if request.status_code == 200:
print('Data item updated at {0}/search?uid={1}&apikey={2} ({3})'.format(endpoint, atom.get_identifier(), api_key, str(request.status_code)))
else:
print('Data item NOT updated at {0}/search?uid={1}&apikey={2} ({3})'.format(endpoint, atom.get_identifier(), api_key, str(request.status_code)))
If a product URL with status 200 is displayed, the ATOM feed has been successfully updated and the title has been changed.
END