Delete transferred files

Author: forsyth2

zstash/create.py

@@ -6,14 +6,15 @@
 import os.path
 import sqlite3
 import sys
-from typing import Any, List, Tuple
+from typing import Any, List, Optional, Tuple
 
 from six.moves.urllib.parse import urlparse
 
 from .globus import globus_activate, globus_finalize
 from .hpss import hpss_put
 from .hpss_utils import add_files
 from .settings import DEFAULT_CACHE, config, get_db_filename, logger
+from .transfer_tracking import GlobusTransferCollection, HPSSTransferCollection
 from .utils import (
     create_tars_table,
     get_files_to_archive,
@@ -52,12 +53,13 @@ def create():
         logger.error(input_path_error_str)
         raise NotADirectoryError(input_path_error_str)
 
+    gtc: Optional[GlobusTransferCollection] = None
     if hpss != "none":
         url = urlparse(hpss)
         if url.scheme == "globus":
             # identify globus endpoints
-            logger.debug(f"{ts_utc()}:Calling globus_activate(hpss)")
-            globus_activate(hpss)
+            logger.debug(f"{ts_utc()}:Calling globus_activate()")
+            gtc = globus_activate(hpss)
         else:
             # config.hpss is not "none", so we need to
             # create target HPSS directory
@@ -88,14 +90,23 @@ def create():
 
     # Create and set up the database
     logger.debug(f"{ts_utc()}: Calling create_database()")
-    failures: List[str] = create_database(cache, args)
+    htc: HPSSTransferCollection = HPSSTransferCollection()
+    failures: List[str] = create_database(cache, args, gtc=gtc, htc=htc)
 
     # Transfer to HPSS. Always keep a local copy.
     logger.debug(f"{ts_utc()}: calling hpss_put() for {get_db_filename(cache)}")
-    hpss_put(hpss, get_db_filename(cache), cache, keep=args.keep, is_index=True)
+    hpss_put(
+        hpss,
+        get_db_filename(cache),
+        cache,
+        keep=args.keep,
+        is_index=True,
+        gtc=gtc,
+        # htc=htc,
+    )
 
     logger.debug(f"{ts_utc()}: calling globus_finalize()")
-    globus_finalize(non_blocking=args.non_blocking)
+    globus_finalize(gtc, htc, non_blocking=args.non_blocking)
 
     if len(failures) > 0:
         # List the failures
@@ -204,7 +215,12 @@ def setup_create() -> Tuple[str, argparse.Namespace]:
     return cache, args
 
 
-def create_database(cache: str, args: argparse.Namespace) -> List[str]:
+def create_database(
+    cache: str,
+    args: argparse.Namespace,
+    gtc: Optional[GlobusTransferCollection],
+    htc: Optional[HPSSTransferCollection],
+) -> List[str]:
     # Create new database
     logger.debug(f"{ts_utc()}:Creating index database")
     if os.path.exists(get_db_filename(cache)):
@@ -263,26 +279,7 @@ def create_database(cache: str, args: argparse.Namespace) -> List[str]:
     files: List[str] = get_files_to_archive(cache, args.include, args.exclude)
 
     failures: List[str]
-    if args.follow_symlinks:
-        try:
-            # Add files to archive
-            failures = add_files(
-                cur,
-                con,
-                -1,
-                files,
-                cache,
-                args.keep,
-                args.follow_symlinks,
-                skip_tars_md5=args.no_tars_md5,
-                non_blocking=args.non_blocking,
-                error_on_duplicate_tar=args.error_on_duplicate_tar,
-                overwrite_duplicate_tars=args.overwrite_duplicate_tars,
-                force_database_corruption=args.for_developers_force_database_corruption,
-            )
-        except FileNotFoundError:
-            raise Exception("Archive creation failed due to broken symlink.")
-    else:
+    try:
         # Add files to archive
         failures = add_files(
             cur,
@@ -297,7 +294,14 @@ def create_database(cache: str, args: argparse.Namespace) -> List[str]:
             error_on_duplicate_tar=args.error_on_duplicate_tar,
             overwrite_duplicate_tars=args.overwrite_duplicate_tars,
             force_database_corruption=args.for_developers_force_database_corruption,
+            gtc=gtc,
+            htc=htc,
         )
+    except FileNotFoundError as e:
+        if args.follow_symlinks:
+            raise Exception("Archive creation failed due to broken symlink.")
+        else:
+            raise e
 
     # Close database
     con.commit()