refactor: use iRODSCommon as parent class

cubi_tk/irods/check.py

@@ -1,19 +1,18 @@
 """``cubi-tk irods check``: Check target iRODS collection (all md5 files? metadata md5 consistent? enough replicas?)."""
 
 import argparse
-from contextlib import contextmanager
 import json
 from multiprocessing.pool import ThreadPool
 import os
 import re
 import typing
 
+from cubi_tk.irods_common import iRODSCommon
 from irods.collection import iRODSCollection
 from irods.column import Like
 from irods.data_object import iRODSDataObject
 from irods.models import Collection as CollectionModel
 from irods.models import DataObject as DataObjectModel
-from irods.session import iRODSSession
 from logzero import logger
 import tqdm
 
@@ -27,43 +26,20 @@
 DEFAULT_HASH_SCHEME = "MD5"
 
 
-class IrodsCheckCommand:
+class IrodsCheckCommand(iRODSCommon):
     """Implementation of iRDOS check command."""
 
     command_name = "check"
 
     def __init__(self, args):
+        super().__init__()
+
         #: Command line arguments.
         self.args = args
 
-        #: Path to iRODS environment file
-        self.irods_env_path = os.path.join(
-            os.path.expanduser("~"), ".irods", "irods_environment.json"
-        )
-
         #: iRODS environment
         self.irods_env = None
 
-    def _init_irods(self):
-        """Connect to iRODS."""
-        try:
-            return iRODSSession(irods_env_file=self.irods_env_path)
-        except Exception as e:
-            logger.error("iRODS connection failed: %s", self.get_irods_error(e))
-            logger.error("Are you logged in? try 'iinit'")
-            raise
-
-    @contextmanager
-    def _get_irods_sessions(self, count=NUM_PARALLEL_TESTS):
-        if count < 1:
-            count = 1
-        irods_sessions = [self._init_irods() for _ in range(count)]
-        try:
-            yield irods_sessions
-        finally:
-            for irods in irods_sessions:
-                irods.cleanup()
-
     @classmethod
     def setup_argparse(cls, parser: argparse.ArgumentParser) -> None:
         parser.add_argument(
@@ -100,12 +76,6 @@ def setup_argparse(cls, parser: argparse.ArgumentParser) -> None:
         )
         parser.add_argument("irods_path", help="Path to an iRODS collection.")
 
-    @classmethod
-    def get_irods_error(cls, e: Exception):
-        """Return logger friendly iRODS exception."""
-        es = str(e)
-        return es if es != "None" else e.__class__.__name__
-
     def get_data_objs(
         self, root_coll: iRODSCollection
     ) -> typing.Dict[

cubi_tk/irods_common.py

@@ -1,5 +1,6 @@
 import getpass
 import os.path
+from contextlib import contextmanager
 from pathlib import Path
 from typing import Iterable
 
@@ -135,6 +136,17 @@ def _save_irods_token(self, token: str):
         else:
             logger.warning("No token found to be saved.")
 
+    @contextmanager
+    def _get_irods_sessions(self, count: int):
+        if count < 1:
+            count = 1
+        irods_sessions = [self._init_irods() for _ in range(count)]
+        try:
+            yield irods_sessions
+        finally:
+            for irods in irods_sessions:
+                irods.cleanup()
+
     @property
     def session(self):
         return self._init_irods()