Replace internal tabs in data to be written out as TSVs

kg_covid_19/utils/transform_utils.py

@@ -3,10 +3,9 @@
 import gzip
 import logging
 import zipfile
-from typing import Any, Dict, List, Union
+from typing import Any, Dict, List, Union, TextIO
 from tqdm import tqdm  # type: ignore
 
-
 class TransformError(Exception):
     """Base class for other exceptions"""
     pass
@@ -63,15 +62,23 @@ def get_header_items(table_data: Any) -> List:
     return header_items
 
 
-def write_node_edge_item(fh: Any, header: List, data: List, sep: str = '\t'):
+def write_node_edge_item(fh: TextIO, header: List, data: List, sep: str = '\t',
+                         sanitize_sep_char=True):
     """Write out a single line for a node or an edge in *.tsv
     :param fh: file handle of node or edge file
     :param header: list of header items
     :param data: data for line to write out
     :param sep: separator [\t]
+    :param sanitize_sep_char: replace sep character in data with hex
+    present in `data`
     """
     if len(header) != len(data):
         raise Exception('Header and data are not the same length.')
+
+    if sanitize_sep_char:
+        for i in range(len(data)):
+            data[i] = data[i].replace(sep, hex(ord(sep)))
+
     try:
         fh.write(sep.join(data) + "\n")
     except IOError:

tests/test_transform_utils.py

@@ -0,0 +1,45 @@
+import os
+import tempfile
+import unittest
+
+from kg_covid_19.utils import write_node_edge_item
+
+
+class TestTransformUtils(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls) -> None:
+        cls.header = ['id', 'name', 'category']
+        cls.valid_data = ['id1234', '1234', 'biolink:Gene']
+
+    def setUp(self) -> None:
+        self.tempdir = tempfile.gettempdir()
+        self.outfile = os.path.join(self.tempdir, 'some.tsv')
+        self.fh = open(self.outfile, 'w')
+
+    def test_write_node_edge_item_bad_fh(self):
+        with self.assertRaises(Exception):
+            write_node_edge_item(fh='',  header=self.header, data=self.valid_data)
+
+    def test_write_node_edge_item(self):
+        write_node_edge_item(fh=self.fh,  header=self.header, data=self.valid_data)
+        self.fh.close()
+        self.assertTrue(os.path.exists(self.outfile))
+        with open(self.outfile, 'r') as tsvfile:
+            lines = tsvfile.read().split('\n')
+            self.assertEqual(['id1234', '1234', 'biolink:Gene'], lines[0].split('\t'))
+
+    def test_write_node_edge_item_with_tabs_in_data(self):
+        write_node_edge_item(fh=self.fh,
+                             header=self.header,
+                             data=['id1234', '1234', 'biolink:Gene\tbiolink:Gene\t'],
+                             sanitize_sep_char=True)
+        self.fh.close()
+        self.assertTrue(os.path.exists(self.outfile))
+        with open(self.outfile, 'r') as tsvfile:
+            lines = tsvfile.read().split('\n')
+            self.assertEqual(['id1234',
+                              '1234',
+                              'biolink:Gene0x9biolink:Gene0x9'],
+                              lines[0].split('\t'))
+
+