Repository: cogent3/cogent3
Pull request #2075 (2024-10-28 21:29:24)

ENH: added util.io.is_url function
[NEW] improve robustness of deciding whether an object
    is a url

## Summary by Sourcery

Add a new utility function 'is_url' to improve the robustness of URL detection in the codebase, replacing the previous method. Update relevant parts of the code to use this new function and add tests to ensure its accuracy.

New Features:
- Introduce a new utility function 'is_url' to determine if a given path is a URL.

Enhancements:
- Replace the previous URL detection method with the new 'is_url' function for improved robustness.

Tests:
- Add tests for the new 'is_url' function to verify its correctness in identifying URLs and non-URLs.


Patch:
diff --git a/src/cogent3/parse/fasta.py b/src/cogent3/parse/fasta.py
index aac43aa96..7fb4d877a 100644
--- a/src/cogent3/parse/fasta.py
+++ b/src/cogent3/parse/fasta.py
@@ -16,7 +16,7 @@ from cogent3.core.info import Info
 from cogent3.core.moltype import ASCII, BYTES
 from cogent3.parse.record import RecordError
 from cogent3.parse.record_finder import LabeledRecordFinder
-from cogent3.util.io import open_
+from cogent3.util.io import is_url, open_
 
 _white_space = re.compile(r"\s+")
 strip = str.strip
@@ -442,12 +442,13 @@ def _(
 
 @iter_fasta_records.register
 def _(data: str, converter: OptConverterType = None, label_to_name: RenamerType = str):
-    try:
-        os.stat(data)
-    except OSError:
-        raise TypeError(
-            "data is a string but not a file path, directly provided data must be bytes"
-        )
+    if not is_url(data):
+        try:
+            os.stat(data)
+        except OSError:
+            raise TypeError(
+                "data is a string but not a file path, directly provided data must be bytes"
+            )
 
     with open_(data, mode="rb") as infile:
         data: bytes = infile.read()
diff --git a/src/cogent3/util/io.py b/src/cogent3/util/io.py
index ad33ad1c7..32b6624c0 100644
--- a/src/cogent3/util/io.py
+++ b/src/cogent3/util/io.py
@@ -1,4 +1,5 @@
 import contextlib
+import functools
 import shutil
 import uuid
 from bz2 import open as bzip_open
@@ -7,7 +8,6 @@ from io import TextIOWrapper
 from lzma import open as lzma_open
 from os import PathLike, remove
 from pathlib import Path, PurePath
-from re import compile
 from tempfile import mkdtemp
 from typing import IO, Callable, Iterator, Optional, Tuple, Union
 from urllib.parse import ParseResult, urlparse
@@ -19,8 +19,27 @@ from chardet import detect
 from cogent3.util.misc import _wout_period
 
 PathType = Union[str, PathLike, PurePath]
-# support prefixes for urls
-_urls = compile("^(http[s]*|file)")
+
+
+@functools.singledispatch
+def is_url(path: Union[str, bytes, Path]) -> bool:
+    """whether a path is a url"""
+    return False
+
+
+@is_url.register
+def _(path: str) -> bool:
+    return is_url(urlparse(path))
+
+
+@is_url.register
+def _(path: bytes) -> bool:
+    return is_url(urlparse(path.decode("utf8")))
+
+
+@is_url.register
+def _(path: ParseResult) -> bool:
+    return path.scheme in {"http", "https", "file"}
 
 
 def _get_compression_open(
@@ -73,10 +92,7 @@ def open_zip(filename: PathType, mode: str = "r", **kwargs) -> IO:
 
         opened = zf.open(zf.namelist()[0], mode=mode, **kwargs)
 
-        if binary_mode:
-            return opened
-
-        return TextIOWrapper(opened, encoding=encoding)
+        return opened if binary_mode else TextIOWrapper(opened, encoding=encoding)
 
 
 _compression_handlers = {
@@ -107,7 +123,7 @@ def open_(filename: PathType, mode="rt", **kwargs) -> IO:
     if not filename:
         raise ValueError(f"{filename} not a valid file name or url")
 
-    if _urls.search(str(filename)):
+    if is_url(filename):
         return open_url(filename, mode=mode, **kwargs)
 
     mode = mode or "rt"
@@ -150,16 +166,16 @@ def open_url(url: Union[str, ParseResult], mode="r", **kwargs) -> IO:
     )  # handling possibility of ParseResult
     mode = "rb" if compression else mode or "r"
 
+    if not is_url(url):
+        raise ValueError(
+            f"URL scheme must be http, https or file, not {str(url)[:20]!r}"
+        )
+
     url_parsed = url if isinstance(url, ParseResult) else urlparse(url)
 
     if "r" not in mode:
         raise ValueError("opening a url only allowed in read mode")
 
-    if not _urls.search(url_parsed.scheme):
-        raise ValueError(
-            f"URL scheme must be http, https or file, not {url_parsed.scheme}"
-        )
-
     response = urlopen(url_parsed.geturl(), timeout=10)
     encoding = response.headers.get_content_charset()
     if compression:
@@ -372,7 +388,7 @@ def iter_splitlines(
     -----
     Loads chunks of data from the file, yields one line at a time
     """
-    if _urls.search(str(path)):
+    if is_url(path):
         chunk_size = None
     else:
         path = Path(path).expanduser()


Test patch:
diff --git a/tests/test_util/test_io.py b/tests/test_util/test_io.py
index 3e9949362..896ca2482 100644
--- a/tests/test_util/test_io.py
+++ b/tests/test_util/test_io.py
@@ -12,6 +12,7 @@ from cogent3.util.io import (
     _path_relative_to_zip_parent,
     atomic_write,
     get_format_suffixes,
+    is_url,
     iter_line_blocks,
     iter_splitlines,
     open_,
@@ -499,3 +500,23 @@ def test_iter_line_blocks_none_num_lines(tmp_path):
     got = list(iter_line_blocks(path, num_lines=None))
     expect = [value]
     assert got == expect
+
+
+@pytest.mark.parametrize(
+    "url",
+    (
+        "http://example.com",
+        b"file://example.txt",
+        pathlib.Path("example.txt").absolute().as_uri(),
+    ),
+)
+def test_is_url(url):
+    assert is_url(url)
+
+
+@pytest.mark.parametrize(
+    "url",
+    ("example.txt", pathlib.Path("example.txt"), b"example.txt", r"D:\foo\example.txt"),
+)
+def test_not_is_url(url):
+    assert not is_url(url)