Repository: cogent3/cogent3
Pull request #1476 (2023-06-13 01:51:52)

ENH: improve performance of fasta parser
[NEW] implemented 2 different parsers, one for when strict is True, which
    is the same as previously. The other for when strict is False. The latter
    checks only for blank lines.

[CHANGED] the strict=False setting is different to previously, so behaviour
     for malformed fasta files has changed.


Patch:
diff --git a/src/cogent3/core/annotation_db.py b/src/cogent3/core/annotation_db.py
index 169eddc59..acb7c68f5 100644
--- a/src/cogent3/core/annotation_db.py
+++ b/src/cogent3/core/annotation_db.py
@@ -1003,9 +1003,7 @@ class SqliteAnnotationDbMixin:
         }
 
         for tname in other_db.table_names:
-            sql, vals = _select_records_sql(
-                table_name=tname, conditions=conditions
-            )
+            sql, vals = _select_records_sql(table_name=tname, conditions=conditions)
             data = other_db._execute_sql(sql, vals)
             val_placeholder = ", ".join("?" * len(col_order[tname]))
             sql = f"INSERT INTO {tname} ({', '.join(col_order[tname])}) VALUES ({val_placeholder})"
diff --git a/src/cogent3/parse/fasta.py b/src/cogent3/parse/fasta.py
index 9b1f95b35..9b9706c1d 100644
--- a/src/cogent3/parse/fasta.py
+++ b/src/cogent3/parse/fasta.py
@@ -5,6 +5,7 @@ import re
 import typing
 
 from collections.abc import Callable
+from functools import singledispatch
 
 import cogent3
 
@@ -46,6 +47,76 @@ FastaFinder = LabeledRecordFinder(is_fasta_label, ignore=is_blank_or_comment)
 PathOrIterableType = typing.Union[os.PathLike, typing.List[str], typing.Tuple[str]]
 
 
+@singledispatch
+def _prep_data(data) -> typing.Iterable[str]:
+    return data
+
+
+@_prep_data.register
+def _(data: str):
+    with open_(data) as infile:
+        return infile.read().splitlines()
+
+
+@_prep_data.register
+def _(data: os.PathLike):
+    return _prep_data(str(data))
+
+
+def _faster_parser(
+    data: typing.Iterable[str], label_to_name: typing.Callable, label_char: str
+) -> typing.Iterable[typing.Tuple[str, str]]:
+    label = None
+    seq = []
+    for line in data:
+        if not line:
+            # ignore empty lines
+            continue
+
+        if line[0] in label_char:
+            if seq:
+                yield label_to_name(label or ""), "".join(seq)
+
+            label = line[1:].strip()
+            seq = []
+        else:
+            seq.append(line.strip())
+
+    if seq:
+        yield label_to_name(label or ""), "".join(seq)
+
+
+def _strict_parser(
+    data: typing.Iterable[str], label_to_name: typing.Callable, label_char: str
+) -> typing.Iterable[typing.Tuple[str, str]]:
+    seq = []
+    label = None
+    for line in data:
+        if not line or line.startswith("#"):
+            # ignore empty or comment lines
+            continue
+
+        if line[0] in label_char:
+            if label is not None:
+                if not seq:
+                    raise RecordError(f"{label} has no data")
+                yield label_to_name(label), "".join(seq)
+            elif seq:
+                raise RecordError("missing a label")
+
+            label = line[1:].strip()
+            seq = []
+        else:
+            seq.append(line.strip())
+
+    if not seq:
+        raise RecordError(f"{label} has no data")
+    if label is None:
+        raise RecordError("missing a label")
+
+    yield label_to_name(label), "".join(seq)
+
+
 @deprecated_args("2023.8", "faster implementation", discontinued=["finder"])
 def MinimalFastaParser(
     path: PathOrIterableType,
@@ -65,50 +136,17 @@ def MinimalFastaParser(
     label_to_name
         function for converting a label to a name
     label_characters
-        the characters the indicate a line is a label line
-
-    Returns
-    -------
-
+        character(s) at the start of a label line
     """
     if not path:
         return []
-    try:
-        with open_(path) as infile:
-            data = infile.read().splitlines()
-    except (AttributeError, TypeError, ValueError):
-        data = path
-
-    label_char = re.compile(f"^[{label_characters}]")
-    seq = []
-    label = None
-    for line in data:
-        if line.startswith("#"):
-            # comment lines ignored
-            continue
-
-        if label_char.search(line):
-            if label is not None:
-                if strict and not seq:
-                    raise RecordError(f"{label} has no data")
-                if strict and label is None:
-                    raise RecordError("missing a label")
-                if seq:
-                    yield label_to_name(label.strip()), "".join(seq)
-
-            label = line[1:].strip()
-            seq = []
-        elif line := line.strip():
-            seq.append(line)
 
+    data = _prep_data(path)
+    label_char = set(label_characters)
     if strict:
-        if not seq:
-            raise RecordError(f"{label} has no data")
-        if label is None:
-            raise RecordError("missing a label")
-
-    if seq:
-        yield label_to_name(label.strip()), "".join(seq)
+        yield from _strict_parser(data, label_to_name, label_char)
+    else:
+        yield from _faster_parser(data, label_to_name, label_char)
 
 
 GdeFinder = LabeledRecordFinder(is_gde_label, ignore=is_blank)


Test patch:
diff --git a/tests/test_parse/test_fasta.py b/tests/test_parse/test_fasta.py
index 56218df55..faf5580e6 100644
--- a/tests/test_parse/test_fasta.py
+++ b/tests/test_parse/test_fasta.py
@@ -251,12 +251,8 @@ class NcbiFastaParserTests(TestCase):
         self.assertEqual(r, ("123", "ucagUCAGtgacNNNN"))
         # test that we get what we expect if not strict
         r = list(NcbiFastaParser(self.nasty, Sequence, strict=False))
-        self.assertEqual(len(r), 4)
-        a, b, c, d = r
-        self.assertEqual(
-            (a[1], a[1].info.GI, a[1].info.RefSeq, a[1].info.Description),
-            ("UCAG", ["abc"], ["def"], ""),
-        )
+        self.assertEqual(len(r), 2)
+        b, c = r
         self.assertEqual(
             (b[1], b[1].info.GI, b[1].info.GenBank, b[1].info.Description),
             ("UUUUCCCC", ["xyz"], ["qwe"], "descr"),
@@ -265,27 +261,14 @@ class NcbiFastaParserTests(TestCase):
             (c[1], c[1].info.GI, c[1].info.RefSeq, c[1].info.Description),
             ("XYZ", ["bad"], ["stuff"], "label"),
         )
-        self.assertEqual(
-            (d[1], d[1].info.GI, d[1].info.DDBJ, d[1].info.Description),
-            ("ucagUCAGtgacNNNN".upper(), ["123"], ["456"], "desc|with|pipes|"),
-        )
         # ...and when we explicitly supply a constructor
         r = list(NcbiFastaParser(self.nasty, Dna, strict=False))
-        self.assertEqual(len(r), 3)
-        a, b, c = r
-        a, b, c = a[1], b[1], c[1]
-        self.assertEqual(
-            (a, a.info.GI, a.info.RefSeq, a.info.Description),
-            ("TCAG", ["abc"], ["def"], ""),
-        )
+        self.assertEqual(len(r), 1)
+        b = r[0][1]
         self.assertEqual(
             (b, b.info.GI, b.info.GenBank, b.info.Description),
             ("TTTTCCCC", ["xyz"], ["qwe"], "descr"),
         )
-        self.assertEqual(
-            (c, c.info.GI, c.info.DDBJ, c.info.Description),
-            ("tcagTCAGtgacNNNN".upper(), ["123"], ["456"], "desc|with|pipes|"),
-        )
 
 
 class LabelParsingTest(TestCase):