Merge pull request #538 from camunda-community-hub/run-pre-commit

fix: run pre-commit, replace OSError with ValueError

docs/channels_configuration.rst

@@ -17,15 +17,15 @@ Common Variables
 This variables is used across all types of channels:
 
 **ZEEBE_ADDRESS**
-  :Description:  
+  :Description:
     The default address of the Zeebe Gateway.
-  
-  :Usage:        
+
+  :Usage:
     Used in both secure and insecure channel configurations.
     :func:`pyzeebe.create_insecure_channel`
     :func:`pyzeebe.create_secure_channel`
-  
-  :Default:      
+
+  :Default:
     ``"localhost:26500"``
 
 Common OAuth2 Variables
@@ -34,19 +34,19 @@ Common OAuth2 Variables
 These variables are specifically for connecting to generic OAuth2 or Camunda Cloud instances.
 
 **CAMUNDA_CLIENT_ID** / **ZEEBE_CLIENT_ID**
-  :Description:  
+  :Description:
     The client ID required for OAuth2 client credential authentication.
-  
-  :Usage:        
+
+  :Usage:
     Required for OAuth2 and Camunda Cloud channels.
     :func:`pyzeebe.create_oauth2_client_credentials_channel`
     :func:`pyzeebe.create_camunda_cloud_channel`
 
 **CAMUNDA_CLIENT_SECRET** / **ZEEBE_CLIENT_SECRET**
-  :Description:  
+  :Description:
     The client secret for the OAuth2 client.
-  
-  :Usage:        
+
+  :Usage:
     Required for OAuth2 and Camunda Cloud channels.
     :func:`pyzeebe.create_oauth2_client_credentials_channel`
     :func:`pyzeebe.create_camunda_cloud_channel`
@@ -57,22 +57,22 @@ OAuth2 Variables (Self-Managed)
 These variables are primarily used for OAuth2 authentication in self-managed Camunda 8 instances.
 
 **CAMUNDA_OAUTH_URL** / **ZEEBE_AUTHORIZATION_SERVER_URL**
-  :Description:  
+  :Description:
     Specifies the URL of the authorization server issuing access tokens to the client.
-  
-  :Usage:        
+
+  :Usage:
     Required if channel initialization argument was not specified.
     :func:`pyzeebe.create_oauth2_client_credentials_channel`
 
 **CAMUNDA_TOKEN_AUDIENCE** / **ZEEBE_TOKEN_AUDIENCE**
-  :Description:  
+  :Description:
     Specifies the audience for the OAuth2 token.
-  
-  :Usage:        
+
+  :Usage:
     Used when creating OAuth2 or Camunda Cloud channels.
     :func:`pyzeebe.create_oauth2_client_credentials_channel`
-  
-  :Default:      
+
+  :Default:
     ``None`` if not provided.
 
 Camunda Cloud Variables (SaaS)
@@ -81,43 +81,42 @@ Camunda Cloud Variables (SaaS)
 These variables are specifically for connecting to Camunda Cloud instances.
 
 **CAMUNDA_OAUTH_URL** / **ZEEBE_AUTHORIZATION_SERVER_URL**
-  :Description:  
+  :Description:
     Specifies the URL of the authorization server issuing access tokens to the client.
-  
-  :Usage:        
+
+  :Usage:
     Used in the OAuth2 and Camunda Cloud channel configurations.
     :func:`pyzeebe.create_camunda_cloud_channel`
-  
-  :Default:      
+
+  :Default:
     ``"https://login.cloud.camunda.io/oauth/token"`` if not specified.
 
 **CAMUNDA_CLUSTER_ID**
-  :Description:  
+  :Description:
     The unique identifier for the Camunda Cloud cluster to connect to.
-  
-  :Usage:        
+
+  :Usage:
     Required if channel initialization argument was not specified.
     :func:`pyzeebe.create_camunda_cloud_channel`
 
 **CAMUNDA_CLUSTER_REGION**
-  :Description:  
+  :Description:
     The region where the Camunda Cloud cluster is hosted.
-  
-  :Usage:        
+
+  :Usage:
     Required for Camunda Cloud channels.
     :func:`pyzeebe.create_camunda_cloud_channel`
-  
-  :Default:      
+
+  :Default:
     ``"bru-2"`` if not provided.
 
 **CAMUNDA_TOKEN_AUDIENCE** / **ZEEBE_TOKEN_AUDIENCE**
-  :Description:  
+  :Description:
     Specifies the audience for the OAuth2 token.
-  
-  :Usage:        
+
+  :Usage:
     Used when creating OAuth2 or Camunda Cloud channels.
     :func:`pyzeebe.create_camunda_cloud_channel`
-
-  :Default:      
-    ``"zeebe.camunda.io"`` if not provided.
 
+  :Default:
+    ``"zeebe.camunda.io"`` if not provided.

docs/errors.rst

@@ -10,6 +10,8 @@ All ``pyzeebe`` errors inherit from :py:class:`PyZeebeError`
 
 .. autoexception:: pyzeebe.errors.NoVariableNameGivenError
 
+.. autoexception:: pyzeebe.errors.SettingsError
+
 .. autoexception:: pyzeebe.errors.DuplicateTaskTypeError
 
 .. autoexception:: pyzeebe.errors.ActivateJobsRequestInvalidError

pyzeebe/channel/oauth_channel.py

@@ -26,7 +26,7 @@ def create_oauth2_client_credentials_channel(
     authorization_server: str = Unset,
     scope: str | None = Unset,
     audience: str | None = Unset,
-    channel_credentials: grpc.ChannelCredentials = grpc.ssl_channel_credentials(),
+    channel_credentials: grpc.ChannelCredentials | None = None,
     channel_options: ChannelArgumentType | None = None,
     leeway: int = 60,
     expire_in: int | None = None,
@@ -52,8 +52,8 @@ def create_oauth2_client_credentials_channel(
         audience (str | None, optional): The audience for authentication.
             Defaults to value from CAMUNDA_TOKEN_AUDIENCE or ZEEBE_TOKEN_AUDIENCE environment variable
 
-        channel_credentials (grpc.ChannelCredentials): The gRPC channel credentials.
-            Defaults to grpc.ssl_channel_credentials().
+        channel_credentials (grpc.ChannelCredentials | None): The gRPC channel credentials.
+            Defaults to `grpc.ssl_channel_credentials`.
         channel_options (ChannelArgumentType | None): Additional options for the gRPC channel.
             Defaults to None.
             See https://grpc.github.io/grpc/python/glossary.html#term-channel_arguments
@@ -120,7 +120,7 @@ def create_camunda_cloud_channel(
     authorization_server: str = Unset,
     scope: str | None = Unset,
     audience: str | None = Unset,
-    channel_credentials: grpc.ChannelCredentials = grpc.ssl_channel_credentials(),
+    channel_credentials: grpc.ChannelCredentials | None = None,
     channel_options: ChannelArgumentType | None = None,
     leeway: int = 60,
     expire_in: int | None = None,
@@ -147,8 +147,8 @@ def create_camunda_cloud_channel(
             or ZEEBE_TOKEN_AUDIENCE environment variable
             or "zeebe.camunda.io".
 
-        channel_credentials (grpc.ChannelCredentials): The gRPC channel credentials.
-            Defaults to grpc.ssl_channel_credentials().
+        channel_credentials (grpc.ChannelCredentials | None): The gRPC channel credentials.
+            Defaults to `grpc.ssl_channel_credentials`.
         channel_options (ChannelArgumentType | None): Additional options for the gRPC channel.
             Defaults to None.
             See https://grpc.github.io/grpc/python/glossary.html#term-channel_arguments

pyzeebe/channel/utils.py

@@ -2,6 +2,8 @@
 
 import os
 
+from pyzeebe.errors import SettingsError
+
 DEFAULT_ZEEBE_ADDRESS = "localhost:26500"
 
 
@@ -29,12 +31,12 @@ def get_camunda_oauth_url(default: str | None = None) -> str:
         str: CAMUNDA_OAUTH_URL or ZEEBE_AUTHORIZATION_SERVER_URL environment variable or provided default
 
     Raises:
-        EnvironmentError: If neither CAMUNDA_OAUTH_URL nor ZEEBE_AUTHORIZATION_SERVER_URL is provided.
+        SettingsError: If neither CAMUNDA_OAUTH_URL nor ZEEBE_AUTHORIZATION_SERVER_URL is provided.
     """
     r = os.getenv("CAMUNDA_OAUTH_URL") or os.getenv("ZEEBE_AUTHORIZATION_SERVER_URL") or default
 
     if r is None:
-        raise EnvironmentError("No CAMUNDA_OAUTH_URL or ZEEBE_AUTHORIZATION_SERVER_URL provided!")
+        raise SettingsError("No CAMUNDA_OAUTH_URL or ZEEBE_AUTHORIZATION_SERVER_URL provided!")
 
     return r
 
@@ -47,12 +49,12 @@ def get_camunda_client_id() -> str:
         str: CAMUNDA_CLIENT_ID or ZEEBE_CLIENT_ID environment variable
 
     Raises:
-        EnvironmentError: If neither CAMUNDA_CLIENT_ID nor ZEEBE_CLIENT_ID is provided.
+        SettingsError: If neither CAMUNDA_CLIENT_ID nor ZEEBE_CLIENT_ID is provided.
     """
     r = os.getenv("CAMUNDA_CLIENT_ID") or os.getenv("ZEEBE_CLIENT_ID")
 
     if r is None:
-        raise EnvironmentError("No CAMUNDA_CLIENT_ID or ZEEBE_CLIENT_ID provided!")
+        raise SettingsError("No CAMUNDA_CLIENT_ID or ZEEBE_CLIENT_ID provided!")
 
     return r
 
@@ -65,12 +67,12 @@ def get_camunda_client_secret() -> str:
         str: CAMUNDA_CLIENT_SECRET or ZEEBE_CLIENT_SECRET environment variable
 
     Raises:
-        EnvironmentError: If neither CAMUNDA_CLIENT_SECRET nor ZEEBE_CLIENT_SECRET is provided.
+        SettingsError: If neither CAMUNDA_CLIENT_SECRET nor ZEEBE_CLIENT_SECRET is provided.
     """
     r = os.getenv("CAMUNDA_CLIENT_SECRET") or os.getenv("ZEEBE_CLIENT_SECRET")
 
     if r is None:
-        raise EnvironmentError("No CAMUNDA_CLIENT_SECRET or ZEEBE_CLIENT_SECRET provided!")
+        raise SettingsError("No CAMUNDA_CLIENT_SECRET or ZEEBE_CLIENT_SECRET provided!")
 
     return r
 
@@ -83,12 +85,12 @@ def get_camunda_cluster_id() -> str:
         str: CAMUNDA_CLUSTER_ID environment variable
 
     Raises:
-        EnvironmentError: If CAMUNDA_CLUSTER_ID is not provided.
+        SettingsError: If CAMUNDA_CLUSTER_ID is not provided.
     """
     r = os.getenv("CAMUNDA_CLUSTER_ID")
 
     if r is None:
-        raise EnvironmentError("No CAMUNDA_CLUSTER_ID provided!")
+        raise SettingsError("No CAMUNDA_CLUSTER_ID provided!")
 
     return r
 
@@ -104,12 +106,12 @@ def get_camunda_cluster_region(default: str | None = None) -> str:
         str: CAMUNDA_CLUSTER_REGION environment variable or provided default
 
     Raises:
-        EnvironmentError: If CAMUNDA_CLUSTER_REGION is not provided.
+        SettingsError: If CAMUNDA_CLUSTER_REGION is not provided.
     """
     r = os.getenv("CAMUNDA_CLUSTER_REGION") or default
 
     if r is None:
-        raise EnvironmentError("No CAMUNDA_CLUSTER_REGION provided!")
+        raise SettingsError("No CAMUNDA_CLUSTER_REGION provided!")
 
     return r
 
@@ -125,12 +127,12 @@ def get_camunda_token_audience(default: str | None = None) -> str:
         str: CAMUNDA_TOKEN_AUDIENCE or ZEEBE_TOKEN_AUDIENCE environment variable or provided default
 
     Raises:
-        EnvironmentError: If neither CAMUNDA_TOKEN_AUDIENCE nor ZEEBE_TOKEN_AUDIENCE is provided.
+        SettingsError: If neither CAMUNDA_TOKEN_AUDIENCE nor ZEEBE_TOKEN_AUDIENCE is provided.
     """
     r = os.getenv("CAMUNDA_TOKEN_AUDIENCE") or os.getenv("ZEEBE_TOKEN_AUDIENCE") or default
 
     if r is None:
-        raise EnvironmentError("No CAMUNDA_TOKEN_AUDIENCE or ZEEBE_TOKEN_AUDIENCE provided!")
+        raise SettingsError("No CAMUNDA_TOKEN_AUDIENCE or ZEEBE_TOKEN_AUDIENCE provided!")
 
     return r
 
@@ -147,9 +149,9 @@ def get_camunda_address(cluster_id: str | None = None, cluster_region: str | Non
         str: The Camunda Cloud gRPC server address.
 
     Raises:
-        EnvironmentError: If either cluster_id or cluster_region is not provided.
+        SettingsError: If either cluster_id or cluster_region is not provided.
     """
     if (cluster_id is None) or (cluster_region is None):
-        raise EnvironmentError("The cluster_id and cluster_region must be provided!")
+        raise SettingsError("The cluster_id and cluster_region must be provided!")
 
     return f"{cluster_id}.{cluster_region}.zeebe.camunda.io:443"

pyzeebe/errors/__init__.py

@@ -18,6 +18,7 @@
     DuplicateTaskTypeError,
     NoVariableNameGivenError,
     PyZeebeError,
+    SettingsError,
     TaskNotFoundError,
 )
 from .zeebe_errors import (
@@ -44,6 +45,7 @@
     "DuplicateTaskTypeError",
     "NoVariableNameGivenError",
     "PyZeebeError",
+    "SettingsError",
     "TaskNotFoundError",
     "UnknownGrpcStatusCodeError",
     "ZeebeBackPressureError",

pyzeebe/errors/pyzeebe_errors.py

@@ -9,6 +9,10 @@ class TaskNotFoundError(PyZeebeError):
     pass
 
 
+class SettingsError(PyZeebeError):
+    pass
+
+
 class NoVariableNameGivenError(PyZeebeError):
     def __init__(self, task_type: str):
         super().__init__(f"No variable name given for single_value task {task_type}")

pyzeebe/types.py

@@ -5,6 +5,6 @@
 
 Headers: TypeAlias = Mapping[str, Any]
 Variables: TypeAlias = Mapping[str, Any]
-Unset = str("UNSET")
+Unset = "UNSET"
 
 ChannelArgumentType: TypeAlias = Sequence[tuple[str, Any]]