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('&', '&amp;'))
        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