Merge pull request #377 from SynBioDex/364-downloadAttachment

Implement downloadAttachment

Author: bbartley

sbol2/partshop.py

@@ -344,6 +344,55 @@ def attachFile(self, top_level_uri, filepath):
         msg = msg.format(response.status_code)
         raise SBOLError(SBOLErrorCode.SBOL_ERROR_BAD_HTTP_REQUEST, msg)
 
+    def downloadAttachment(self, attachment_uri, filepath='.'):
+        '''Download a file attachment from SynBioHub.
+        :param attachment_uri: The URI of the file to download
+        :param filepath: The local path or filename to which the file will be written
+        Returns None if successful.
+
+        Raises EnvironmentError if the filepath is invalid
+
+        Raises SBOLError with code SBOL_ERROR_HTTP_UNAUTHORIZED if it
+        there is an HTTP Unauthorized response.
+
+        Raises SBOLError with code SBOL_ERROR_BAD_HTTP_REQUEST on any
+        other HTTP error. The actual status code is embedded in the
+        string message.
+        '''
+        url = self._uri2url(attachment_uri)
+        url = posixpath.join(url, 'download')
+        filepath = os.path.expanduser(filepath)
+
+        # HTTP request headers
+        headers = {
+            'Accept': 'text/plain',
+            'X-authorization': self.key
+        }
+
+        # Issue GET request
+        response = requests.get(url, headers=headers)
+        if response.ok:
+            filename = response.headers['Content-Disposition']
+            filename = filename[22:-1]  # remove extraneous text: attachment; filename="
+            if os.path.isdir(filepath):
+                filepath = posixpath.join(filepath, filename)
+            with open(filepath, 'wb') as filehandle:
+                filehandle.write(response.content)
+            return
+        if response.status_code == http.HTTPStatus.UNAUTHORIZED:
+            # HTTP 401
+            msg = 'You must login with valid credentials before downloading a file'
+            raise SBOLError(SBOLErrorCode.SBOL_ERROR_HTTP_UNAUTHORIZED, msg)
+        if response.status_code == http.HTTPStatus.NOT_FOUND:
+            # HTTP 404
+            msg = 'Unable to download. Attachment {} not found.'.format(attachment_uri)
+            raise SBOLError(SBOLErrorCode.SBOL_ERROR_NOT_FOUND, msg)
+
+        # Not sure what went wrong
+        msg = 'HTTP Error code {} trying to download file.'
+        msg = msg.format(response.status_code)
+        raise SBOLError(SBOLErrorCode.SBOL_ERROR_BAD_HTTP_REQUEST, msg)
+
     def _make_search_item(self, item: dict) -> Identified:
         obj = Identified()
         obj.identity = item['uri']

test/test_partshop.py

@@ -1,6 +1,6 @@
 import os
 import unittest
-
+import json
 import requests
 
 import sbol2 as sbol
@@ -213,6 +213,7 @@ def test_uri2url_spoofed(self):
 
     @unittest.skipIf(password is None, "No password supplied")
     def test_attach_file(self):
+        sbol2.Config.setOption('sbol_typed_uris', False)
         doc = sbol2.Document()
         doc.displayId = 'test_attachment'
         doc.name = 'test attachment'
@@ -222,11 +223,44 @@ def test_attach_file(self):
         sbh = sbol2.PartShop(TEST_RESOURCE)
         sbh.login(username, password)
         sbh.submit(doc, overwrite=1)
-        md_uri = '{}/user/{}/{}/{}/{}'.format(sbh.getURL(), sbh.getUser(),
-                                              doc.displayId, md.displayId,
-                                              md.version)
+        md_uri = '{}/user/{}/{}/{}/{}'.format(sbh.getURL(), sbh.getUser(), doc.displayId,
+                                              md.displayId, md.version)
         sbh.attachFile(md_uri, CRISPR_LOCATION)
 
+        # SBH will autogenerate an Attachment object, so now we query to find out
+        # what the URI of that Attachment object is so we can download it
+        query = '''
+PREFIX rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#>
+PREFIX dcterms: <http://purl.org/dc/terms/>
+PREFIX dc: <http://purl.org/dc/elements/1.1/>
+PREFIX sbh: <http://wiki.synbiohub.org/wiki/Terms/synbiohub#>
+PREFIX prov: <http://www.w3.org/ns/prov#>
+PREFIX sbol: <http://sbols.org/v2#>
+PREFIX xsd: <http://www.w3.org/2001/XMLSchema#>
+PREFIX rdfs: <http://www.w3.org/2000/01/rdf-schema#>
+PREFIX purl: <http://purl.obolibrary.org/obo/>
+SELECT DISTINCT ?attachment_uri
+WHERE {{
+  <{}> sbol:attachment ?attachment_uri
+}}'''.format(md_uri)
+        response = sbh.sparqlQuery(query)
+        content = json.loads(response.content)
+        attachment_uri = content['results']['bindings'][0]['attachment_uri']['value']
+        # Save to default filename
+        sbh.downloadAttachment(attachment_uri)
+        self.assertTrue(os.path.exists('./crispr_example.xml'))
+        os.remove('crispr_example.xml')
+        # Save to new file name
+        sbh.downloadAttachment(attachment_uri, 'foo.xml')
+        self.assertTrue(os.path.exists('./foo.xml'))
+        os.remove('foo.xml')
+        # Confirm error handling
+        sbh.remove(attachment_uri)
+        with self.assertRaises(sbol2.SBOLError) as cm:
+            sbh.downloadAttachment(attachment_uri)
+        e = cm.exception
+        self.assertEqual(e.error_code(), sbol2.SBOLErrorCode.SBOL_ERROR_NOT_FOUND)
+
     def test_search_general(self):
         sbh = sbol2.PartShop(TEST_RESOURCE_MAIN)
         # sbh.login(username, password)