Merge pull request #558 from camunda-community-hub/allow-pathlike

fix(typing): allow to pass os.PathLike to deploy_resource

pyzeebe/client/client.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import os
 from collections.abc import Iterable
 
 import grpc
@@ -124,7 +125,9 @@ async def cancel_process_instance(self, process_instance_key: int) -> CancelProc
         """
         return await self.zeebe_adapter.cancel_process_instance(process_instance_key=process_instance_key)
 
-    async def deploy_resource(self, *resource_file_path: str, tenant_id: str | None = None) -> DeployResourceResponse:
+    async def deploy_resource(
+        self, *resource_file_path: str | os.PathLike[str], tenant_id: str | None = None
+    ) -> DeployResourceResponse:
         """
         Deploy one or more processes
 

pyzeebe/client/sync_client.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+import os
 from functools import partial, wraps
 
 import grpc
@@ -55,7 +56,9 @@ def cancel_process_instance(self, process_instance_key: int) -> CancelProcessIns
         return self.loop.run_until_complete(self.client.cancel_process_instance(process_instance_key))
 
     @copy_docstring(ZeebeClient.deploy_resource)
-    def deploy_resource(self, *resource_file_path: str, tenant_id: str | None = None) -> DeployResourceResponse:
+    def deploy_resource(
+        self, *resource_file_path: str | os.PathLike[str], tenant_id: str | None = None
+    ) -> DeployResourceResponse:
         return self.loop.run_until_complete(self.client.deploy_resource(*resource_file_path, tenant_id=tenant_id))
 
     @copy_docstring(ZeebeClient.broadcast_signal)

pyzeebe/grpc_internals/zeebe_process_adapter.py

@@ -128,7 +128,9 @@ async def cancel_process_instance(self, process_instance_key: int) -> CancelProc
 
         return CancelProcessInstanceResponse()
 
-    async def deploy_resource(self, *resource_file_path: str, tenant_id: str | None = None) -> DeployResourceResponse:
+    async def deploy_resource(
+        self, *resource_file_path: str | os.PathLike[str], tenant_id: str | None = None
+    ) -> DeployResourceResponse:
         try:
             response = await self._gateway_stub.DeployResource(
                 DeployResourceRequest(
@@ -218,6 +220,6 @@ def _create_form_from_raw_form(response: FormMetadata) -> DeployResourceResponse
 }
 
 
-async def _create_resource_request(resource_file_path: str) -> Resource:
+async def _create_resource_request(resource_file_path: str | os.PathLike[str]) -> Resource:
     async with await anyio.open_file(resource_file_path, "rb") as file:
         return Resource(name=os.path.basename(resource_file_path), content=await file.read())