Channels Reference

Channels

pyzeebe.create_insecure_channel(grpc_address: str | None = None, 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 | None = None, 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, client_id: str, client_secret: str, authorization_server: str, scope: str | None = None, audience: str | None = None, channel_credentials: grpc.ChannelCredentials | None = None, 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.

Parameters:

grpc_address (str) – Zeebe Gateway Address.
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.
channel_credentials (grpc.ChannelCredentials) – The gRPC channel credentials. Defaults to grpc.ssl_channel_credentials().
channel_options (Optional[ChannelArgumentType]) – 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 (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.

Returns:

A gRPC channel connected to the Zeebe Gateway.

Return type:

grpc.aio.Channel

pyzeebe.create_camunda_cloud_channel(client_id: str, client_secret: str, cluster_id: str, region: str = 'bru-2', authorization_server: str = 'https://login.cloud.camunda.io/oauth/token', scope: str | None = None, audience: str | None = 'zeebe.camunda.io', channel_credentials: grpc.ChannelCredentials | None = None, 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) – The client id.
client_secret (str) – The client secret.
cluster_id (str) – The ID of the cluster to connect to.
region (Optional[str]) – The region of the cluster. Defaults to “bru-2”.
authorization_server (Optional[str]) – The authorization server issuing access tokens to the client after successfully authenticating the client. Defaults to “https://login.cloud.camunda.io/oauth/token”.
scope (Optional[str]) – The scope of the access request. Can be set to CAMUNDA_CLUSTER_ID. Defaults to None.
audience (Optional[str]) – The audience for authentication. Defaults to “zeebe.camunda.io”.
channel_credentials (grpc.ChannelCredentials) – The gRPC channel credentials. Defaults to grpc.ssl_channel_credentials().
channel_options (Optional[ChannelArgumentType]) – 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 (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.

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.