Repository: cogent3/cogent3
Pull request #1065 (2022-06-27 19:21:05)

Add open_url function in util.io and its test in test_util.io


Patch:
diff --git a/src/cogent3/util/io.py b/src/cogent3/util/io.py
index f2bffdf2d..67548ff98 100644
--- a/src/cogent3/util/io.py
+++ b/src/cogent3/util/io.py
@@ -9,6 +9,10 @@ from pathlib import Path
 from tempfile import mkdtemp
 from typing import Union
 from zipfile import ZipFile
+from urllib.parse import ParseResult
+from io import TextIOWrapper
+from urllib.request import urlopen
+from urllib.parse import urlparse
 
 from chardet import detect
 
@@ -17,7 +21,7 @@ from cogent3.util.misc import _wout_period
 
 __author__ = "Gavin Huttley"
 __copyright__ = "Copyright 2007-2022, The Cogent Project"
-__credits__ = ["Gavin Huttley"]
+__credits__ = ["Gavin Huttley", "Nick Shahmaras"]
 __license__ = "BSD-3"
 __version__ = "2022.5.25a1"
 __maintainer__ = "Gavin Huttley"
@@ -81,6 +85,39 @@ def open_(filename: Union[str, Path], mode="rt", **kwargs):
     return op(filename, mode, encoding=encoding, **kwargs)
 
 
+def open_url(url: Union[str, ParseResult], mode="rb", **kwargs):
+    """open a url
+
+    Parameters
+    ----------
+    url : urllib.parse.ParseResult or str
+        A url of file in http or https web address
+    mode : str
+        mode of reading file, {"rb", "rt", "r"}
+
+    Notes
+    -----
+    If mode="b" or "rb" (binary read), the function returns file object to read
+    else returns TextIOWrapper to read text with specified encoding in the URL
+    """
+
+    url_parsed = url if isinstance(url, ParseResult) else urlparse(url)
+
+    if "r" not in mode:
+        raise ValueError("openning a url only allowed in read mode")
+
+    if url_parsed.scheme not in "https":
+        raise ValueError("URL scheme must be http or https.")
+
+    response = urlopen(url_parsed.geturl(), timeout=10)
+
+    return (
+        response
+        if "b" in mode
+        else TextIOWrapper(response, encoding=response.headers.get_content_charset())
+    )
+
+
 def _path_relative_to_zip_parent(zip_path, member_path):
     """returns member_path relative to zip_path
 


Test patch:
diff --git a/tests/test_util/test_io.py b/tests/test_util/test_io.py
index f8bdf2f20..17eb1344e 100644
--- a/tests/test_util/test_io.py
+++ b/tests/test_util/test_io.py
@@ -4,6 +4,7 @@ import os
 import pathlib
 import tempfile
 import zipfile
+import pytest
 
 from os import remove, rmdir
 from os.path import exists
@@ -17,12 +18,13 @@ from cogent3.util.io import (
     open_,
     path_exists,
     remove_files,
+    open_url,
 )
 
 
 __author__ = "Gavin Huttley"
 __copyright__ = "Copyright 2007-2022, The Cogent Project"
-__credits__ = ["Gavin Huttley"]
+__credits__ = ["Gavin Huttley", "Nick Shahmaras"]
 __license__ = "BSD-3"
 __version__ = "2022.5.25a1"
 __maintainer__ = "Gavin Huttley"
@@ -302,6 +304,43 @@ class UtilsTests(TestCase):
             with self.assertRaises(ValueError):
                 open_(zip_path)
 
+    def test_open_url(self):
+        from urllib.parse import urlparse
+
+        local_root = pathlib.Path("data")
+        remote_root = (
+            "https://raw.githubusercontent.com/cogent3/cogent3/develop/tests/data/{}"
+        )
+        for file_name in ("c_elegans_WS199_dna_shortened.fasta", "gff2_test.gff"):
+            for mode in ("r", "rb", "rt"):
+                with open_(local_root / file_name, mode=mode) as infile:
+                    local_data = infile.read()
+                with open_url(remote_root.format(file_name), mode=mode) as infile:
+                    remote_data = infile.read()
+                self.assertEqual(remote_data.splitlines(), local_data.splitlines())
+
+                # Test ParseResult for url
+                with open_url(
+                    urlparse(remote_root.format(file_name)), mode=mode
+                ) as infile:
+                    remote_data = infile.read()
+                self.assertEqual(remote_data.splitlines(), local_data.splitlines())
+
+                if "b" in mode:  # Test no value for mode, use default "rb" mode
+                    with open_url(remote_root.format(file_name)) as infile:
+                        remote_data = infile.read()
+                    self.assertEqual(remote_data.splitlines(), local_data.splitlines())
+
+        # Test "w" mode (should raise Exception)
+        url_data = "https://raw.githubusercontent.com/cogent3/cogent3/develop/tests/data/gff2_test.gff"
+        with pytest.raises(Exception):
+            file_url = open_url(url_data, mode="w")
+
+        # Test  non-http(s) address for url (should raise Exception)
+        url_data = "ftp://raw.githubusercontent.com/cogent3/cogent3/develop/tests/data/gff2_test.gff"
+        with pytest.raises(Exception):
+            file_url = open_url(url_data, mode="r")
+
 
 if __name__ == "__main__":
     main()