Consolidate exception classes

conda-store-server/conda_store_server/_internal/orm.py

@@ -40,6 +40,7 @@
 )
 
 from conda_store_server._internal import conda_utils, schema, utils
+from conda_store_server.exception import BuildPathError
 from conda_store_server._internal.environment import validate_environment
 
 logger = logging.getLogger("orm")
@@ -316,7 +317,7 @@ def build_path(self, conda_store):
         # conda prefix must be less or equal to 255 chars
         # https://github.com/conda-incubator/conda-store/issues/649
         if len(str(res)) > 255:
-            raise utils.BuildPathError("build_path too long: must be <= 255 characters")
+            raise BuildPathError("build_path too long: must be <= 255 characters")
         # Note: cannot use the '/' operator to prepend the extended-length
         # prefix
         if sys.platform == "win32" and conda_store.win_extended_length_prefix:

conda-store-server/conda_store_server/_internal/schema.py

@@ -20,7 +20,8 @@
     ValidationError,
 )
 
-from conda_store_server._internal import conda_utils, utils
+from conda_store_server._internal import conda_utils
+from conda_store_server.exception import CondaStoreError
 
 
 def _datetime_factory(offset: datetime.timedelta):
@@ -403,7 +404,7 @@ def model_validate(cls, specification):
 
                 all_errors_hr.append(human_readable_error)
 
-            raise utils.CondaStoreError(all_errors_hr)
+            raise CondaStoreError(all_errors_hr)
 
 
 class LockfileSpecification(BaseModel):

conda-store-server/conda_store_server/_internal/server/views/api.py

@@ -17,6 +17,7 @@
 from conda_store_server._internal.schema import AuthenticationToken, Permissions
 from conda_store_server._internal.server import dependencies
 from conda_store_server.server.auth import Authentication
+from conda_store_server.exception import CondaStoreError
 
 
 class PaginatedArgs(TypedDict):
@@ -620,7 +621,7 @@ async def api_delete_namespace(
 
         try:
             conda_store.delete_namespace(db, namespace)
-        except utils.CondaStoreError as e:
+        except CondaStoreError as e:
             raise HTTPException(status_code=400, detail=e.message)
 
         return {"status": "ok"}
@@ -786,7 +787,7 @@ async def api_update_environment_build(
                     db, namespace, name, description
                 )
 
-        except utils.CondaStoreError as e:
+        except CondaStoreError as e:
             raise HTTPException(status_code=400, detail=e.message)
 
         return {"status": "ok"}
@@ -813,7 +814,7 @@ async def api_delete_environment(
 
         try:
             conda_store.delete_environment(db, namespace, name)
-        except utils.CondaStoreError as e:
+        except CondaStoreError as e:
             raise HTTPException(status_code=400, detail=e.message)
 
         return {"status": "ok"}
@@ -899,7 +900,7 @@ async def api_post_specification(
                 specification = schema.CondaSpecification.model_validate(specification)
         except yaml.error.YAMLError:
             raise HTTPException(status_code=400, detail="Unable to parse. Invalid YAML")
-        except utils.CondaStoreError as e:
+        except CondaStoreError as e:
             raise HTTPException(status_code=400, detail="\n".join(e.args[0]))
         except pydantic.ValidationError as e:
             raise HTTPException(status_code=400, detail=str(e))
@@ -921,7 +922,7 @@ async def api_post_specification(
             )
         except ValueError as e:
             raise HTTPException(status_code=400, detail=str(e.args[0]))
-        except utils.CondaStoreError as e:
+        except CondaStoreError as e:
             raise HTTPException(status_code=400, detail=str(e.message))
 
         return {"status": "ok", "data": {"build_id": build_id}}
@@ -1020,7 +1021,7 @@ async def api_put_build(
             new_build = conda_store.create_build(
                 db, build.environment_id, build.specification.sha256
             )
-        except utils.CondaStoreError as e:
+        except CondaStoreError as e:
             raise HTTPException(status_code=400, detail=e.message)
 
         return {
@@ -1115,7 +1116,7 @@ async def api_delete_build(
 
         try:
             conda_store.delete_build(db, build_id)
-        except utils.CondaStoreError as e:
+        except CondaStoreError as e:
             raise HTTPException(status_code=400, detail=e.message)
 
         return {"status": "ok"}

conda-store-server/conda_store_server/_internal/utils.py

@@ -18,16 +18,6 @@
 from filelock import FileLock
 
 
-class CondaStoreError(Exception):
-    @property
-    def message(self):
-        return self.args[0]
-
-
-class BuildPathError(CondaStoreError):
-    pass
-
-
 def symlink(source, target):
     # Multiple builds call this, so this lock avoids race conditions on unlink
     # and symlink operations

conda-store-server/conda_store_server/_internal/worker/build.py

@@ -18,6 +18,7 @@
 
 from conda_store_server import api
 from conda_store_server._internal import action, conda_utils, orm, schema, utils
+from conda_store_server.exception import BuildPathError
 from conda_store_server.plugins import plugin_context
 
 
@@ -322,7 +323,7 @@ def build_conda_environment(db: Session, conda_store, build):
         conda_store.log.exception(e)
         append_to_logs(db, conda_store, build, e.output)
         raise e
-    except utils.BuildPathError as e:
+    except BuildPathError as e:
         # Provide status_info, which will be exposed to the user, ONLY in this
         # case because the error message doesn't expose sensitive information
         set_build_failed(db, build, status_info=e.message)

conda-store-server/conda_store_server/app.py

@@ -26,6 +26,7 @@
 from traitlets.config import LoggingConfigurable
 
 from conda_store_server import CONDA_STORE_DIR, BuildKey, api, storage
+from conda_store_server.exception import CondaStoreError
 from conda_store_server._internal import conda_utils, environment, orm, schema, utils
 from conda_store_server.plugins import hookspec, plugin_manager
 from conda_store_server.plugins.types import lock
@@ -65,7 +66,7 @@ def conda_store_validate_action(
         schema.Permissions.ENVIRONMENT_CREATE,
         schema.Permissions.ENVIRONMENT_UPDATE,
     ) and (settings.storage_threshold > system_metrics.disk_free):
-        raise utils.CondaStoreError(
+        raise CondaStoreError(
             f"`CondaStore.storage_threshold` reached. Action {action.value} prevented due to insufficient storage space"
         )
 
@@ -749,21 +750,21 @@ def update_environment_build(
 
         build = api.get_build(db, build_id)
         if build is None:
-            raise utils.CondaStoreError(f"build id={build_id} does not exist")
+            raise CondaStoreError(f"build id={build_id} does not exist")
 
         environment = api.get_environment(db, namespace=namespace, name=name)
         if environment is None:
-            raise utils.CondaStoreError(
+            raise CondaStoreError(
                 f"environment namespace={namespace} name={name} does not exist"
             )
 
         if build.status != schema.BuildStatus.COMPLETED:
-            raise utils.CondaStoreError(
+            raise CondaStoreError(
                 "cannot update environment to build id since not completed"
             )
 
         if build.specification.name != name:
-            raise utils.CondaStoreError(
+            raise CondaStoreError(
                 "cannot update environment to build id since specification does not match environment name"
             )
 
@@ -781,7 +782,7 @@ def update_environment_description(
     ):
         environment = api.get_environment(db, namespace=namespace, name=name)
         if environment is None:
-            raise utils.CondaStoreError(
+            raise CondaStoreError(
                 f"environment namespace={namespace} name={name} does not exist"
             )
 
@@ -798,7 +799,7 @@ def delete_namespace(self, db: Session, namespace: str):
 
         namespace = api.get_namespace(db, name=namespace)
         if namespace is None:
-            raise utils.CondaStoreError(f"namespace={namespace} does not exist")
+            raise CondaStoreError(f"namespace={namespace} does not exist")
 
         utcnow = datetime.datetime.utcnow()
         namespace.deleted_on = utcnow
@@ -825,7 +826,7 @@ def delete_environment(self, db: Session, namespace: str, name: str):
 
         environment = api.get_environment(db, namespace=namespace, name=name)
         if environment is None:
-            raise utils.CondaStoreError(
+            raise CondaStoreError(
                 f"environment namespace={namespace} name={name} does not exist"
             )
 
@@ -856,7 +857,7 @@ def delete_build(self, db: Session, build_id: int):
             schema.BuildStatus.FAILED,
             schema.BuildStatus.COMPLETED,
         ]:
-            raise utils.CondaStoreError(
+            raise CondaStoreError(
                 "cannot delete build since not finished building"
             )
 

conda-store-server/conda_store_server/exception.py

@@ -2,18 +2,30 @@
 # Use of this source code is governed by a BSD-style
 # license that can be found in the LICENSE file.
 
+import typing
+
 
 class CondaStoreError(Exception):
     pass
 
 
+class BuildPathError(CondaStoreError):
+    """Exception raised by conda store when there is an issue with the requested build path
+    Attributes:
+        message -- message to output to users
+    """
+
+    def __init__(self, msg: str):
+        super().__init__(msg)
+
+
 class CondaStorePluginNotFoundError(CondaStoreError):
     """Exception raised by conda store when a specified plugin is not found
     Attributes:
         plugin -- plugin that was not found
         available_plugins -- list of registered plugins
     """
 
-    def __init__(self, plugin, available_plugins):
+    def __init__(self, plugin: str, available_plugins: typing.List[str]):
         self.message = f"Plugin {plugin} was requested but not found! The following plugins are available: {', '.join(available_plugins)}"
         super().__init__(self.message)

conda-store-server/tests/test_api.py

@@ -6,7 +6,7 @@
 
 from conda_store_server import api
 from conda_store_server._internal.orm import NamespaceRoleMapping
-from conda_store_server._internal.utils import BuildPathError
+from conda_store_server.exception import BuildPathError
 
 
 @pytest.fixture