feat: use context manager to temp deck file deletion. Fixes #423 (#427)

feat: use context manager to temp deck file deletion. Fixes #423

implement

personal_mnemonic_medium/data_access/ankiconnect_gateway.py

@@ -4,7 +4,7 @@
 import logging
 import traceback
 import urllib.request
-from collections.abc import Mapping, Sequence
+from collections.abc import Iterator, Mapping, Sequence
 from dataclasses import dataclass
 from enum import Enum
 from pathlib import Path
@@ -18,6 +18,23 @@
 
 log = logging.getLogger(__name__)
 
+import shutil
+from contextlib import contextmanager
+
+
+@contextmanager
+def tempdir(tmp_path: Path) -> Iterator[Path]:
+    """Context manager for a temporary directory that is deleted after use."""
+    try:
+        tmp_path.mkdir(parents=True, exist_ok=True)
+        yield tmp_path
+    except:
+        # If there's an error, ensure the directory is deleted before the error is propagated.
+        shutil.rmtree(str(tmp_path))
+        raise
+    else:
+        shutil.rmtree(str(tmp_path))
+
 
 class AnkiField(pydantic.BaseModel):
     value: str
@@ -83,20 +100,22 @@ def update_model(self, model: genanki.Model) -> None:
         )
 
     def import_package(self, package: genanki.Package) -> None:
-        apkg_name = (
-            f"{datetime.datetime.now().strftime('%Y-%m-%d-%H-%M-%S')}.apkg"
-        )
-        write_path = self.tmp_write_dir / apkg_name
-        package.write_to_file(write_path)  # type: ignore
-
-        read_path = self.tmp_read_dir / apkg_name
-        try:
-            self._invoke(AnkiConnectCommand.IMPORT_PACKAGE, path=str(read_path))
-            log.info(f"Imported from {read_path}!")
-            write_path.unlink()
-        except Exception:
-            log.error(f"""Unable to sync from {read_path}.""")
-            traceback.print_exc()
+        subdir = "tmp_apkg_dir"
+        with tempdir(self.tmp_write_dir / subdir) as tmp_write_subdir:
+            apkg_name = (
+                f"{datetime.datetime.now().strftime('%Y-%m-%d-%H-%M-%S')}.apkg"
+            )
+            package.write_to_file(tmp_write_subdir / apkg_name)  # type: ignore
+
+            read_path = self.tmp_read_dir / subdir / apkg_name
+            try:
+                self._invoke(
+                    AnkiConnectCommand.IMPORT_PACKAGE, path=str(read_path)
+                )
+                log.info(f"Imported from {read_path}!")
+            except Exception as e:
+                log.error(f"""Unable to sync from {read_path}, {e}""")
+                traceback.print_exc()
 
     def delete_notes(self, note_ids: Sequence[int]) -> None:
         if len(note_ids) > self.max_deletions_per_run:

personal_mnemonic_medium/data_access/test_ankiconnect_gateway.py

@@ -32,15 +32,6 @@ class TestAnkiConnectGateway:
     output_path = Path("/output")
 
     def test_import_get_delete_happy_path(self):
-        gateway = AnkiConnectGateway(
-            ankiconnect_url=ANKICONNECT_URL,
-            base_deck="Test deck",
-            tmp_read_dir=(get_host_home_dir() / "ankidecks"),
-            tmp_write_dir=self.output_path,
-            max_deletions_per_run=1,
-            max_wait_seconds=0,
-        )
-
         # Delete all .apkg in the output directory
         for f in self.output_path.glob("*.apkg"):
             f.unlink()
@@ -68,10 +59,20 @@ def test_import_get_delete_happy_path(self):
             )
         )
 
-        package = genanki.Package(deck_or_decks=deck)
+        tmp_read_dir = get_host_home_dir() / "ankidecks"
+        gateway = AnkiConnectGateway(
+            ankiconnect_url=ANKICONNECT_URL,
+            base_deck="Test deck",
+            tmp_read_dir=tmp_read_dir,
+            tmp_write_dir=self.output_path,
+            max_deletions_per_run=1,
+            max_wait_seconds=0,
+        )
 
         # Phase 1: Importing
+        package = genanki.Package(deck_or_decks=deck)
         gateway.import_package(package=package)
+        assert self.output_path.is_dir()
         assert len(list(Path("/output").glob("*.apkg"))) == 0
 
         # Phase 2: Getting