Merge pull request #29 from JonatanMartens/development

1.2.0

.github/workflows/test-zeebe-integration.yml

@@ -13,7 +13,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        zeebe-version: [ "0.23.5", "0.24.2" ]
+        zeebe-version: [ "0.23.6", "0.24.3" ]
         python-version: [ 3.5, 3.6, 3.7, 3.8 ]
 
     container: python:${{ matrix.python-version }}

LICENSE.md

@@ -1,6 +1,6 @@
 MIT License
 
-Copyright (c) [year] [fullname]
+Copyright (c) 2020 Jonatan Martens
 
 Permission is hereby granted, free of charge, to any person obtaining a copy
 of this software and associated documentation files (the "Software"), to deal

Pipfile

@@ -17,3 +17,5 @@ pyzeebe = {editable = true, path = "."}
 [packages]
 grpcio = "~=1.31.0"
 protobuf = "~=3.13.0"
+oauthlib = "~=3.1.0"
+requests-oauthlib = "~=1.3.0"

README.md

@@ -16,7 +16,7 @@ Zeebe version support:
 
 | Pyzeebe version | Tested Zeebe versions |
 |:---------------:|----------------|
-| 1.1.0           | 0.23, 0.24         |
+| 1.x.x           | 0.23, 0.24         |
 
 ## Getting Started
 To install:

examples/client.py

@@ -1,8 +1,16 @@
-from pyzeebe import ZeebeClient
+from pyzeebe import ZeebeClient, CamundaCloudCredentials
 
-# Create a zeebe client
+# Create a zeebe client without credentials
 zeebe_client = ZeebeClient(hostname='localhost', port=26500)
 
+# Create a zeebe client with TLS
+zeebe_client = ZeebeClient(hostname='localhost', port=26500, secure_connection=True)
+
+# Create a zeebe client for camunda cloud
+camunda_cloud_credentials = CamundaCloudCredentials(client_id='<my_client_id>', client_secret='<my_client_secret>',
+                                                    cluster_id='<my_cluster_id>')
+zeebe_client = ZeebeClient(credentials=camunda_cloud_credentials)
+
 # Run a workflow
 workflow_instance_key = zeebe_client.run_workflow(bpmn_process_id='My zeebe workflow', variables={})
 

examples/worker.py

@@ -1,22 +1,46 @@
 from typing import Dict
 
-from pyzeebe import Task, TaskContext, TaskStatusController, ZeebeWorker
+from pyzeebe import Task, TaskContext, TaskStatusController, ZeebeWorker, CamundaCloudCredentials
 
 
 def example_task() -> Dict:
-    return {'output': f'Hello world, test!'}
+    return {"output": f"Hello world, test!"}
 
 
 def example_exception_handler(exc: Exception, context: TaskContext, controller: TaskStatusController) -> None:
     print(exc)
     print(context)
-    controller.error(f'Failed to run task {context.type}. Reason: {exc}')
+    controller.error(f"Failed to run task {context.type}. Reason: {exc}")
 
 
-task = Task(task_type='test', task_handler=example_task, exception_handler=example_exception_handler)
+task = Task(task_type="test", task_handler=example_task, exception_handler=example_exception_handler)
 
-worker = ZeebeWorker()  # Will use environment variable ZEEBE_ADDRESS or localhost:26500
 
+# Use decorators to add functionality before and after tasks. These will not fail the task
+def example_logging_task_decorator(task_context: TaskContext) -> TaskContext:
+    print(task_context)
+    return task_context
+
+
+task.before(example_logging_task_decorator)
+task.after(example_logging_task_decorator)
+
+# Will use environment variable ZEEBE_ADDRESS or localhost:26500 and NOT use TLS
+worker = ZeebeWorker()
+
+# Will use environment variable ZEEBE_ADDRESS or localhost:26500 and use TLS
+worker = ZeebeWorker(secure_connection=True)
+
+# Connect to zeebe cluster in camunda cloud
+camunda_cloud_credentials = CamundaCloudCredentials(client_id="<my_client_id>", client_secret="<my_client_secret>",
+                                                    cluster_id="<my_cluster_id>")
+worker = ZeebeWorker(credentials=camunda_cloud_credentials)
+
+# We can also use decorators on workers. These decorators will happen before all tasks
+worker.before(example_logging_task_decorator)
+worker.after(example_logging_task_decorator)
+
+# Add task to worker
 worker.add_task(task)
 
 if __name__ == '__main__':

pyzeebe/__init__.py

@@ -1,5 +1,7 @@
 from pyzeebe.client.client import ZeebeClient
 from pyzeebe.common import exceptions
+from pyzeebe.credentials.camunda_cloud_credentials import CamundaCloudCredentials
+from pyzeebe.credentials.oauth_credentials import OAuthCredentials
 from pyzeebe.task.task import Task
 from pyzeebe.task.task_context import TaskContext
 from pyzeebe.task.task_status_controller import TaskStatusController

pyzeebe/client/client.py

@@ -2,12 +2,15 @@
 
 import grpc
 
+from pyzeebe.credentials.base_credentials import BaseCredentials
 from pyzeebe.grpc_internals.zeebe_adapter import ZeebeAdapter
 
 
 class ZeebeClient(object):
-    def __init__(self, hostname: str = None, port: int = None, channel: grpc.Channel = None):
-        self.zeebe_adapter = ZeebeAdapter(hostname=hostname, port=port, channel=channel)
+    def __init__(self, hostname: str = None, port: int = None, credentials: BaseCredentials = None,
+                 channel: grpc.Channel = None, secure_connection: bool = False):
+        self.zeebe_adapter = ZeebeAdapter(hostname=hostname, port=port, credentials=credentials, channel=channel,
+                                          secure_connection=secure_connection)
 
     def run_workflow(self, bpmn_process_id: str, variables: Dict = None, version: int = -1) -> int:
         return self.zeebe_adapter.create_workflow_instance(bpmn_process_id=bpmn_process_id, variables=variables or {},

pyzeebe/credentials/base_credentials.py

@@ -0,0 +1,11 @@
+from abc import ABC, abstractmethod
+
+from grpc import ChannelCredentials
+
+
+class BaseCredentials(ABC):
+    grpc_credentials: ChannelCredentials
+
+    @abstractmethod
+    def get_connection_uri(self) -> str:
+        pass

pyzeebe/credentials/camunda_cloud_credentials.py

@@ -0,0 +1,10 @@
+from pyzeebe.credentials.oauth_credentials import OAuthCredentials
+
+
+class CamundaCloudCredentials(OAuthCredentials):
+    def __init__(self, client_id: str, client_secret: str, cluster_id: str):
+        super().__init__(url="https://login.cloud.camunda.io/oauth/token", client_id=client_id,
+                         client_secret=client_secret, audience=f"{cluster_id}.zeebe.camunda.io")
+
+    def get_connection_uri(self) -> str:
+        return f"{self.audience}:443"

pyzeebe/credentials/camunda_cloud_credentials_test.py

@@ -0,0 +1,15 @@
+from unittest.mock import patch
+from uuid import uuid4
+
+from pyzeebe.credentials.camunda_cloud_credentials import CamundaCloudCredentials
+
+
+def test_init():
+    client_id = str(uuid4())
+    client_secret = str(uuid4())
+    cluster_id = str(uuid4())
+
+    with patch("pyzeebe.credentials.oauth_credentials.OAuthCredentials.__init__") as init:
+        CamundaCloudCredentials(client_id, client_secret, cluster_id)
+        init.assert_called_with(url=f"https://login.cloud.camunda.io/oauth/token", client_id=client_id,
+                                client_secret=client_secret, audience=f"{cluster_id}.zeebe.camunda.io")

pyzeebe/credentials/oauth_credentials.py

@@ -0,0 +1,33 @@
+import grpc
+from oauthlib import oauth2
+from requests_oauthlib import OAuth2Session
+
+from pyzeebe.credentials.base_credentials import BaseCredentials
+
+
+class OAuthCredentials(BaseCredentials):
+    def __init__(self, url: str, client_id: str, client_secret: str, audience: str):
+        self.url = url
+        self.client_id = client_id
+        self.client_secret = client_secret
+        self.audience = audience
+
+        self.access_token = self.get_access_token(url, client_id, client_secret, audience)
+        token_credentials = grpc.access_token_call_credentials(self.access_token)
+        ssl_credentials = grpc.ssl_channel_credentials()
+        self.grpc_credentials = grpc.composite_channel_credentials(ssl_credentials, token_credentials)
+
+    @staticmethod
+    def get_access_token(url: str, client_id: str, client_secret: str, audience: str) -> str:
+        client = oauth2.BackendApplicationClient(client_id)
+        client.prepare_request_body(include_client_id=True)
+        with OAuth2Session(client=client) as session:
+            return session.post(url,
+                                data={
+                                    "client_id": client_id,
+                                    "client_secret": client_secret,
+                                    "audience": audience
+                                }).json()["access_token"]
+
+    def get_connection_uri(self) -> str:
+        return None

pyzeebe/credentials/oauth_credentials_test.py

@@ -0,0 +1,19 @@
+from unittest.mock import patch
+from uuid import uuid4
+
+from pyzeebe.credentials.oauth_credentials import OAuthCredentials
+
+
+def test_get_access_token():
+    with patch('requests_oauthlib.OAuth2Session.post') as post_mock:
+        url = f"https://{str(uuid4())}/oauth/token"
+        client_id = str(uuid4())
+        client_secret = str(uuid4())
+        audience = str(uuid4())
+        OAuthCredentials.get_access_token(url=url, client_id=client_id, client_secret=client_secret, audience=audience)
+        post_mock.assert_called_with(url,
+                                     data={
+                                         'client_id': client_id,
+                                         'client_secret': client_secret,
+                                         'audience': audience
+                                     })