Repository: cogent3/cogent3
Pull request #1470 (2023-06-06 08:21:30)

MAINT: added method to merge two dbs 


Patch:
diff --git a/src/cogent3/core/annotation_db.py b/src/cogent3/core/annotation_db.py
index c8af9fa4d..1f061a962 100644
--- a/src/cogent3/core/annotation_db.py
+++ b/src/cogent3/core/annotation_db.py
@@ -634,7 +634,7 @@ class SqliteAnnotationDbMixin:
     def _get_feature_by_id(
         self,
         table_name: str,
-        columns: typing.Optional[list[str], tuple[str]],
+        columns: typing.Optional[typing.Union[list[str], tuple[str]]],
         column: str,
         name: str,
         start: OptionalInt = None,
@@ -927,7 +927,7 @@ class SqliteAnnotationDbMixin:
         if not annot_db or not len(annot_db):
             return
 
-        self._update_db_from_rich_dict(annot_db.to_rich_dict(), seqids=seqids)
+        self._update_db_from_other_db(annot_db, seqids=seqids)
 
     def union(self, annot_db: SupportsFeatures) -> SupportsFeatures:
         """returns a new instance with merged records with other
@@ -988,6 +988,36 @@ class SqliteAnnotationDbMixin:
                 )
                 self._execute_sql(sql, vals)
 
+    def _update_db_from_other_db(
+        self, other_db: SupportsFeatures, seqids: OptionalStr = None
+    ):
+        if isinstance(seqids, str):
+            seqids = {seqids}
+        elif seqids is not None:
+            seqids = None if len(set(seqids) - {None}) == 0 else set(seqids) - {None}
+
+        table_names = other_db.table_names
+
+        with other_db.db:
+            for table_name in table_names:
+                sql, values = _select_records_sql(
+                    table_name=table_name, conditions={"seqid": seqids}
+                )
+
+                cursor = other_db._execute_sql(sql, values)
+
+                with self.db:
+                    target_cursor = self.db.cursor()
+                    columns = ", ".join([desc[0] for desc in cursor.description])
+                    placeholders = ", ".join(["?"] * len(cursor.description))
+                    insert_sql = (
+                        f"INSERT INTO {table_name} ({columns}) VALUES ({placeholders})"
+                    )
+
+                    target_cursor.executemany(insert_sql, cursor)
+
+                    self.db.commit()  # not sure if this is needed...
+
     def to_json(self) -> str:
         return json.dumps(self.to_rich_dict())
 


Test patch:
diff --git a/tests/data/simple.gff b/tests/data/simple.gff
index 2748c1e8b..0ceb2dbe3 100644
--- a/tests/data/simple.gff
+++ b/tests/data/simple.gff
@@ -4,3 +4,4 @@ test_seq	.	CDS 	3	9	.	+	.	ID=cds1
 test_seq	.	exon	12	20	.	+	.	ID=exon2
 test_seq	.	CDS	    4	10	.	+	.	ID=cds2
 test_seq	.	CpG 	3	12	.	+	.	ID=CpG1
+test_seq2	.	CpG 	3	12	.	+	.	ID=CpG3
diff --git a/tests/test_core/test_annotation_db.py b/tests/test_core/test_annotation_db.py
index 9cb165109..8beffd06c 100644
--- a/tests/test_core/test_annotation_db.py
+++ b/tests/test_core/test_annotation_db.py
@@ -716,6 +716,29 @@ def test_gff_update_existing_specify_seqid(gff_db, gff_small_db, seqids):
     assert gff_db.num_matches() == expect
 
 
+def test_gff_update_db_from_other_db_existing(gff_db, gff_small_db):
+    expect = gff_db.num_matches() + gff_small_db.num_matches()
+    gff_db._update_db_from_other_db(other_db=gff_small_db)
+    assert gff_db.num_matches() == expect
+
+
+@pytest.mark.parametrize(
+    "seqids", ("test_seq", ["test_seq"], "test_seq2", ["test_seq2"], None)
+)
+def test_gff_update_db_from_other_db_existing_specify_seqid(
+    gff_db, gff_small_db, seqids
+):
+    expect = gff_db.num_matches() + gff_small_db.num_matches(seqid=seqids)
+    gff_db._update_db_from_other_db(other_db=gff_small_db, seqids=seqids)
+    assert gff_db.num_matches() == expect
+
+
+def test_gff_update_db_from_other_db_existing_none_seqid(gff_db, gff_small_db):
+    expect = gff_db.num_matches() + gff_small_db.num_matches()
+    gff_db._update_db_from_other_db(other_db=gff_small_db, seqids=None)
+    assert gff_db.num_matches() == expect
+
+
 def test_relative_position_negative_feature(seq_db):
     orig_feat_span = list(seq_db.get_features(name="Transcript:B0019.1"))[0].map