Replace "PERMISSION_DENIED" with const

Author: tsmith023

weaviate/collections/batch/grpc_batch_delete.py

@@ -9,7 +9,7 @@
 from weaviate.collections.classes.config import ConsistencyLevel
 from weaviate.collections.classes.filters import _Filters
 from weaviate.collections.filters import _FilterToGRPC
-from weaviate.collections.grpc.shared import _BaseGRPC
+from weaviate.collections.grpc.shared import _BaseGRPC, PERMISSION_DENIED
 from weaviate.collections.queries.base import _WeaviateUUIDInt
 from weaviate.connect import ConnectionV4
 from weaviate.exceptions import WeaviateDeleteManyError, InsufficientPermissionsError
@@ -62,6 +62,6 @@ async def batch_delete(
                 )
 
         except AioRpcError as e:
-            if e.code().name == "PERMISSION_DENIED":
+            if e.code().name == PERMISSION_DENIED:
                 raise InsufficientPermissionsError(e)
             raise WeaviateDeleteManyError(str(e))

weaviate/collections/batch/grpc_batch_objects.py

@@ -15,7 +15,7 @@
 from weaviate.collections.classes.config import ConsistencyLevel
 from weaviate.collections.classes.internal import ReferenceToMulti, ReferenceInputs
 from weaviate.collections.classes.types import GeoCoordinate, PhoneNumber
-from weaviate.collections.grpc.shared import _BaseGRPC
+from weaviate.collections.grpc.shared import _BaseGRPC, PERMISSION_DENIED
 from weaviate.connect import ConnectionV4
 from weaviate.exceptions import (
     WeaviateBatchError,
@@ -153,7 +153,7 @@ async def __send_batch(
                 objects[result.index] = result.error
             return objects
         except AioRpcError as e:
-            if e.code().name == "PERMISSION_DENIED":
+            if e.code().name == PERMISSION_DENIED:
                 raise InsufficientPermissionsError(e)
             raise WeaviateBatchError(str(e)) from e
 

weaviate/collections/grpc/query.py

@@ -48,7 +48,7 @@
 )
 from weaviate.collections.filters import _FilterToGRPC
 from weaviate.collections.grpc.retry import _Retry
-from weaviate.collections.grpc.shared import _BaseGRPC
+from weaviate.collections.grpc.shared import _BaseGRPC, PERMISSION_DENIED
 from weaviate.connect import ConnectionV4
 from weaviate.exceptions import (
     InsufficientPermissionsError,
@@ -812,7 +812,7 @@ async def __call(self, request: search_get_pb2.SearchRequest) -> search_get_pb2.
             )
             return cast(search_get_pb2.SearchReply, res)
         except AioRpcError as e:
-            if e.code().name == "PERMISSION_DENIED":
+            if e.code().name == PERMISSION_DENIED:
                 raise InsufficientPermissionsError(e)
             raise WeaviateQueryError(str(e), "GRPC search")  # pyright: ignore
         except WeaviateRetryError as e:

weaviate/collections/grpc/shared.py

@@ -4,6 +4,8 @@
 from weaviate.connect import ConnectionV4
 from weaviate.proto.v1 import base_pb2
 
+PERMISSION_DENIED = "PERMISSION_DENIED"
+
 
 class _BaseGRPC:
     def __init__(

weaviate/collections/grpc/tenants.py

@@ -5,7 +5,7 @@
 from weaviate.collections.classes.config import ConsistencyLevel
 from weaviate.collections.classes.tenants import TenantActivityStatus
 from weaviate.collections.grpc.retry import _Retry
-from weaviate.collections.grpc.shared import _BaseGRPC
+from weaviate.collections.grpc.shared import _BaseGRPC, PERMISSION_DENIED
 from weaviate.connect import ConnectionV4
 from weaviate.exceptions import WeaviateTenantGetError, InsufficientPermissionsError
 from weaviate.proto.v1 import tenants_pb2
@@ -38,7 +38,7 @@ async def get(self, names: Optional[Sequence[str]]) -> tenants_pb2.TenantsGetRep
                 timeout=self._connection.timeout_config.query,
             )
         except AioRpcError as e:
-            if e.code().name == "PERMISSION_DENIED":
+            if e.code().name == PERMISSION_DENIED:
                 raise InsufficientPermissionsError(e)
             raise WeaviateTenantGetError(str(e)) from e