From a84573113106e7b67b63e15abd23445ab5995de1 Mon Sep 17 00:00:00 2001 From: Matthew Evans Date: Mon, 12 Feb 2024 10:53:25 +0000 Subject: [PATCH] Optionally read Arbin into in-memory sqlite without temporary file --- galvani/res2sqlite.py | 21 +++++++++++++++++++-- tests/test_Arbin.py | 15 +++++++++++++++ 2 files changed, 34 insertions(+), 2 deletions(-) diff --git a/galvani/res2sqlite.py b/galvani/res2sqlite.py index 591b945..5864f0c 100755 --- a/galvani/res2sqlite.py +++ b/galvani/res2sqlite.py @@ -571,13 +571,25 @@ def mdb_get_version(filename): return version_tuple -def convert_arbin_to_sqlite(input_file, output_file): +def convert_arbin_to_sqlite(input_file, output_file=None): """Read data from an Arbin .res data file and write to a sqlite file. - Any data currently in the sqlite file will be erased! + Any data currently in an sqlite file at `output_file` will be erased! + + Parameters: + input_file (str): The path to the Arbin .res file to read from. + output_file (str or None): The path to the sqlite file to write to; if None, + return a `sqlite3.Connection` into an in-memory database. + + Returns: + None or sqlite3.Connection + """ arbin_version = mdb_get_version(input_file) + if output_file is None: + output_file = ":memory:" + s3db = sqlite3.connect(output_file) tables_to_convert = copy(mdb_tables) @@ -602,6 +614,11 @@ def convert_arbin_to_sqlite(input_file, output_file): print("Vacuuming database...") s3db.executescript("VACUUM; ANALYZE;") + if output_file == ":memory:": + return s3db + + s3db.close() + def main(argv=None): parser = argparse.ArgumentParser( diff --git a/tests/test_Arbin.py b/tests/test_Arbin.py index d708d3f..3b5d0d4 100644 --- a/tests/test_Arbin.py +++ b/tests/test_Arbin.py @@ -53,6 +53,21 @@ def test_convert_Arbin_to_sqlite_function(testdata_dir, tmpdir, basename): csr.fetchone() +@pytest.mark.parametrize("basename", ["arbin1", "UM34_Test005E"]) +def test_convert_Arbin_to_sqlite_function_in_memory(testdata_dir, tmpdir, basename): + """Convert an Arbin file to an in-memory SQLite database.""" + res_file = os.path.join(testdata_dir, basename + ".res") + conn = None + try: + conn = res2sqlite.convert_arbin_to_sqlite(res_file) + assert conn is not None + csr = conn.execute("SELECT * FROM Channel_Normal_Table;") + csr.fetchone() + finally: + if conn is not None: + conn.close() + + @pytest.mark.skipif( not have_mdbtools, reason="Reading the Arbin file requires MDBTools" )