Workflow management python SDK

[RyanLettieri]

Description

This PR covers the implementation of the management portion for workflow.

Note that this PR will fail until the following PRs are submitted:
dapr/dapr#6163
dapr/components-contrib#2729

Issue reference

Please reference the issue this PR will close: #542

Checklist

Please make sure you've completed the relevant tasks for this PR, out of the following list:

• Code compiles correctly
• Created/updated tests
• Extended the documentation

[berndverst]

Please add the async implementation in dapr.aio.clients.grpc.client.py and update the tests in tests/clients/test_dapr_async_grpc_client.py

After that it would be awesome if for a change we update your example to use async instead of synchronous.

from dapr.aio.clients import DaprClient
async with DaprClient() as d:
  await d.wait()
  await d.get_workflow()
  print("stuff")

etc -- you get the idea

[berndverst]

As also discussed in chat - you must use grpcio-tools==1.48.2 to generate the protos.

You used version 1.49.1 of grpcio-tools which generates protos not compatible with our version of protobuf. Please make sure to use grpcio-tools<1.49 and regenerate the protos.

[berndverst]

dapr/clients/grpc/_response.py:963: error: Too many arguments for "__init__" of "object"  [call-arg]
dapr/clients/grpc/_response.py:989: error: Too many arguments for "__init__" of "object"  [call-arg]

[berndverst]

This is a super generic response - how does someone know whether the termination request was received properly?

If there something in the protos that provides a better response?

[cgillum]

LGTM

[berndverst]

The Python SDK always uses Union[bytes, str] for any value which is bytes in the proto.

We should maintain this.

So if the data is str you probably need to convert it to bytes first. The bytes and None cases are otherwise handled correctly.

It is better to let a user manually define how they wish to convert any other object to bytes or string themselves, rather than doing this for the user. This will reduce a lot of support burden as users will be confused how another object type was encoded and then be unable to decode it.

[cgillum]

@berndverst please reconsider regarding support for Any and implicit serialization for the workflow management APIs.

The workflow authoring SDKs already support implicit deserialization of JSON data. In fact, they do not support reading the binary data directly, in spite of the fact that the proto contract defines these payloads as binary. Disallowing implicit serialization in these APIs would mean that every developer would have to write JSON serialization code in their client apps, which is a bunch of boilerplate that they might get wrong and file issues for. The Dapr Client SDKs will be much more usable if we can take care of the serialization for them equally on both sides.

FWIW, we do the implicit client-side serialization in the .NET Dapr SDK already. Having the same behavior in the Python SDK would be nice for consistency.

[berndverst]

Here is a better idea:

If you replace Any with Dict then you still have convenience for the user, but you don't support completely arbitrary type conversions (which is a bad idea).

Please update the signature to Optional[Union[Dict, bytes, str]]

I believe other than replacing Any for Dict, and changing this signature (and doc strings) you might not need to make any other code changes here.

[berndverst]

Suggested change

	if isinstance(input, bytes):
	if send_raw_bytes and isinstance(input, bytes):

We decided to go a different direction.

[berndverst]

also add parameter send_raw_bytes: bool = False

[berndverst]

Suggested change

	encoded_data = json.dumps(input).encode("utf-8") if input is not None else bytes([])
	try:
	encoded_data = json.dumps(input).encode("utf-8") if input is not None else bytes([])
	except TypeError:
	raise DaprInternalError(message='start_workflow input must be json serializable')

Add try except around this step

In the except block, please throw a DaprInteralError exception with a message indicating that the input value must be json serializable.

[berndverst]

Suggested change

	event_data: Union[Any, bytes, None] = None) -> DaprResponse:
	event_data: Optional[Union[Any, bytes]] = None) -> DaprResponse:

[berndverst]

Suggested change

	input: Union[Any, bytes, None] = None,
	input: Optional[Union[Any, bytes]] = None,

[berndverst]

This still needs updating

[berndverst]

add param send_raw_bytes: bool = False

[berndverst]

Suggested change

	if isinstance(input, bytes):
	if `send_raw_data and isinstance(input, bytes):`

[berndverst]

Suggested change

	event_data: Union[Any, bytes, None] = None) -> DaprResponse:
	event_data: Optional[Union[Any, bytes]] = None) -> DaprResponse:

[berndverst]

combine the following lines into the single line statement

Add try / except block as before

[berndverst]

encoded_data = json.dumps(input).encode("utf-8") if input is not None else bytes([])

[berndverst]

Suggested change

	event_data: Union[Any, bytes, None] = None,
	event_data: Optional[Union[Any, bytes]] = None,

[berndverst]

Suggested change

	input: Union[Any, bytes, None] = None,
	input: Optional[Union[Any, bytes]] = None,

[berndverst]

Suggested change

	event_data: Union[Any, bytes, None] = None,
	event_data: Optional[Union[Any, bytes]] = None,

[berndverst]

Suggested change

	raise DaprInternalError("Json Data is not serializable")
	raise DaprInternalError("raise_workflow_event: event_data must be JSON serializable")

[berndverst]

Suggested change

	raise DaprInternalError(f"Json serialization error: {e}")
	raise DaprInternalError(f"raise_workflow_event: JSON serialization error: {e}")

[berndverst]

Suggested change

	raise DaprInternalError("Json Data is not serializable")
	raise DaprInternalError("start_workflow: input must be JSON serializable")

[berndverst]

Suggested change

	raise DaprInternalError(f"Json serialization error: {e}")
	raise DaprInternalError(f"start_workflow: JSON serialization error: {e}")

[cgillum]

LGTM

[berndverst]

update