New to Python SDK and couldn’t get a sample HelloWorld working. Any help is appreciated:
Error message is at the bottom.
async def run():
config = read_config_file("local_test.yaml")
client = await Client.connect("localhost:7233")
async with Worker(
client,
task_queue=config.get_client_machine_name(),
activities=[basic_activity],
workflows=[BasicJobWorkflow],
):
result = await client.execute_workflow(
BasicJobWorkflow.run,
"test input",
id=add_uuid_to_string("test_id"),
task_queue=config.get_client_machine_name(),
)
print(f"Result: {result}")
return result
def test_main():
asyncio.run(run())
@activity.defn
async def basic_activity(s:str) → str :
return s + " activity is completed"
@workflow.defn
class BasicJobWorkflow:
@workflow.run
async def run(self, s: str) -> str:
print("BasicJobWorkflow() started:" + s)
return s + " is completed"
Error message ------------
Error:
venv/lib/python3.9/site-packages/temporalio/client.py:2055: in fetch_next_page
history_event_filter_type=temporalio.api.enums.v1.HistoryEventFilterType.ValueType(
self = <google.protobuf.internal.enum_type_wrapper.EnumTypeWrapper object at 0x7f4024437be0>
name = ‘ValueType’
def __getattr__(self, name):
"""Returns the value corresponding to the given enum name."""
try:
return super(
EnumTypeWrapper,
self).__getattribute__('_enum_type').values_by_name[name].number
except KeyError:
pass # fall out to break exception chaining
raise AttributeError('Enum {} has no value defined for name {!r}'.format(
self._enum_type.name, name))
E AttributeError: Enum HistoryEventFilterType has no value defined for name ‘ValueType’
venv/lib/python3.9/site-packages/google/protobuf/internal/enum_type_wrapper.py:114: AttributeError
============================== 1 failed in 0.42s ===============================
Process finished with exit code 1