Merge pull request #19 from ENCODE-DCC/dev

hotfix

.circleci/config.yml

@@ -10,7 +10,7 @@ defaults:
 
 machine_defaults: &machine_defaults
   machine:
-    image: ubuntu-1604:201903-01
+    image: ubuntu-1604:202007-01
   working_directory: ~/autouri
 
 
@@ -41,7 +41,7 @@ install_precommit: &install_precommit
 install_py3_packages: &install_py3_packages
   name: Install Python packages
   command: |
-    sudo pip3 install pytest requests dateparser filelock
+    sudo pip3 install pytest requests dateparser filelock "six>=1.13.0"
     sudo pip3 install --upgrade pyasn1-modules
 
 
@@ -111,9 +111,9 @@ jobs:
 
             # clean up
             rm -f tmp_key.json
-            gsutil -m rm -rf ${S3_ROOT}/${CIRCLE_WORKFLOW_ID}
-            gsutil -m rm -rf ${GCS_ROOT}/${CIRCLE_WORKFLOW_ID}
-            gsutil -m rm -rf ${GCS_ROOT_URL}/${CIRCLE_WORKFLOW_ID}
+            gsutil -m rm -rf ${S3_ROOT}/${CIRCLE_WORKFLOW_ID} || true
+            gsutil -m rm -rf ${GCS_ROOT}/${CIRCLE_WORKFLOW_ID} || true
+            gsutil -m rm -rf ${GCS_ROOT_URL}/${CIRCLE_WORKFLOW_ID} || true
 
 
 # Define workflow here

autouri/__init__.py

@@ -5,4 +5,4 @@
 from .s3uri import S3URI
 
 __all__ = ["AbsPath", "AutoURI", "URIBase", "GCSURI", "HTTPURL", "S3URI"]
-__version__ = "0.2.2"
+__version__ = "0.2.3"

autouri/autouri.py

@@ -13,12 +13,12 @@
 logger = logging.getLogger(__name__)
 
 
-def autouri_rm(uri, thread_id):
+def autouri_rm(uri, thread_id, no_lock=False):
     """Wrapper for AutoURI(uri).rm().
     This function is used for multiprocessing.map() which requires a picklable function
     outside the scope of the class.
     """
-    AutoURI(uri, thread_id=thread_id).rm()
+    AutoURI(uri, thread_id=thread_id).rm(no_lock=no_lock)
 
 
 class AutoURIRecursionError(RuntimeError):
@@ -370,8 +370,9 @@ def rmdir(self, dry_run=False, num_threads=DEFAULT_NUM_THREADS, no_lock=False):
             return
         num_files = len(files)
         thread_ids = [i % num_threads for i in range(num_files)]
+        no_locks = [no_lock] * num_files
 
-        args = list(zip(files, thread_ids))
+        args = list(zip(files, thread_ids, no_locks))
         with multiprocessing.Pool(num_threads) as p:
             p.starmap(autouri_rm, args)
 
@@ -392,6 +393,7 @@ def localize_on(
         make_md5_file=False,
         return_flag=False,
         depth=0,
+        no_lock=False,
     ) -> Tuple[str, bool]:
         """Wrapper for classmethod localize().
         Localizes self on target directory loc_prefix.
@@ -403,6 +405,7 @@ def localize_on(
             loc_prefix=loc_prefix,
             return_flag=return_flag,
             depth=depth,
+            no_lock=no_lock,
         )
 
     @abstractmethod
@@ -517,6 +520,7 @@ def localize(
         loc_prefix=None,
         return_flag=False,
         depth=0,
+        no_lock=False,
     ) -> Tuple[str, bool]:
         """Localize a source URI on this URI class (cls).
 
@@ -547,6 +551,8 @@ def localize(
                     See "Returns" section for details about flag
             depth:
                 To count recursion depth.
+            no_lock:
+                No file locking.
         Returns:
             loc_uri:
                 Localized URI STRING (not a AutoURI instance) since it should be used
@@ -602,6 +608,7 @@ def fnc_loc(uri):
                     loc_prefix=loc_prefix,
                     return_flag=True,
                     depth=depth + 1,
+                    no_lock=no_lock,
                 )
 
             for ext, fnc_recurse in AutoURI.LOC_RECURSE_EXT_AND_FNC.items():
@@ -618,14 +625,14 @@ def fnc_loc(uri):
             basename = src_uri.basename_wo_ext + cls.get_loc_suffix() + src_uri.ext
             dirname = src_uri.loc_dirname
             loc_uri = cls.get_path_sep().join([loc_prefix, dirname, basename])
-            AutoURI(loc_uri).write(maybe_modified_contents)
+            AutoURI(loc_uri).write(maybe_modified_contents, no_lock=no_lock)
 
         elif on_different_storage:
             basename = src_uri.basename
             dirname = src_uri.loc_dirname
 
             loc_uri = cls.get_path_sep().join([loc_prefix, dirname, basename])
-            src_uri.cp(dest_uri=loc_uri, make_md5_file=make_md5_file)
+            src_uri.cp(dest_uri=loc_uri, make_md5_file=make_md5_file, no_lock=no_lock)
         else:
             loc_uri = src_uri._uri
 

autouri/cli.py

@@ -29,6 +29,11 @@ def parse_args():
     parent_src = argparse.ArgumentParser(add_help=False)
     parent_src.add_argument("src", help="Source file URI")
 
+    parent_lock = argparse.ArgumentParser(add_help=False)
+    parent_lock.add_argument(
+        "--no-lock", action="store_true", help="No file locking for write/rm action."
+    )
+
     parent_target = argparse.ArgumentParser(add_help=False)
     parent_target.add_argument(
         "target",
@@ -68,7 +73,7 @@ def parse_args():
         "target must be a full filename/directory. "
         "Target directory must have a trailing directory separator "
         "(e.g. /)",
-        parents=[parent_src, parent_target, parent_cp, parent_all],
+        parents=[parent_src, parent_lock, parent_target, parent_cp, parent_all],
     )
     subparser.add_parser(
         "find",
@@ -86,19 +91,21 @@ def parse_args():
     p_write = subparser.add_parser(
         "write",
         help="AutoURI(src).write(text): Write text on source.",
-        parents=[parent_src, parent_all],
+        parents=[parent_src, parent_lock, parent_all],
     )
     p_write.add_argument("text", help="Text to be written to source file.")
 
     subparser.add_parser(
-        "rm", help="AutoURI(src).rm(): Delete source.", parents=[parent_src, parent_all]
+        "rm",
+        help="AutoURI(src).rm(): Delete source.",
+        parents=[parent_src, parent_lock, parent_all],
     )
 
     p_rmdir = subparser.add_parser(
         "rmdir",
         help="AutoURI(src).rmdir(): Recursively delete all files on "
         "source directory.",
-        parents=[parent_src, parent_all],
+        parents=[parent_src, parent_lock, parent_all],
     )
     p_rmdir.add_argument("--delete", action="store_true", help="DELETE outputs.")
     p_rmdir.add_argument(
@@ -114,7 +121,7 @@ def parse_args():
         "loc",
         help="AutoURI(src).localize_on(target): Localize source on target directory "
         "Target directory must end with directory separator",
-        parents=[parent_src, parent_target, parent_cp, parent_all],
+        parents=[parent_src, parent_target, parent_cp, parent_lock, parent_all],
     )
     p_loc.add_argument(
         "--recursive",
@@ -185,7 +192,12 @@ def main():
 
     elif args.action == "cp":
         u_src = AutoURI(src)
-        _, flag = u_src.cp(target, make_md5_file=args.make_md5_file, return_flag=True)
+        _, flag = u_src.cp(
+            target,
+            make_md5_file=args.make_md5_file,
+            return_flag=True,
+            no_lock=args.no_lock,
+        )
 
         if flag == 0:
             logger.info("Copying from file {s} to {t} done".format(s=src, t=target))
@@ -210,18 +222,20 @@ def main():
             print(uri)
 
     elif args.action == "write":
-        AutoURI(src).write(args.text)
+        AutoURI(src).write(args.text, no_lock=args.no_lock)
         logger.info("Text has been written to {s}".format(s=src))
 
     elif args.action == "rm":
         u = AutoURI(src)
         if not u.exists:
             raise ValueError("File does not exist. {s}".format(s=src))
-        u.rm()
+        u.rm(no_lock=args.no_lock)
         logger.info("Deleted {s}".format(s=src))
 
     elif args.action == "rmdir":
-        AutoURI(src).rmdir(dry_run=not args.delete, num_threads=args.num_threads)
+        AutoURI(src).rmdir(
+            dry_run=not args.delete, num_threads=args.num_threads, no_lock=args.no_lock
+        )
         if not args.delete:
             logger.warning(
                 "rmdir ran in a dry-run mode. "
@@ -234,6 +248,7 @@ def main():
             recursive=args.recursive,
             make_md5_file=args.make_md5_file,
             return_flag=True,
+            no_lock=args.no_lock,
         )
         if localized:
             logger.info("Localized {s} on {t}".format(s=src, t=target))

setup.py

@@ -5,7 +5,7 @@
 
 setuptools.setup(
     name="autouri",
-    version="0.2.2",
+    version="0.2.3",
     python_requires=">=3.6",
     scripts=["bin/autouri"],
     author="Jin wook Lee",
@@ -28,5 +28,6 @@
         "awscli",
         "dateparser",
         "filelock",
+        "six>=1.13.0",
     ],
 )