cas_wrapper.py

@@ -4,12 +4,6 @@ import logging
 import typing
 
 from plumbum import local, ProcessExecutionError
-from pydantic import BaseModel
-
-
-class CasArtifact(BaseModel):
-    path: str
-    cas_hash: typing.Optional[str]
 
 
 class CasWrapper:
@@ -37,16 +31,12 @@ class CasWrapper:
         if self._logger is None:
             self._logger = logging.getLogger()
 
-    def __enter__(self):
+    def ensure_login(self):
         with local.env(
             CAS_API_KEY=self._cas_api_key,
             SIGNER_ID=self._cas_signer_id,
         ):
             self._cas['login']()
-        return self
-
-    def __exit__(self, exc_type, value, traceback):
-        pass
 
     def notarize(
             self,
@@ -125,17 +115,17 @@ class CasWrapper:
     ) -> typing.Tuple[bool, typing.Optional[str]]:
         is_authenticated = False
         commit_cas_hash = None
-        with self as cas:
-            try:
-                result_json = cas.authenticate(local_path, return_json=True)
-                # it should return 0 for authenticated and trusted commits
-                is_authenticated = not bool(
-                    result_json.get('status', 1))
-                commit_cas_hash = result_json.get('hash')
-            # we can fall with ProcessExecutionError,
-            # because source can be not notarized
-            except ProcessExecutionError:
-                self._logger.exception('Cannot authenticate %s:', local_path)
+        self.ensure_login()
+        try:
+            result_json = self.authenticate(local_path, return_json=True)
+            # it should return 0 for authenticated and trusted commits
+            is_authenticated = not bool(
+                result_json.get('status', 1))
+            commit_cas_hash = result_json.get('hash')
+        # we can fall with ProcessExecutionError,
+        # because source can be not notarized
+        except ProcessExecutionError:
+            self._logger.exception('Cannot authenticate %s:', local_path)
         return is_authenticated, commit_cas_hash
 
     def authenticate_artifact(
@@ -145,36 +135,39 @@ class CasWrapper:
         return_json: bool = False,
     ) -> bool:
         is_authenticated = False
-        with self as cas:
-            try:
-                is_authenticated = cas.authenticate(local_path,
-                                                    use_hash=use_hash,
-                                                    return_json=return_json)
-            # we can fall with ProcessExecutionError,
-            # because source can be not notarized
-            except ProcessExecutionError:
-                self._logger.exception('Cannot authenticate %s:', local_path)
+        self.ensure_login()
+        try:
+            is_authenticated = self.authenticate(
+                local_path,
+                use_hash=use_hash,
+                return_json=return_json,
+            )
+        # we can fall with ProcessExecutionError,
+        # because source can be not notarized
+        except ProcessExecutionError:
+            self._logger.exception('Cannot authenticate %s:', local_path)
         return is_authenticated
 
     def notarize_artifacts(
         self,
-        artifacts: typing.List[CasArtifact],
+        artifact_paths: typing.List[str],
         metadata: typing.Dict[str, typing.Any],
-    ) -> bool:
+    ) -> typing.Tuple[bool, typing.Dict[str, str]]:
         all_artifacts_is_notarized = True
-        with self as cas, ThreadPoolExecutor(max_workers=4) as executor:
+        notarized_artifacts = {}
+        self.ensure_login()
+        with ThreadPoolExecutor(max_workers=4) as executor:
             futures = {
-                executor.submit(cas.notarize, artifact.path, metadata): artifact
-                for artifact in artifacts
-                if not artifact.cas_hash
+                executor.submit(self.notarize, artifact_path, metadata): artifact_path
+                for artifact_path in artifact_paths
             }
             for future in as_completed(futures):
-                artifact = futures[future]
+                artifact_path = futures[future]
                 try:
                     cas_artifact_hash = future.result()
                 except Exception:
                     self._logger.exception('Cannot notarize artifact:')
                     all_artifacts_is_notarized = False
                     continue
-                artifact.cas_hash = cas_artifact_hash
-        return all_artifacts_is_notarized
+                notarized_artifacts[artifact_path] = cas_artifact_hash
+        return all_artifacts_is_notarized, notarized_artifacts

setup.py

@@ -21,7 +21,6 @@ setup(
     scripts=['cas_wrapper.py'],
     install_requires=[
         'plumbum>=1.7.2',
-        'pydantic>=1.8.1',
     ],
     python_requires=">=3.6",
 )