From a84573113106e7b67b63e15abd23445ab5995de1 Mon Sep 17 00:00:00 2001 From: Matthew Evans Date: Mon, 12 Feb 2024 10:53:25 +0000 Subject: [PATCH 1/2] 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" ) From fbc90fc9618c7fd699647f36820fa552371fcb64 Mon Sep 17 00:00:00 2001 From: Matthew Evans <7916000+ml-evs@users.noreply.github.com> Date: Sat, 2 Mar 2024 18:13:40 +0100 Subject: [PATCH 2/2] Update tests/test_Arbin.py Co-authored-by: Chris Kerr --- tests/test_Arbin.py | 7 +------ 1 file changed, 1 insertion(+), 6 deletions(-) diff --git a/tests/test_Arbin.py b/tests/test_Arbin.py index 3b5d0d4..dca9eaa 100644 --- a/tests/test_Arbin.py +++ b/tests/test_Arbin.py @@ -58,14 +58,9 @@ def test_convert_Arbin_to_sqlite_function_in_memory(testdata_dir, tmpdir, basena """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 + with res2sqlite.convert_arbin_to_sqlite(res_file) as conn: csr = conn.execute("SELECT * FROM Channel_Normal_Table;") csr.fetchone() - finally: - if conn is not None: - conn.close() @pytest.mark.skipif(