Channels Reference

Channels

pyzeebe.create_insecure_channel(grpc_address: str = 'UNSET', channel_options: ChannelArgumentType | None = None) → grpc.aio.Channel

Create an insecure channel

Parameters:

grpc_address (Optional[str]) – Zeebe Gateway Address Default: None, alias the ZEEBE_ADDRESS environment variable or “localhost:26500”
channel_options (Optional[ChannelArgumentType]) – GRPC channel options. See https://grpc.github.io/grpc/python/glossary.html#term-channel_arguments

Returns:

A GRPC Channel connected to the Zeebe gateway.

Return type:

grpc.aio.Channel

pyzeebe.create_secure_channel(grpc_address: str = 'UNSET', channel_options: ChannelArgumentType | None = None, channel_credentials: grpc.ChannelCredentials | None = None) → grpc.aio.Channel

Create a secure channel

Parameters:

grpc_address (Optional[str]) – Zeebe Gateway Address Default: None, alias the ZEEBE_ADDRESS environment variable or “localhost:26500”
channel_options (Optional[ChannelArgumentType]) – GRPC channel options. See https://grpc.github.io/grpc/python/glossary.html#term-channel_arguments
channel_credentials (Optional[grpc.ChannelCredentials]) – Channel credentials to use. Will use grpc.ssl_channel_credentials() if not provided.

Returns:

A GRPC Channel connected to the Zeebe gateway.

Return type:

grpc.aio.Channel

pyzeebe.create_oauth2_client_credentials_channel(grpc_address: str = 'UNSET', client_id: str = 'UNSET', client_secret: str = 'UNSET', authorization_server: str = 'UNSET', scope: str | None = 'UNSET', audience: str | None = 'UNSET', channel_credentials: grpc.ChannelCredentials = <grpc.ChannelCredentials object>, channel_options: ChannelArgumentType | None = None, leeway: int = 60, expire_in: int | None = None) → grpc.aio.Channel

Create a gRPC channel for connecting to Camunda 8 (Self-Managed) with OAuth2ClientCredentials.

https://oauth.net/2/grant-types/client-credentials/ https://datatracker.ietf.org/doc/html/rfc6749#section-11.2.2

Parameters:

grpc_address (str, optional) – Zeebe Gateway Address. Defaults to value from ZEEBE_ADDRESS environment variable or “{CAMUNDA_CLUSTER_ID}.{CAMUNDA_CLUSTER_REGION}.zeebe.camunda.io:443” or “localhost:26500”.
client_id (str, optional) – The client id. Defaults to value from CAMUNDA_CLIENT_ID or ZEEBE_CLIENT_ID environment variable
client_secret (str, optional) – The client secret. Defaults to value from CAMUNDA_CLIENT_SECRET or ZEEBE_CLIENT_SECRET environment variable
authorization_server (str, optional) – The authorization server issuing access tokens to the client after successfully authenticating the client. Defaults to value from CAMUNDA_OAUTH_URL or ZEEBE_AUTHORIZATION_SERVER_URL environment variable
scope (str | None, optional) – The scope of the access request.
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_options (ChannelArgumentType | None) – Additional options for the gRPC channel. Defaults to None. See https://grpc.github.io/grpc/python/glossary.html#term-channel_arguments
leeway (int) – The number of seconds to consider the token as expired before the actual expiration time. Defaults to 60.
expire_in (int | None) – The number of seconds the token is valid for. Defaults to None. Should only be used if the token does not contain an “expires_in” attribute.

Returns:

A gRPC channel connected to the Zeebe Gateway.

Return type:

grpc.aio.Channel

Raises:

InvalidOAuthCredentialsError – One of the provided camunda credentials is not correct

pyzeebe.create_camunda_cloud_channel(client_id: str = 'UNSET', client_secret: str = 'UNSET', cluster_id: str = 'UNSET', region: str = 'UNSET', authorization_server: str = 'UNSET', scope: str | None = 'UNSET', audience: str | None = 'UNSET', channel_credentials: grpc.ChannelCredentials = <grpc.ChannelCredentials object>, channel_options: ChannelArgumentType | None = None, leeway: int = 60, expire_in: int | None = None) → grpc.aio.Channel

Create a gRPC channel for connecting to Camunda 8 Cloud (SaaS).

Parameters:

client_id (str, optional) – The client id. Defaults to value from CAMUNDA_CLIENT_ID or ZEEBE_CLIENT_ID environment variable.
client_secret (str, optional) – The client secret. Defaults to value from CAMUNDA_CLIENT_SECRET or ZEEBE_CLIENT_SECRET environment variable.
cluster_id (str, optional) – The ID of the cluster to connect to. Defaults to value from CAMUNDA_CLUSTER_ID environment variable.
region (str, optional) – The region of the cluster. Defaults to value from CAMUNDA_CLUSTER_REGION environment variable or ‘bru-2’.
authorization_server (str, optional) – The authorization server issuing access tokens to the client after successfully authenticating the client. Defaults to value from CAMUNDA_OAUTH_URL or ZEEBE_AUTHORIZATION_SERVER_URL environment variable or “https://login.cloud.camunda.io/oauth/token”.
scope (str | None, optional) – The scope of the access request.
audience (str | None, optional) – The audience for authentication. Defaults to value from CAMUNDA_TOKEN_AUDIENCE 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_options (ChannelArgumentType | None) – Additional options for the gRPC channel. Defaults to None. See https://grpc.github.io/grpc/python/glossary.html#term-channel_arguments
leeway (int) – The number of seconds to consider the token as expired before the actual expiration time. Defaults to 60.
expire_in (int | None) – The number of seconds the token is valid for. Defaults to None. Should only be used if the token does not contain an “expires_in” attribute.

Returns:

The gRPC channel for connecting to Camunda Cloud.

Return type:

grpc.aio.Channel

Credentials

class pyzeebe.credentials.OAuth2MetadataPlugin(oauth2session: OAuth2Session, func_retrieve_token: partial[dict[str, Any]], leeway: int = 60, expire_in: int | None = None)

AuthMetadataPlugin for OAuth2 Authentication.

Implements the AuthMetadataPlugin interface for OAuth2 Authentication based on oauthlib and requests_oauthlib.

__call__(context: AuthMetadataContext, callback: AuthMetadataPluginCallback) → None

Implements authentication by passing metadata to a callback.

This method will be invoked asynchronously in a separate thread.

Parameters:

context – An AuthMetadataContext providing information on the RPC that the plugin is being called to authenticate.
callback – An AuthMetadataPluginCallback to be invoked either synchronously or asynchronously.

__init__(oauth2session: OAuth2Session, func_retrieve_token: partial[dict[str, Any]], leeway: int = 60, expire_in: int | None = None) → None

AuthMetadataPlugin for OAuth2 Authentication.

Parameters:

oauth2session (requests_oauthlib.OAuth2Session) – The OAuth2Session object.
func_fetch_token (Callable) – The function to fetch the token.
leeway (int) – The number of seconds to consider the token as expired before the actual expiration time. Defaults to 60.
expire_in (Optional[int]) – The number of seconds the token is valid for. Defaults to None. Should only be used if the token does not contain an “expires_in” attribute.

_no_expiration(r: Response) → Response

Sets the expiration time for the token if it is not provided in the response.

Parameters:: r (requests.Response) – The response object containing the token.
Returns:: The modified response object with the updated token.
Return type:: requests.Response

is_token_expired() → bool: Check if the token is still valid.

retrieve_token() → None: Retrieve the access token from the authorization server.

class pyzeebe.credentials.Oauth2ClientCredentialsMetadataPlugin(client_id: str, client_secret: str, authorization_server: str, scope: str | None = None, audience: str | None = None, leeway: int = 60, expire_in: int | None = None)

AuthMetadataPlugin for OAuth2 Client Credentials Authentication based on Oauth2MetadataPlugin.

__init__(client_id: str, client_secret: str, authorization_server: str, scope: str | None = None, audience: str | None = None, leeway: int = 60, expire_in: int | None = None)

AuthMetadataPlugin for OAuth2 Client Credentials Authentication based on Oauth2MetadataPlugin.

Parameters:

client_id (str) – The client id.
client_secret (str) – The client secret.
authorization_server (str) – The authorization server issuing access tokens to the client after successfully authenticating the client.
scope (Optional[str]) – The scope of the access request. Defaults to None.
audience (Optional[str]) – The audience for authentication. Defaults to None.
leeway (int) – The number of seconds to consider the token as expired before the actual expiration time. Defaults to 60.
expire_in (Optional[int]) – The number of seconds the token is valid for. Defaults to None. Should only be used if the token does not contain an “expires_in” attribute.

Utilities (Environment)

pyzeebe.channel.utils.get_zeebe_address(default: str | None = None) → str

Get the Zeebe Gateway Address.

Parameters:: default (str, optional) – Default value to be used if no other value was discovered.
Returns:: ZEEBE_ADDRESS environment variable or provided default or “localhost:26500”
Return type:: str

pyzeebe.channel.utils.get_camunda_oauth_url(default: str | None = None) → str

Get the Camunda OAuth URL or Zeebe Authorization Server URL.

Parameters:: default (str, optional) – Default value to be used if no other value was discovered.
Returns:: CAMUNDA_OAUTH_URL or ZEEBE_AUTHORIZATION_SERVER_URL environment variable or provided default
Return type:: str
Raises:: EnvironmentError – If neither CAMUNDA_OAUTH_URL nor ZEEBE_AUTHORIZATION_SERVER_URL is provided.

pyzeebe.channel.utils.get_camunda_client_id() → str

Get the Camunda Client ID.

Returns:: CAMUNDA_CLIENT_ID or ZEEBE_CLIENT_ID environment variable
Return type:: str
Raises:: EnvironmentError – If neither CAMUNDA_CLIENT_ID nor ZEEBE_CLIENT_ID is provided.

pyzeebe.channel.utils.get_camunda_client_secret() → str

Get the Camunda Client Secret.

Returns:: CAMUNDA_CLIENT_SECRET or ZEEBE_CLIENT_SECRET environment variable
Return type:: str
Raises:: EnvironmentError – If neither CAMUNDA_CLIENT_SECRET nor ZEEBE_CLIENT_SECRET is provided.

pyzeebe.channel.utils.get_camunda_cluster_id() → str

Get the Camunda Cluster ID.

Returns:: CAMUNDA_CLUSTER_ID environment variable
Return type:: str
Raises:: EnvironmentError – If CAMUNDA_CLUSTER_ID is not provided.

pyzeebe.channel.utils.get_camunda_cluster_region(default: str | None = None) → str

Get the Camunda Cluster Region.

Parameters:: default (str, optional) – Default value to be used if no other value was discovered.
Returns:: CAMUNDA_CLUSTER_REGION environment variable or provided default
Return type:: str
Raises:: EnvironmentError – If CAMUNDA_CLUSTER_REGION is not provided.

pyzeebe.channel.utils.get_camunda_token_audience(default: str | None = None) → str

Get the Camunda Token Audience.

Parameters:: default (str, optional) – Default value to be used if no other value was discovered.
Returns:: CAMUNDA_TOKEN_AUDIENCE or ZEEBE_TOKEN_AUDIENCE environment variable or provided default
Return type:: str
Raises:: EnvironmentError – If neither CAMUNDA_TOKEN_AUDIENCE nor ZEEBE_TOKEN_AUDIENCE is provided.

pyzeebe.channel.utils.get_camunda_address(cluster_id: str | None = None, cluster_region: str | None = None) → str

Get the Camunda Cloud gRPC server address.

Parameters:

cluster_id (str, optional) – The Camunda cluster ID provided as parameter.
cluster_region (str, optional) – The Camunda cluster region provided as parameter.

Returns:

The Camunda Cloud gRPC server address.

Return type:

str

Raises:

EnvironmentError – If either cluster_id or cluster_region is not provided.