Convert the sync client into a wrapper of the async

[acroca]

This PR introduces anyio to reduce duplication between the clients. The sync client is now just a wrapper, using the async client behind the scenes.

I kept all the functions explicitly defined in the sync client so the code editors can show the tips to users. We could generate those functions dynamically and create a .pyi file, but I think it's manageable and simpler to duplicate them.

Opening as draft to get feedback on this approach.

[filintod]

nice effort and in general people use asyncio code so better to optimize for that path, and slowly consolidate

[Copilot]

Pull request overview

This PR refactors the synchronous client to be a thin wrapper around the async client using anyio's start_blocking_portal() to bridge sync and async code. The core orchestration logic (OrchestrationStatus, OrchestrationState, OrchestrationFailedError) has been moved from the sync client to the async client, eliminating code duplication.

Changes:

• Moved OrchestrationStatus, OrchestrationState, OrchestrationFailedError, and related helper functions from durabletask/client.py to durabletask/aio/client.py
• Converted TaskHubGrpcClient to a wrapper that uses anyio's blocking portal to call the async client methods
• Updated tests to use AsyncMock instead of Mock and import OrchestrationStatus from durabletask.aio.client
• Added anyio>=4.0.0,<5 as a dependency

Reviewed changes

Copilot reviewed 6 out of 6 changed files in this pull request and generated 8 comments.

Show a summary per file

File	Description
durabletask/client.py	Replaced the full sync implementation with a wrapper that uses anyio's blocking portal to call async methods
durabletask/aio/client.py	Moved OrchestrationStatus, OrchestrationState, OrchestrationFailedError and helper functions from sync client to centralize the implementation
tests/durabletask/test_orchestration_wait.py	Updated to use AsyncMock and access internal _async_client._stub for mocking
tests/durabletask/test_orchestration_e2e.py	Updated to import OrchestrationStatus from durabletask.aio.client
tests/durabletask/test_orchestration_e2e_async.py	Updated import to get OrchestrationStatus from durabletask.aio.client instead of durabletask.client
pyproject.toml	Added anyio>=4.0.0,<5 dependency

Comments suppressed due to low confidence (1)

tests/durabletask/test_orchestration_wait.py:69

• The tests create a TaskHubGrpcClient but never call close() on it, which will leave the portal and async resources open. This can lead to resource leaks and warnings about unclosed resources. Consider using a context manager or explicitly calling close() at the end of the test, or use a pytest fixture with proper cleanup.

    c = TaskHubGrpcClient()
    c._async_client._stub = AsyncMock()
    c._async_client._stub.WaitForInstanceStart.return_value = response

    grpc_timeout = None if timeout is None else timeout
    c.wait_for_orchestration_start(instance_id, timeout=grpc_timeout)

    # Verify WaitForInstanceStart was called with timeout=None
    c._async_client._stub.WaitForInstanceStart.assert_called_once()
    _, kwargs = c._async_client._stub.WaitForInstanceStart.call_args
    if timeout is None or timeout == 0:
        assert kwargs.get("timeout") is None
    else:
        assert kwargs.get("timeout") == timeout


@pytest.mark.parametrize("timeout", [None, 0, 5])
def test_wait_for_orchestration_completion_timeout(timeout):
    instance_id = "test-instance"

    from durabletask.internal.orchestrator_service_pb2 import (
        ORCHESTRATION_STATUS_COMPLETED,
        GetInstanceResponse,
        OrchestrationState,
    )

    response = GetInstanceResponse()
    state = OrchestrationState()
    state.instanceId = instance_id
    state.orchestrationStatus = ORCHESTRATION_STATUS_COMPLETED
    response.orchestrationState.CopyFrom(state)

    c = TaskHubGrpcClient()
    c._async_client._stub = AsyncMock()
    c._async_client._stub.WaitForInstanceCompletion.return_value = response

    grpc_timeout = None if timeout is None else timeout
    c.wait_for_orchestration_completion(instance_id, timeout=grpc_timeout)

    # Verify WaitForInstanceStart was called with timeout=None
    c._async_client._stub.WaitForInstanceCompletion.assert_called_once()
    _, kwargs = c._async_client._stub.WaitForInstanceCompletion.call_args
    if timeout is None or timeout == 0:
        assert kwargs.get("timeout") is None
    else:
        assert kwargs.get("timeout") == timeout

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

The sync client no longer exports OrchestrationStatus and OrchestrationFailedError, which are part of the public API. Examples in the repository use client.OrchestrationStatus.COMPLETED, and this change will break existing code. These classes should be re-exported from durabletask.aio.client at the module level to maintain backward compatibility.

Suggested change

	OrchestrationState,
	OrchestrationState,
	OrchestrationStatus,
	OrchestrationFailedError,

[Copilot]

Using del for cleanup with blocking operations is problematic. The del method may be called during interpreter shutdown when the event loop and other resources may be unavailable, potentially causing deadlocks or hangs. The close() method involves blocking calls to the portal which may not be safe during garbage collection. Consider removing del and relying on context managers or explicit close() calls, or at the minimum, add checks to ensure the portal is still valid before attempting cleanup.

Suggested change

	# Best-effort cleanup in case the user didn't call close() or use context manager
	try:
	self.close()
	except Exception:
	# Avoid blocking operations during garbage collection; just log if not properly closed.
	try:
	async_client = getattr(self, "_async_client", None)
	portal = getattr(self, "_portal", None)
	portal_cm = getattr(self, "_portal_cm", None)
	if async_client is not None or portal is not None or portal_cm is not None:
	logger = logging.getLogger(__name__)
	logger.warning(
	"TaskHubGrpcClient was not closed explicitly; resources may not be fully released. "
	"Use 'with TaskHubGrpcClient(...) as client:' or call 'client.close()' explicitly."
	)
	except Exception:
	# Suppress all exceptions in destructor

[Copilot]

The portal setup should handle exceptions during async client creation. If _create_async_client() raises an exception, the portal will remain open but self._async_client will not be set. This will cause issues in the close() method. Wrap the async client creation in a try-except block and ensure proper portal cleanup on failure.

Suggested change

	self._portal_cm = start_blocking_portal()
	self._portal = self._portal_cm.__enter__()
	self._async_client = self._portal.call(self._create_async_client)
	# Initialize portal-related attributes to safe defaults
	self._portal_cm = None
	self._portal = None
	self._async_client = None
	# Set up the blocking portal and async client, ensuring cleanup on failure
	self._portal_cm = start_blocking_portal()
	self._portal = self._portal_cm.__enter__()
	try:
	self._async_client = self._portal.call(self._create_async_client)
	except Exception:
	# If async client creation fails, make sure to close the portal
	try:
	if self._portal_cm is not None:
	self._portal_cm.__exit__(None, None, None)
	finally:
	self._portal = None
	self._portal_cm = None
	raise

[Copilot]

The synchronous wrapper uses start_blocking_portal() which spawns a new thread with its own event loop. This design means every TaskHubGrpcClient instance creates its own background thread, which could lead to scalability issues if many clients are created. Consider documenting this behavior in the class docstring, and potentially adding a warning if multiple instances are created to help users understand the resource implications.

[Copilot]

The sync e2e tests create TaskHubGrpcClient instances but never call close() on them. Since the new implementation uses a portal with a background event loop thread, these resources will leak. Consider using context managers (with statements) or explicitly calling close() at the end of each test to properly clean up resources.

[Copilot]

The async method _create_async_client is called from init via portal.call(), but it doesn't need to be async since it only constructs an object and doesn't perform any awaitable operations. The AsyncTaskHubGrpcClient.init is synchronous. This async function could be made synchronous, simplifying the code and avoiding the overhead of scheduling it on the event loop.

[Copilot]

The close() method has a potential issue where it checks if self._async_client is not None but doesn't guard against the client being uninitialized. If init fails before setting self._async_client, or if close() is called multiple times, this could raise AttributeError. Consider adding a hasattr check or initializing self._async_client = None before the portal setup.

[Copilot]

There are numerous type: ignore comments throughout the wrapper methods. This suggests type checking issues with the portal.call() pattern. Consider adding proper type hints to _portal and _async_client attributes, or using typing.cast() where appropriate to make the types explicit, which would improve code maintainability and IDE support.