Exposes the same interface to import JSON or VOTable

.github/workflows/test_main.yml

@@ -43,11 +43,11 @@ jobs:
         pytest --cov=./ --cov-report=xml
     - name: Upload coverage to Codecov
       if: matrix.python-version == '3.8'
-      uses: codecov/codecov-action@v1
+      uses: codecov/codecov-action@v4
       with:
         token: ${{ secrets.CODECOV_TOKEN }}
         file: ./coverage.xml
         flags: unittests
         name: codecov-umbrella
         yml: ./codecov.yml
-        fail_ci_if_error: true
+        fail_ci_if_error: false

.github/workflows/test_pull_requests.yml

@@ -43,11 +43,11 @@ jobs:
         pytest --cov=./ --cov-report=xml
     - name: Upload coverage to Codecov
       if: matrix.python-version == '3.8'
-      uses: codecov/codecov-action@v1
+      uses: codecov/codecov-action@v4
       with:
         token: ${{ secrets.CODECOV_TOKEN }}
         file: ./coverage.xml
         flags: unittests
         name: codecov-umbrella
         yml: ./codecov.yml
-        fail_ci_if_error: true
+        fail_ci_if_error: false

tests/test_api.py

@@ -9,9 +9,11 @@
 from ddt import ddt, data, unpack  # type: ignore
 from tscat import create_event, create_catalogue, add_events_to_catalogue, get_events, \
     discard, save, has_unsaved_changes, export_json, import_json, get_catalogues, \
-    import_votable, export_votable
+    import_votable, import_votable_file, import_votable_str, export_votable, export_votable_str
 from tscat.filtering import Comparison, Field
 
+__here__ = os.path.dirname(__file__)
+
 
 @ddt
 class TestAPIAttributes(unittest.TestCase):
@@ -758,8 +760,33 @@ def test_data_is_preserved_with_multiple_export_import_cycles_in_empty_database(
 
                 self.assertEqual(len(get_events()), 0)
                 self.assertEqual(len(get_catalogues()), 0)
+                f.seek(0)
+                import_votable_str(f.read(), os.path.basename(f.name))
+
+                self.assertListEqual(events, get_events())
+
+                new_catalogue = get_catalogues()[0]
+
+                assert new_catalogue.name == catalogue.name
+                assert new_catalogue.author == 'Patrick'
+                assert new_catalogue.uuid != catalogue.uuid
+
+
+    def test_data_is_preserved_with_multiple_export_import_as_string_cycles_in_empty_database(self):
+        events = [generate_event() for _ in range(10)]
+        catalogue = create_catalogue("TestExportImportCatalogue", "Patrick", events=events)
+
+        for _ in range(3):
+            with tempfile.NamedTemporaryFile('w+') as f:
+                export_vot:str = export_votable_str(catalogue)
+                f.write(export_vot)
 
-                import_votable(f.name)
+                discard()
+
+                self.assertEqual(len(get_events()), 0)
+                self.assertEqual(len(get_catalogues()), 0)
+                f.seek(0)
+                import_votable_str(f.read(), os.path.basename(f.name))
 
                 self.assertListEqual(events, get_events())
 
@@ -780,7 +807,7 @@ def test_import_names_are_correct_when_votable_contains_multiple_tables(self):
 
             discard()
 
-            import_votable(f.name)
+            import_votable_file(f.name)
 
             self.assertListEqual(events, get_events())
 
@@ -806,7 +833,7 @@ def test_data_is_preserved_when_importing_over_existing_events_and_catalogues_in
                 export_vot = export_votable(catalogue)
                 export_vot.to_xml(f)
 
-                import_votable(f.name)
+                import_votable_file(f.name)
 
                 assert events == get_events()
                 assert len(get_catalogues()) == i + 2
@@ -821,7 +848,7 @@ def test_importing_a_catalogue_where_all_events_are_already_present(self):
 
             catalogue.remove(permanently=True)
 
-            import_votable(f.name)
+            import_votable_file(f.name)
 
         assert events == get_events()
         assert len(get_catalogues()) == 1
@@ -837,7 +864,7 @@ def test_exception_raised_upon_event_import_with_same_uuid_but_different_attrs(s
             events[0].author = "Someone Else"
 
             with self.assertRaises(ValueError):
-                import_votable(f.name)
+                import_votable_file(f.name)
 
     def test_votable_no_exception_raised_reimporting_catalogue(self):
         events = [generate_event() for _ in range(2)]
@@ -851,7 +878,7 @@ def test_votable_no_exception_raised_reimporting_catalogue(self):
 
             add_events_to_catalogue(catalogue, event)
 
-            import_votable(f.name)
+            import_votable_file(f.name)
 
     def test_export_import_multiple_catalogues_with_shared_and_individual_events(self):
         shared_events = [generate_event() for _ in range(2)]
@@ -870,7 +897,7 @@ def test_export_import_multiple_catalogues_with_shared_and_individual_events(sel
 
             discard()
 
-            catalogue1, catalogue2 = import_votable(f.name)
+            catalogue1, catalogue2 = import_votable_file(f.name)
 
             def s(l):
                 return sorted(l, key=lambda x: x.uuid)
@@ -896,7 +923,7 @@ def test_import_multiple_catalogues_with_shared_and_individual_but_already_exist
             export_vot = export_votable([catalogue1, catalogue2])
             export_vot.to_xml(f)
 
-            catalogue1, catalogue2 = import_votable(f.name)
+            catalogue1, catalogue2 = import_votable_file(f.name)
 
             def s(l):
                 return sorted(l, key=lambda x: x.uuid)
@@ -922,22 +949,33 @@ def test_of_existing_dynamic_catalogue_doesnt_crash(self):
             export_vot = export_votable(c)
             export_vot.to_xml(f)
 
-            d = import_votable(f.name)
+            d = import_votable_file(f.name)
 
         assert get_catalogues() == [c, d[0]]
         assert get_events() == events
         assert get_events(c)[0] == [events[1]]
 
     def test_raise_when_importing_votable_with_unhandled_field_type(self):
         with self.assertRaises(ValueError):
-            import_votable('tests/invalid-type-votable.xml')
+            import_votable_file(f'{__here__}/invalid-type-votable.xml')
 
     def test_raise_when_missing_required_field(self):
         with self.assertRaises(ValueError):
-            import_votable('tests/missing-required-field-votable.xml')
+            import_votable_file(f'{__here__}/missing-required-field-votable.xml')
 
     def test_import_amda_exported_table(self):
-        catalogue, = import_votable('tests/Dst_Li2020.xml')
+        catalogue, = import_votable_file(f'{__here__}/Dst_Li2020.xml')
+
+        assert get_catalogues() == [catalogue]
+        assert get_catalogues()[0].name == 'Dst_Li2020'
+        assert len(get_events()) == 95
+        assert len(get_events(catalogue)[0]) == 95
+        assert len(get_events(Comparison('==', Field('author'), '[email protected]'))) == 95
+
+    def test_import_amda_exported_table_from_votable_object(self):
+        from astropy.io.votable import parse
+        table = parse(f'{__here__}/Dst_Li2020.xml')
+        catalogue, = import_votable(table)
 
         assert get_catalogues() == [catalogue]
         assert get_catalogues()[0].name == 'Dst_Li2020'
@@ -965,3 +1003,7 @@ def test_raise_if_attributes_with_same_name_have_different_types(self):
 
         with self.assertRaises(ValueError):
             export_votable(catalogue)
+
+    def test_raises_when_importing_non_existing_file(self):
+        with self.assertRaises(FileNotFoundError):
+            import_votable_file(f'{__here__}/non-existing-file.xml')

tscat/__init__.py

@@ -10,7 +10,8 @@
     get_catalogues, get_events, \
     _Catalogue, _Event, Session, EventQueryInformation
 
-from .import_export import export_votable, export_json, \
-    import_json, import_votable, \
+from .import_export import export_votable, export_votable_str, export_json, \
+    import_json, import_json_file, import_votable, import_votable_file, import_votable_str, \
+    __canonicalize_votable_import as canonicalize_votable_import, \
     __canonicalize_json_import as canonicalize_json_import, \
     __import_canonicalized_dict as import_canonicalized_dict

tscat/import_export.py

@@ -3,9 +3,12 @@
 import itertools
 import json
 import os
+from io import StringIO ,BytesIO
 from typing import Dict, List, Union, Tuple, Any, Optional, Type, Callable
 from uuid import uuid4
 
+from sqlalchemy_utils import table_name
+
 from .base import get_catalogues, get_events, _Catalogue, _Event, backend, Session, _listify
 from .filtering import UUID
 
@@ -163,6 +166,12 @@ def import_json(jsons: str) -> List[_Catalogue]:
     return __import_canonicalized_dict(import_dict)
 
 
+def import_json_file(filename: str) -> List[_Catalogue]:
+    with open(filename, 'r') as file:
+        return import_json(file.read())
+    raise FileNotFoundError(f'File {filename} not found')
+
+
 ### VOTable (AMDA compatible)
 
 from astropy.io.votable import parse
@@ -294,11 +303,15 @@ def export_votable(catalogues: Union[List[_Catalogue], _Catalogue]) -> VOTableFi
     return votable
 
 
-def import_votable(filename: str) -> List[_Catalogue]:
-    votable = parse(filename)
+def export_votable_str(catalogues: Union[List[_Catalogue], _Catalogue]) -> str:
+    content = BytesIO()
+    export_votable(catalogues).to_xml(content)
+    return content.getvalue().decode()
 
+
+def __canonicalize_votable_import(votable: VOTableFile, table_name: Optional[str] = None) -> __CanonicalizedTSCatData:
     author = 'VOTable Import'
-    table_name = os.path.basename(filename)
+    table_name = table_name or f'Imported Catalogue from {dt.datetime.now()}'
 
     if votable.description:
         for line in str(votable.description).split(';'):
@@ -371,5 +384,22 @@ def import_votable(filename: str) -> List[_Catalogue]:
 
         ddict['catalogues'].append(catalogue)
 
-    data = __canonicalize_from_dict(ddict)
-    return __import_canonicalized_dict(data)
+    return __canonicalize_from_dict(ddict)
+
+
+def import_votable(votable: VOTableFile, table_name: Optional[str] = None) -> List[_Catalogue]:
+    dict = __canonicalize_votable_import(votable, table_name=table_name)
+    return __import_canonicalized_dict(dict)
+
+
+def import_votable_file(filename: str, table_name: Optional[str] = None) -> List[_Catalogue]:
+    if os.path.exists(filename):
+        table_name = table_name or os.path.basename(filename)
+        dict = __canonicalize_votable_import(parse(filename), table_name=table_name)
+        return __import_canonicalized_dict(dict)
+    raise FileNotFoundError(f'File {filename} not found')
+
+
+def import_votable_str(xml_content: str, table_name: Optional[str] = None) -> List[_Catalogue]:
+    dict = __canonicalize_votable_import(parse(BytesIO(xml_content.encode())), table_name=table_name)
+    return __import_canonicalized_dict(dict)