From afd969a25708acb9c0e94faf55db190704225f44 Mon Sep 17 00:00:00 2001 From: Moe Dashti Date: Fri, 13 Nov 2026 17:38:38 -0700 Subject: [PATCH 1/2] Added the generated code of the Streams API. These are the files that `poe gen-protos` writes for the Stream service. They are in a PR of their own, so that the hand-written change shows alone. --- temporalio/api/enums/v1/__init__.py | 4 + temporalio/api/enums/v1/stream_pb2.py | 31 + temporalio/api/enums/v1/stream_pb2.pyi | 175 +++ temporalio/api/stream/__init__.py | 0 temporalio/api/stream/v1/__init__.py | 29 + temporalio/api/stream/v1/message_pb2.py | 58 + temporalio/api/stream/v1/message_pb2.pyi | 465 ++++++ temporalio/api/streamservice/__init__.py | 0 temporalio/api/streamservice/v1/__init__.py | 25 + .../streamservice/v1/request_response_pb2.py | 54 + .../streamservice/v1/request_response_pb2.pyi | 446 ++++++ .../v1/request_response_pb2_grpc.py | 4 + .../v1/request_response_pb2_grpc.pyi | 21 + temporalio/bridge/_visitor.py | 64 + .../nexus/system/stream_service/__init__.py | 71 + .../_resource_context_manager.py | 93 ++ .../system/stream_service/client/__init__.py | 11 + .../stream_service/client/_extensions.py | 180 +++ .../system/stream_service/client/_poll.py | 77 + .../client/_resources/__init__.py | 3 + .../client/_resources/stream.py | 261 ++++ .../nexus/system/stream_service/models.py | 1371 +++++++++++++++++ .../nexus/system/stream_service/services.py | 53 + .../stream_service/workflow/__init__.py | 77 + .../workflow/_resources/__init__.py | 9 + .../workflow/_resources/stream.py | 101 ++ .../workflow/_system_nexus_interceptor.py | 299 ++++ .../workflow/operations/__init__.py | 3 + .../workflow/operations/create_stream.py | 90 ++ temporalio/workflow/__init__.py | 6 + 30 files changed, 4081 insertions(+) create mode 100644 temporalio/api/enums/v1/stream_pb2.py create mode 100644 temporalio/api/enums/v1/stream_pb2.pyi create mode 100644 temporalio/api/stream/__init__.py create mode 100644 temporalio/api/stream/v1/__init__.py create mode 100644 temporalio/api/stream/v1/message_pb2.py create mode 100644 temporalio/api/stream/v1/message_pb2.pyi create mode 100644 temporalio/api/streamservice/__init__.py create mode 100644 temporalio/api/streamservice/v1/__init__.py create mode 100644 temporalio/api/streamservice/v1/request_response_pb2.py create mode 100644 temporalio/api/streamservice/v1/request_response_pb2.pyi create mode 100644 temporalio/api/streamservice/v1/request_response_pb2_grpc.py create mode 100644 temporalio/api/streamservice/v1/request_response_pb2_grpc.pyi create mode 100644 temporalio/nexus/system/stream_service/__init__.py create mode 100644 temporalio/nexus/system/stream_service/_resource_context_manager.py create mode 100644 temporalio/nexus/system/stream_service/client/__init__.py create mode 100644 temporalio/nexus/system/stream_service/client/_extensions.py create mode 100644 temporalio/nexus/system/stream_service/client/_poll.py create mode 100644 temporalio/nexus/system/stream_service/client/_resources/__init__.py create mode 100644 temporalio/nexus/system/stream_service/client/_resources/stream.py create mode 100644 temporalio/nexus/system/stream_service/models.py create mode 100644 temporalio/nexus/system/stream_service/services.py create mode 100644 temporalio/nexus/system/stream_service/workflow/__init__.py create mode 100644 temporalio/nexus/system/stream_service/workflow/_resources/__init__.py create mode 100644 temporalio/nexus/system/stream_service/workflow/_resources/stream.py create mode 100644 temporalio/nexus/system/stream_service/workflow/_system_nexus_interceptor.py create mode 100644 temporalio/nexus/system/stream_service/workflow/operations/__init__.py create mode 100644 temporalio/nexus/system/stream_service/workflow/operations/create_stream.py diff --git a/temporalio/api/enums/v1/__init__.py b/temporalio/api/enums/v1/__init__.py index 518a2f32bc..a48c7a44d1 100644 --- a/temporalio/api/enums/v1/__init__.py +++ b/temporalio/api/enums/v1/__init__.py @@ -44,6 +44,7 @@ from .query_pb2 import QueryRejectCondition, QueryResultType from .reset_pb2 import ResetReapplyExcludeType, ResetReapplyType, ResetType from .schedule_pb2 import ScheduleOverlapPolicy +from .stream_pb2 import StreamErrorReason, StreamIdConflictPolicy, StreamStatus from .task_queue_pb2 import ( BuildIdTaskReachability, DescribeTaskQueueMode, @@ -120,6 +121,9 @@ "Severity", "SignalExternalWorkflowExecutionFailedCause", "StartChildWorkflowExecutionFailedCause", + "StreamErrorReason", + "StreamIdConflictPolicy", + "StreamStatus", "SuggestContinueAsNewReason", "TaskQueueKind", "TaskQueueType", diff --git a/temporalio/api/enums/v1/stream_pb2.py b/temporalio/api/enums/v1/stream_pb2.py new file mode 100644 index 0000000000..b2e79a0e96 --- /dev/null +++ b/temporalio/api/enums/v1/stream_pb2.py @@ -0,0 +1,31 @@ +# -*- coding: utf-8 -*- +# Generated by the protocol buffer compiler. DO NOT EDIT! +# source: temporal/api/enums/v1/stream.proto +"""Generated protocol buffer code.""" + +from google.protobuf import descriptor as _descriptor +from google.protobuf import descriptor_pool as _descriptor_pool +from google.protobuf import symbol_database as _symbol_database +from google.protobuf.internal import builder as _builder + +# @@protoc_insertion_point(imports) + +_sym_db = _symbol_database.Default() + + +DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile( + b'\n"temporal/api/enums/v1/stream.proto\x12\x15temporal.api.enums.v1*z\n\x0cStreamStatus\x12\x1d\n\x19STREAM_STATUS_UNSPECIFIED\x10\x00\x12\x16\n\x12STREAM_STATUS_OPEN\x10\x01\x12\x18\n\x14STREAM_STATUS_CLOSED\x10\x02\x12\x19\n\x15STREAM_STATUS_EXPIRED\x10\x03*\x93\x01\n\x16StreamIdConflictPolicy\x12)\n%STREAM_ID_CONFLICT_POLICY_UNSPECIFIED\x10\x00\x12"\n\x1eSTREAM_ID_CONFLICT_POLICY_FAIL\x10\x01\x12*\n&STREAM_ID_CONFLICT_POLICY_USE_EXISTING\x10\x02*\xe8\x04\n\x11StreamErrorReason\x12#\n\x1fSTREAM_ERROR_REASON_UNSPECIFIED\x10\x00\x12!\n\x1dSTREAM_ERROR_REASON_NOT_FOUND\x10\x01\x12\x1f\n\x1bSTREAM_ERROR_REASON_EXPIRED\x10\x02\x12\x1e\n\x1aSTREAM_ERROR_REASON_CLOSED\x10\x03\x12\x1e\n\x1aSTREAM_ERROR_REASON_FENCED\x10\x04\x12(\n$STREAM_ERROR_REASON_SEQUENCE_TOO_LOW\x10\x05\x12%\n!STREAM_ERROR_REASON_SEQ_DIVERGENT\x10\x06\x12!\n\x1dSTREAM_ERROR_REASON_NOT_OWNER\x10\x07\x12&\n"STREAM_ERROR_REASON_FORMAT_TOO_NEW\x10\x08\x12\'\n#STREAM_ERROR_REASON_OFFSET_TOO_HIGH\x10\t\x12!\n\x1dSTREAM_ERROR_REASON_TOO_LARGE\x10\n\x12\x1f\n\x1bSTREAM_ERROR_REASON_INVALID\x10\x0b\x12+\n\'STREAM_ERROR_REASON_STORAGE_UNAVAILABLE\x10\x0c\x12#\n\x1fSTREAM_ERROR_REASON_UNSUPPORTED\x10\r\x12&\n"STREAM_ERROR_REASON_ALREADY_EXISTS\x10\x0e\x12\'\n#STREAM_ERROR_REASON_STORAGE_CORRUPT\x10\x0f\x42\x83\x01\n\x18io.temporal.api.enums.v1B\x0bStreamProtoP\x01Z!go.temporal.io/api/enums/v1;enums\xaa\x02\x17Temporalio.Api.Enums.V1\xea\x02\x1aTemporalio::Api::Enums::V1b\x06proto3' +) + +_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, globals()) +_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, __name__, globals()) +if _descriptor._USE_C_DESCRIPTORS == False: + DESCRIPTOR._options = None + DESCRIPTOR._serialized_options = b"\n\030io.temporal.api.enums.v1B\013StreamProtoP\001Z!go.temporal.io/api/enums/v1;enums\252\002\027Temporalio.Api.Enums.V1\352\002\032Temporalio::Api::Enums::V1" + _STREAMSTATUS._serialized_start = 61 + _STREAMSTATUS._serialized_end = 183 + _STREAMIDCONFLICTPOLICY._serialized_start = 186 + _STREAMIDCONFLICTPOLICY._serialized_end = 333 + _STREAMERRORREASON._serialized_start = 336 + _STREAMERRORREASON._serialized_end = 952 +# @@protoc_insertion_point(module_scope) diff --git a/temporalio/api/enums/v1/stream_pb2.pyi b/temporalio/api/enums/v1/stream_pb2.pyi new file mode 100644 index 0000000000..ac57ccd183 --- /dev/null +++ b/temporalio/api/enums/v1/stream_pb2.pyi @@ -0,0 +1,175 @@ +""" +@generated by mypy-protobuf. Do not edit manually! +isort:skip_file +""" + +import builtins +import sys +import typing + +import google.protobuf.descriptor +import google.protobuf.internal.enum_type_wrapper + +if sys.version_info >= (3, 10): + import typing as typing_extensions +else: + import typing_extensions + +DESCRIPTOR: google.protobuf.descriptor.FileDescriptor + +class _StreamStatus: + ValueType = typing.NewType("ValueType", builtins.int) + V: typing_extensions.TypeAlias = ValueType + +class _StreamStatusEnumTypeWrapper( + google.protobuf.internal.enum_type_wrapper._EnumTypeWrapper[ + _StreamStatus.ValueType + ], + builtins.type, +): + DESCRIPTOR: google.protobuf.descriptor.EnumDescriptor + STREAM_STATUS_UNSPECIFIED: _StreamStatus.ValueType # 0 + STREAM_STATUS_OPEN: _StreamStatus.ValueType # 1 + STREAM_STATUS_CLOSED: _StreamStatus.ValueType # 2 + STREAM_STATUS_EXPIRED: _StreamStatus.ValueType # 3 + +class StreamStatus(_StreamStatus, metaclass=_StreamStatusEnumTypeWrapper): + """Experimental. Might change in the future. + The state of a Stream, as a describe gives it. + (-- api-linter: core::0216::synonyms=disabled + aip.dev/not-precedent: Named consistently with WorkflowExecutionStatus. --) + """ + +STREAM_STATUS_UNSPECIFIED: StreamStatus.ValueType # 0 +STREAM_STATUS_OPEN: StreamStatus.ValueType # 1 +STREAM_STATUS_CLOSED: StreamStatus.ValueType # 2 +STREAM_STATUS_EXPIRED: StreamStatus.ValueType # 3 +global___StreamStatus = StreamStatus + +class _StreamIdConflictPolicy: + ValueType = typing.NewType("ValueType", builtins.int) + V: typing_extensions.TypeAlias = ValueType + +class _StreamIdConflictPolicyEnumTypeWrapper( + google.protobuf.internal.enum_type_wrapper._EnumTypeWrapper[ + _StreamIdConflictPolicy.ValueType + ], + builtins.type, +): + DESCRIPTOR: google.protobuf.descriptor.EnumDescriptor + STREAM_ID_CONFLICT_POLICY_UNSPECIFIED: _StreamIdConflictPolicy.ValueType # 0 + STREAM_ID_CONFLICT_POLICY_FAIL: _StreamIdConflictPolicy.ValueType # 1 + """The create fails with ALREADY_EXISTS.""" + STREAM_ID_CONFLICT_POLICY_USE_EXISTING: _StreamIdConflictPolicy.ValueType # 2 + """The create gets the open Stream, which keeps its own Retention.""" + +class StreamIdConflictPolicy( + _StreamIdConflictPolicy, metaclass=_StreamIdConflictPolicyEnumTypeWrapper +): + """Experimental. Might change in the future. + What a create does when its Stream exists. Unspecified is FAIL, as for a Workflow start. + """ + +STREAM_ID_CONFLICT_POLICY_UNSPECIFIED: StreamIdConflictPolicy.ValueType # 0 +STREAM_ID_CONFLICT_POLICY_FAIL: StreamIdConflictPolicy.ValueType # 1 +"""The create fails with ALREADY_EXISTS.""" +STREAM_ID_CONFLICT_POLICY_USE_EXISTING: StreamIdConflictPolicy.ValueType # 2 +"""The create gets the open Stream, which keeps its own Retention.""" +global___StreamIdConflictPolicy = StreamIdConflictPolicy + +class _StreamErrorReason: + ValueType = typing.NewType("ValueType", builtins.int) + V: typing_extensions.TypeAlias = ValueType + +class _StreamErrorReasonEnumTypeWrapper( + google.protobuf.internal.enum_type_wrapper._EnumTypeWrapper[ + _StreamErrorReason.ValueType + ], + builtins.type, +): + DESCRIPTOR: google.protobuf.descriptor.EnumDescriptor + STREAM_ERROR_REASON_UNSPECIFIED: _StreamErrorReason.ValueType # 0 + STREAM_ERROR_REASON_NOT_FOUND: _StreamErrorReason.ValueType # 1 + """Nobody created the Stream. NOT_FOUND.""" + STREAM_ERROR_REASON_EXPIRED: _StreamErrorReason.ValueType # 2 + """The Stream's Retention or its last-write TTL passed. NOT_FOUND.""" + STREAM_ERROR_REASON_CLOSED: _StreamErrorReason.ValueType # 3 + """The Stream is closed. FAILED_PRECONDITION.""" + STREAM_ERROR_REASON_FENCED: _StreamErrorReason.ValueType # 4 + """The Producer's epoch is lower than the newest one. FAILED_PRECONDITION.""" + STREAM_ERROR_REASON_SEQUENCE_TOO_LOW: _StreamErrorReason.ValueType # 5 + """The batch has a lower `sequence` than the last one of its epoch. FAILED_PRECONDITION.""" + STREAM_ERROR_REASON_SEQ_DIVERGENT: _StreamErrorReason.ValueType # 6 + """The batch has the `sequence` of the last one of its epoch, and other content. + FAILED_PRECONDITION. + """ + STREAM_ERROR_REASON_NOT_OWNER: _StreamErrorReason.ValueType # 7 + """The caller may not create or close this Stream. FAILED_PRECONDITION.""" + STREAM_ERROR_REASON_FORMAT_TOO_NEW: _StreamErrorReason.ValueType # 8 + """A newer format version wrote the Stream. FAILED_PRECONDITION.""" + STREAM_ERROR_REASON_OFFSET_TOO_HIGH: _StreamErrorReason.ValueType # 9 + """The offset is past the newest offset, for example after a failover lost the newest Records. + A read from it works again once the Stream has Records up to it. OUT_OF_RANGE. + """ + STREAM_ERROR_REASON_TOO_LARGE: _StreamErrorReason.ValueType # 10 + """A Record or a batch is over the Store's limit. INVALID_ARGUMENT.""" + STREAM_ERROR_REASON_INVALID: _StreamErrorReason.ValueType # 11 + """The request is malformed. INVALID_ARGUMENT.""" + STREAM_ERROR_REASON_STORAGE_UNAVAILABLE: _StreamErrorReason.ValueType # 12 + """The Store failed or did not answer. A retry can succeed. UNAVAILABLE.""" + STREAM_ERROR_REASON_UNSUPPORTED: _StreamErrorReason.ValueType # 13 + """The Store lacks a capability that the call needs. UNIMPLEMENTED.""" + STREAM_ERROR_REASON_ALREADY_EXISTS: _StreamErrorReason.ValueType # 14 + """A create with the FAIL conflict policy is for a Stream that exists. ALREADY_EXISTS.""" + STREAM_ERROR_REASON_STORAGE_CORRUPT: _StreamErrorReason.ValueType # 15 + """The Store gave back data that isn't valid, such as a stored value with the wrong length. A + retry reads the same data. DATA_LOSS. + """ + +class StreamErrorReason( + _StreamErrorReason, metaclass=_StreamErrorReasonEnumTypeWrapper +): + """Experimental. Might change in the future. + Why a Stream call failed. Each reason has one gRPC status code, and several reasons share a + code. So code matches on the reason, and not on the status code. An SDK raises one error type + that carries the reason. + """ + +STREAM_ERROR_REASON_UNSPECIFIED: StreamErrorReason.ValueType # 0 +STREAM_ERROR_REASON_NOT_FOUND: StreamErrorReason.ValueType # 1 +"""Nobody created the Stream. NOT_FOUND.""" +STREAM_ERROR_REASON_EXPIRED: StreamErrorReason.ValueType # 2 +"""The Stream's Retention or its last-write TTL passed. NOT_FOUND.""" +STREAM_ERROR_REASON_CLOSED: StreamErrorReason.ValueType # 3 +"""The Stream is closed. FAILED_PRECONDITION.""" +STREAM_ERROR_REASON_FENCED: StreamErrorReason.ValueType # 4 +"""The Producer's epoch is lower than the newest one. FAILED_PRECONDITION.""" +STREAM_ERROR_REASON_SEQUENCE_TOO_LOW: StreamErrorReason.ValueType # 5 +"""The batch has a lower `sequence` than the last one of its epoch. FAILED_PRECONDITION.""" +STREAM_ERROR_REASON_SEQ_DIVERGENT: StreamErrorReason.ValueType # 6 +"""The batch has the `sequence` of the last one of its epoch, and other content. +FAILED_PRECONDITION. +""" +STREAM_ERROR_REASON_NOT_OWNER: StreamErrorReason.ValueType # 7 +"""The caller may not create or close this Stream. FAILED_PRECONDITION.""" +STREAM_ERROR_REASON_FORMAT_TOO_NEW: StreamErrorReason.ValueType # 8 +"""A newer format version wrote the Stream. FAILED_PRECONDITION.""" +STREAM_ERROR_REASON_OFFSET_TOO_HIGH: StreamErrorReason.ValueType # 9 +"""The offset is past the newest offset, for example after a failover lost the newest Records. +A read from it works again once the Stream has Records up to it. OUT_OF_RANGE. +""" +STREAM_ERROR_REASON_TOO_LARGE: StreamErrorReason.ValueType # 10 +"""A Record or a batch is over the Store's limit. INVALID_ARGUMENT.""" +STREAM_ERROR_REASON_INVALID: StreamErrorReason.ValueType # 11 +"""The request is malformed. INVALID_ARGUMENT.""" +STREAM_ERROR_REASON_STORAGE_UNAVAILABLE: StreamErrorReason.ValueType # 12 +"""The Store failed or did not answer. A retry can succeed. UNAVAILABLE.""" +STREAM_ERROR_REASON_UNSUPPORTED: StreamErrorReason.ValueType # 13 +"""The Store lacks a capability that the call needs. UNIMPLEMENTED.""" +STREAM_ERROR_REASON_ALREADY_EXISTS: StreamErrorReason.ValueType # 14 +"""A create with the FAIL conflict policy is for a Stream that exists. ALREADY_EXISTS.""" +STREAM_ERROR_REASON_STORAGE_CORRUPT: StreamErrorReason.ValueType # 15 +"""The Store gave back data that isn't valid, such as a stored value with the wrong length. A +retry reads the same data. DATA_LOSS. +""" +global___StreamErrorReason = StreamErrorReason diff --git a/temporalio/api/stream/__init__.py b/temporalio/api/stream/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/temporalio/api/stream/v1/__init__.py b/temporalio/api/stream/v1/__init__.py new file mode 100644 index 0000000000..73024c06b8 --- /dev/null +++ b/temporalio/api/stream/v1/__init__.py @@ -0,0 +1,29 @@ +from .message_pb2 import ( + ActivityEpoch, + ActivityProducer, + AppendRecord, + ClientProducer, + LocalActivityEpoch, + LocalActivityProducer, + Producer, + StreamErrorDetails, + StreamRecord, + StreamRef, + WorkflowProducer, + WorkflowTaskEpoch, +) + +__all__ = [ + "ActivityEpoch", + "ActivityProducer", + "AppendRecord", + "ClientProducer", + "LocalActivityEpoch", + "LocalActivityProducer", + "Producer", + "StreamErrorDetails", + "StreamRecord", + "StreamRef", + "WorkflowProducer", + "WorkflowTaskEpoch", +] diff --git a/temporalio/api/stream/v1/message_pb2.py b/temporalio/api/stream/v1/message_pb2.py new file mode 100644 index 0000000000..5bac69d283 --- /dev/null +++ b/temporalio/api/stream/v1/message_pb2.py @@ -0,0 +1,58 @@ +# -*- coding: utf-8 -*- +# Generated by the protocol buffer compiler. DO NOT EDIT! +# source: temporal/api/stream/v1/message.proto +"""Generated protocol buffer code.""" + +from google.protobuf import descriptor as _descriptor +from google.protobuf import descriptor_pool as _descriptor_pool +from google.protobuf import symbol_database as _symbol_database +from google.protobuf.internal import builder as _builder + +# @@protoc_insertion_point(imports) + +_sym_db = _symbol_database.Default() + + +from google.protobuf import timestamp_pb2 as google_dot_protobuf_dot_timestamp__pb2 + +from temporalio.api.common.v1 import ( + message_pb2 as temporal_dot_api_dot_common_dot_v1_dot_message__pb2, +) +from temporalio.api.enums.v1 import ( + stream_pb2 as temporal_dot_api_dot_enums_dot_v1_dot_stream__pb2, +) + +DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile( + b'\n$temporal/api/stream/v1/message.proto\x12\x16temporal.api.stream.v1\x1a\x1fgoogle/protobuf/timestamp.proto\x1a$temporal/api/common/v1/message.proto\x1a"temporal/api/enums/v1/stream.proto"&\n\tStreamRef\x12\n\n\x02id\x18\x01 \x01(\t\x12\r\n\x05store\x18\x03 \x01(\t"p\n\x11WorkflowTaskEpoch\x12\x18\n\x10started_event_id\x18\x01 \x01(\x03\x12\x0f\n\x07\x61ttempt\x18\x02 \x01(\x05\x12\x30\n\x0cstarted_time\x18\x03 \x01(\x0b\x32\x1a.google.protobuf.Timestamp"^\n\x12LocalActivityEpoch\x12\x37\n\x04task\x18\x01 \x01(\x0b\x32).temporal.api.stream.v1.WorkflowTaskEpoch\x12\x0f\n\x07\x61ttempt\x18\x02 \x01(\x05"\\\n\rActivityEpoch\x12:\n\x16\x61ttempt_scheduled_time\x18\x01 \x01(\x0b\x32\x1a.google.protobuf.Timestamp\x12\x0f\n\x07\x61ttempt\x18\x02 \x01(\x05"\\\n\x10WorkflowProducer\x12\x0e\n\x06run_id\x18\x01 \x01(\t\x12\x38\n\x05\x65poch\x18\x02 \x01(\x0b\x32).temporal.api.stream.v1.WorkflowTaskEpoch"b\n\x15LocalActivityProducer\x12\x0e\n\x06run_id\x18\x01 \x01(\t\x12\x39\n\x05\x65poch\x18\x02 \x01(\x0b\x32*.temporal.api.stream.v1.LocalActivityEpoch"X\n\x10\x41\x63tivityProducer\x12\x0e\n\x06run_id\x18\x01 \x01(\t\x12\x34\n\x05\x65poch\x18\x02 \x01(\x0b\x32%.temporal.api.stream.v1.ActivityEpoch"\x10\n\x0e\x43lientProducer"\xef\x02\n\x08Producer\x12\n\n\x02id\x18\x01 \x01(\t\x12<\n\x08workflow\x18\x02 \x01(\x0b\x32(.temporal.api.stream.v1.WorkflowProducerH\x00\x12G\n\x0elocal_activity\x18\x03 \x01(\x0b\x32-.temporal.api.stream.v1.LocalActivityProducerH\x00\x12\x45\n\x11workflow_activity\x18\x04 \x01(\x0b\x32(.temporal.api.stream.v1.ActivityProducerH\x00\x12G\n\x13standalone_activity\x18\x05 \x01(\x0b\x32(.temporal.api.stream.v1.ActivityProducerH\x00\x12\x38\n\x06\x63lient\x18\x06 \x01(\x0b\x32&.temporal.api.stream.v1.ClientProducerH\x00\x42\x06\n\x04kind"=\n\x0c\x41ppendRecord\x12-\n\x04\x62ody\x18\x01 \x01(\x0b\x32\x1f.temporal.api.common.v1.Payload"\x81\x01\n\x0cStreamRecord\x12\x32\n\x08producer\x18\x01 \x01(\x0b\x32 .temporal.api.stream.v1.Producer\x12-\n\x04\x62ody\x18\x02 \x01(\x0b\x32\x1f.temporal.api.common.v1.Payload\x12\x0e\n\x06offset\x18\x03 \x01(\x0c"N\n\x12StreamErrorDetails\x12\x38\n\x06reason\x18\x01 \x01(\x0e\x32(.temporal.api.enums.v1.StreamErrorReasonB\x89\x01\n\x19io.temporal.api.stream.v1B\x0cMessageProtoP\x01Z#go.temporal.io/api/stream/v1;stream\xaa\x02\x18Temporalio.Api.Stream.V1\xea\x02\x1bTemporalio::Api::Stream::V1b\x06proto3' +) + +_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, globals()) +_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, __name__, globals()) +if _descriptor._USE_C_DESCRIPTORS == False: + DESCRIPTOR._options = None + DESCRIPTOR._serialized_options = b"\n\031io.temporal.api.stream.v1B\014MessageProtoP\001Z#go.temporal.io/api/stream/v1;stream\252\002\030Temporalio.Api.Stream.V1\352\002\033Temporalio::Api::Stream::V1" + _STREAMREF._serialized_start = 171 + _STREAMREF._serialized_end = 209 + _WORKFLOWTASKEPOCH._serialized_start = 211 + _WORKFLOWTASKEPOCH._serialized_end = 323 + _LOCALACTIVITYEPOCH._serialized_start = 325 + _LOCALACTIVITYEPOCH._serialized_end = 419 + _ACTIVITYEPOCH._serialized_start = 421 + _ACTIVITYEPOCH._serialized_end = 513 + _WORKFLOWPRODUCER._serialized_start = 515 + _WORKFLOWPRODUCER._serialized_end = 607 + _LOCALACTIVITYPRODUCER._serialized_start = 609 + _LOCALACTIVITYPRODUCER._serialized_end = 707 + _ACTIVITYPRODUCER._serialized_start = 709 + _ACTIVITYPRODUCER._serialized_end = 797 + _CLIENTPRODUCER._serialized_start = 799 + _CLIENTPRODUCER._serialized_end = 815 + _PRODUCER._serialized_start = 818 + _PRODUCER._serialized_end = 1185 + _APPENDRECORD._serialized_start = 1187 + _APPENDRECORD._serialized_end = 1248 + _STREAMRECORD._serialized_start = 1251 + _STREAMRECORD._serialized_end = 1380 + _STREAMERRORDETAILS._serialized_start = 1382 + _STREAMERRORDETAILS._serialized_end = 1460 +# @@protoc_insertion_point(module_scope) diff --git a/temporalio/api/stream/v1/message_pb2.pyi b/temporalio/api/stream/v1/message_pb2.pyi new file mode 100644 index 0000000000..1d01036026 --- /dev/null +++ b/temporalio/api/stream/v1/message_pb2.pyi @@ -0,0 +1,465 @@ +""" +@generated by mypy-protobuf. Do not edit manually! +isort:skip_file +""" + +import builtins +import typing + +import google.protobuf.descriptor +import google.protobuf.message +import google.protobuf.timestamp_pb2 + +import temporalio.api.common.v1.message_pb2 +import temporalio.api.enums.v1.stream_pb2 + +DESCRIPTOR: google.protobuf.descriptor.FileDescriptor + +@typing.final +class StreamRef(google.protobuf.message.Message): + """Experimental. Might change in the future. + Identifies one Stream. The id is opaque to callers. A Stream has no incarnations. A create of a + closed or expired Stream fails, and it doesn't make a second Stream with the same id. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + ID_FIELD_NUMBER: builtins.int + STORE_FIELD_NUMBER: builtins.int + id: builtins.str + store: builtins.str + """The name of the Store that holds the Stream, as the Worker and client config register it, + so every process routes a ref it receives. With no name, the SDK picks the Store from its + config. Only an SDK sets and reads it, since only the SDK knows the name that its config + gives a Store. A request carries the whole ref, and the server and a Store ignore this + field. + """ + def __init__( + self, + *, + id: builtins.str = ..., + store: builtins.str = ..., + ) -> None: ... + def ClearField( + self, field_name: typing.Literal["id", b"id", "store", b"store"] + ) -> None: ... + +global___StreamRef = StreamRef + +@typing.final +class WorkflowTaskEpoch(google.protobuf.message.Message): + """Experimental. Might change in the future. + One Workflow Task attempt of a run. It orders by started_event_id, then attempt, then + started_time. An epoch orders the attempts of one Producer only, and a Producer belongs to one + run. So a reset run and the next run of a chain have other Producers, and their epochs start + again. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + STARTED_EVENT_ID_FIELD_NUMBER: builtins.int + ATTEMPT_FIELD_NUMBER: builtins.int + STARTED_TIME_FIELD_NUMBER: builtins.int + started_event_id: builtins.int + """Required.""" + attempt: builtins.int + """Required. 1 or more.""" + @property + def started_time(self) -> google.protobuf.timestamp_pb2.Timestamp: + """Tells a reused started event id apart, for example after a dropped speculative task. Core + leaves it empty when the server gave the task no started time. + """ + + def __init__( + self, + *, + started_event_id: builtins.int = ..., + attempt: builtins.int = ..., + started_time: google.protobuf.timestamp_pb2.Timestamp | None = ..., + ) -> None: ... + def HasField( + self, field_name: typing.Literal["started_time", b"started_time"] + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing.Literal[ + "attempt", + b"attempt", + "started_event_id", + b"started_event_id", + "started_time", + b"started_time", + ], + ) -> None: ... + +global___WorkflowTaskEpoch = WorkflowTaskEpoch + +@typing.final +class LocalActivityEpoch(google.protobuf.message.Message): + """Experimental. Might change in the future. + The attempt of a Local Activity. It orders by task, then attempt, so the same call in a later + Workflow Task is a higher epoch. The Producer carries the run. Records of two runs have no + order by epoch, only their order in the Stream. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + TASK_FIELD_NUMBER: builtins.int + ATTEMPT_FIELD_NUMBER: builtins.int + attempt: builtins.int + """Required. The Local Activity's own attempt in that Workflow Task, 1 or more.""" + @property + def task(self) -> global___WorkflowTaskEpoch: + """Required. The Workflow Task attempt that dispatched the Local Activity.""" + + def __init__( + self, + *, + task: global___WorkflowTaskEpoch | None = ..., + attempt: builtins.int = ..., + ) -> None: ... + def HasField( + self, field_name: typing.Literal["task", b"task"] + ) -> builtins.bool: ... + def ClearField( + self, field_name: typing.Literal["attempt", b"attempt", "task", b"task"] + ) -> None: ... + +global___LocalActivityEpoch = LocalActivityEpoch + +@typing.final +class ActivityEpoch(google.protobuf.message.Message): + """Experimental. Might change in the future. + The attempt of an Activity. It orders by attempt_scheduled_time, then attempt. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + ATTEMPT_SCHEDULED_TIME_FIELD_NUMBER: builtins.int + ATTEMPT_FIELD_NUMBER: builtins.int + attempt: builtins.int + """Required. 1 or more.""" + @property + def attempt_scheduled_time(self) -> google.protobuf.timestamp_pb2.Timestamp: + """Required. When the server scheduled this attempt. An Activity reset starts the attempt count + again, so only this time orders the attempts across a reset. + """ + + def __init__( + self, + *, + attempt_scheduled_time: google.protobuf.timestamp_pb2.Timestamp | None = ..., + attempt: builtins.int = ..., + ) -> None: ... + def HasField( + self, + field_name: typing.Literal["attempt_scheduled_time", b"attempt_scheduled_time"], + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing.Literal[ + "attempt", b"attempt", "attempt_scheduled_time", b"attempt_scheduled_time" + ], + ) -> None: ... + +global___ActivityEpoch = ActivityEpoch + +@typing.final +class WorkflowProducer(google.protobuf.message.Message): + """Experimental. Might change in the future. + A Workflow run Producer. On the server's Streams, a Workflow writes as this Producer in its + Workflow Task. On a Store, Workflow code writes through a Local Activity, so its writes carry a + Local Activity Producer. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + RUN_ID_FIELD_NUMBER: builtins.int + EPOCH_FIELD_NUMBER: builtins.int + run_id: builtins.str + """Required. The run id of the Workflow run, not of its chain. Each run is its own Producer.""" + @property + def epoch(self) -> global___WorkflowTaskEpoch: + """The Workflow Task of the write.""" + + def __init__( + self, + *, + run_id: builtins.str = ..., + epoch: global___WorkflowTaskEpoch | None = ..., + ) -> None: ... + def HasField( + self, field_name: typing.Literal["epoch", b"epoch"] + ) -> builtins.bool: ... + def ClearField( + self, field_name: typing.Literal["epoch", b"epoch", "run_id", b"run_id"] + ) -> None: ... + +global___WorkflowProducer = WorkflowProducer + +@typing.final +class LocalActivityProducer(google.protobuf.message.Message): + """Experimental. Might change in the future. + A Local Activity Producer, one for each call of a run. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + RUN_ID_FIELD_NUMBER: builtins.int + EPOCH_FIELD_NUMBER: builtins.int + run_id: builtins.str + """Required. The run id of the Workflow run.""" + @property + def epoch(self) -> global___LocalActivityEpoch: + """Required.""" + + def __init__( + self, + *, + run_id: builtins.str = ..., + epoch: global___LocalActivityEpoch | None = ..., + ) -> None: ... + def HasField( + self, field_name: typing.Literal["epoch", b"epoch"] + ) -> builtins.bool: ... + def ClearField( + self, field_name: typing.Literal["epoch", b"epoch", "run_id", b"run_id"] + ) -> None: ... + +global___LocalActivityProducer = LocalActivityProducer + +@typing.final +class ActivityProducer(google.protobuf.message.Message): + """Experimental. Might change in the future. + An Activity Producer, for an Activity of a Workflow or a Standalone Activity. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + RUN_ID_FIELD_NUMBER: builtins.int + EPOCH_FIELD_NUMBER: builtins.int + run_id: builtins.str + """The run id of the Workflow run that scheduled the Activity. Empty for a Standalone + Activity, whose own run id is in the Producer id. + """ + @property + def epoch(self) -> global___ActivityEpoch: + """Required.""" + + def __init__( + self, + *, + run_id: builtins.str = ..., + epoch: global___ActivityEpoch | None = ..., + ) -> None: ... + def HasField( + self, field_name: typing.Literal["epoch", b"epoch"] + ) -> builtins.bool: ... + def ClearField( + self, field_name: typing.Literal["epoch", b"epoch", "run_id", b"run_id"] + ) -> None: ... + +global___ActivityProducer = ActivityProducer + +@typing.final +class ClientProducer(google.protobuf.message.Message): + """Experimental. Might change in the future. + A client Producer. A client has no run and no epoch. So nothing fences a client's writes, and a + reader never hides them. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + def __init__( + self, + ) -> None: ... + +global___ClientProducer = ClientProducer + +@typing.final +class Producer(google.protobuf.message.Message): + """Experimental. Might change in the future. + The code that wrote a batch, which is one unit of work. Its id stays the same across attempts, + and an epoch is one attempt. The kind's case orders its epochs. After a write of a higher epoch, + a write of a lower epoch of the same Producer fails with FENCED, and a reader may hide the + Records of the lower epoch. The rule is the same for every kind that has an epoch. A client has + none. The epochs of two Producers never compare. A new run has new Producers, so no write + fences or supersedes the Records of an earlier run, such as the base run of a reset. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + ID_FIELD_NUMBER: builtins.int + WORKFLOW_FIELD_NUMBER: builtins.int + LOCAL_ACTIVITY_FIELD_NUMBER: builtins.int + WORKFLOW_ACTIVITY_FIELD_NUMBER: builtins.int + STANDALONE_ACTIVITY_FIELD_NUMBER: builtins.int + CLIENT_FIELD_NUMBER: builtins.int + id: builtins.str + """Required. Stable across attempts, such as `wf:` for the Workflow Producer of a run. + It is unique in one Stream. The id of a run-based kind has its run id, so it stays unique + when Producers of more than one namespace write to one Stream. + """ + @property + def workflow(self) -> global___WorkflowProducer: ... + @property + def local_activity(self) -> global___LocalActivityProducer: ... + @property + def workflow_activity(self) -> global___ActivityProducer: + """An Activity that a Workflow scheduled.""" + + @property + def standalone_activity(self) -> global___ActivityProducer: + """An Activity with no Workflow. Both Activity cases order their epochs the same way. The + case tells which identity the Producer has, since only the first has a Workflow run. + """ + + @property + def client(self) -> global___ClientProducer: ... + def __init__( + self, + *, + id: builtins.str = ..., + workflow: global___WorkflowProducer | None = ..., + local_activity: global___LocalActivityProducer | None = ..., + workflow_activity: global___ActivityProducer | None = ..., + standalone_activity: global___ActivityProducer | None = ..., + client: global___ClientProducer | None = ..., + ) -> None: ... + def HasField( + self, + field_name: typing.Literal[ + "client", + b"client", + "kind", + b"kind", + "local_activity", + b"local_activity", + "standalone_activity", + b"standalone_activity", + "workflow", + b"workflow", + "workflow_activity", + b"workflow_activity", + ], + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing.Literal[ + "client", + b"client", + "id", + b"id", + "kind", + b"kind", + "local_activity", + b"local_activity", + "standalone_activity", + b"standalone_activity", + "workflow", + b"workflow", + "workflow_activity", + b"workflow_activity", + ], + ) -> None: ... + def WhichOneof( + self, oneof_group: typing.Literal["kind", b"kind"] + ) -> ( + typing.Literal[ + "workflow", + "local_activity", + "workflow_activity", + "standalone_activity", + "client", + ] + | None + ): ... + +global___Producer = Producer + +@typing.final +class AppendRecord(google.protobuf.message.Message): + """Experimental. Might change in the future. + One Record of a batch to append. The Store adds the batch's Producer and an offset. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + BODY_FIELD_NUMBER: builtins.int + @property + def body(self) -> temporalio.api.common.v1.message_pb2.Payload: + """Required.""" + + def __init__( + self, + *, + body: temporalio.api.common.v1.message_pb2.Payload | None = ..., + ) -> None: ... + def HasField( + self, field_name: typing.Literal["body", b"body"] + ) -> builtins.bool: ... + def ClearField(self, field_name: typing.Literal["body", b"body"]) -> None: ... + +global___AppendRecord = AppendRecord + +@typing.final +class StreamRecord(google.protobuf.message.Message): + """Experimental. Might change in the future. + One Record that a Producer appended, as a read gives it. A read gives only Records. A close + stores no Record, so a read tells of it with the done flag of its response. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + PRODUCER_FIELD_NUMBER: builtins.int + BODY_FIELD_NUMBER: builtins.int + OFFSET_FIELD_NUMBER: builtins.int + offset: builtins.bytes + """Required. The Record's offset. An offset is ordered and unique in one Stream, and it is + sparse. Offsets have gaps, so they don't count Records. Two offsets of one Stream compare + bytewise, as unsigned bytes, in the order of their Records. So a reader, a Store and the + server order them without decoding them. An offset is never empty. A read gives its + Stream, so the read checks the offset against that Stream. + """ + @property + def producer(self) -> global___Producer: ... + @property + def body(self) -> temporalio.api.common.v1.message_pb2.Payload: ... + def __init__( + self, + *, + producer: global___Producer | None = ..., + body: temporalio.api.common.v1.message_pb2.Payload | None = ..., + offset: builtins.bytes = ..., + ) -> None: ... + def HasField( + self, field_name: typing.Literal["body", b"body", "producer", b"producer"] + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing.Literal[ + "body", b"body", "offset", b"offset", "producer", b"producer" + ], + ) -> None: ... + +global___StreamRecord = StreamRecord + +@typing.final +class StreamErrorDetails(google.protobuf.message.Message): + """Experimental. Might change in the future. + The details of a failed Stream call, in the gRPC status details and in the details of the + application failure of a Nexus operation. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + REASON_FIELD_NUMBER: builtins.int + reason: temporalio.api.enums.v1.stream_pb2.StreamErrorReason.ValueType + def __init__( + self, + *, + reason: temporalio.api.enums.v1.stream_pb2.StreamErrorReason.ValueType = ..., + ) -> None: ... + def ClearField(self, field_name: typing.Literal["reason", b"reason"]) -> None: ... + +global___StreamErrorDetails = StreamErrorDetails diff --git a/temporalio/api/streamservice/__init__.py b/temporalio/api/streamservice/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/temporalio/api/streamservice/v1/__init__.py b/temporalio/api/streamservice/v1/__init__.py new file mode 100644 index 0000000000..40801cf9a5 --- /dev/null +++ b/temporalio/api/streamservice/v1/__init__.py @@ -0,0 +1,25 @@ +from .request_response_pb2 import ( + AppendRecordsRequest, + AppendRecordsResponse, + CloseStreamRequest, + CloseStreamResponse, + CreateStreamRequest, + CreateStreamResponse, + DescribeStreamRequest, + DescribeStreamResponse, + ReadRecordsRequest, + ReadRecordsResponse, +) + +__all__ = [ + "AppendRecordsRequest", + "AppendRecordsResponse", + "CloseStreamRequest", + "CloseStreamResponse", + "CreateStreamRequest", + "CreateStreamResponse", + "DescribeStreamRequest", + "DescribeStreamResponse", + "ReadRecordsRequest", + "ReadRecordsResponse", +] diff --git a/temporalio/api/streamservice/v1/request_response_pb2.py b/temporalio/api/streamservice/v1/request_response_pb2.py new file mode 100644 index 0000000000..161aa0c07b --- /dev/null +++ b/temporalio/api/streamservice/v1/request_response_pb2.py @@ -0,0 +1,54 @@ +# -*- coding: utf-8 -*- +# Generated by the protocol buffer compiler. DO NOT EDIT! +# source: temporal/api/streamservice/v1/request_response.proto +"""Generated protocol buffer code.""" + +from google.protobuf import descriptor as _descriptor +from google.protobuf import descriptor_pool as _descriptor_pool +from google.protobuf import symbol_database as _symbol_database +from google.protobuf.internal import builder as _builder + +# @@protoc_insertion_point(imports) + +_sym_db = _symbol_database.Default() + + +from google.protobuf import duration_pb2 as google_dot_protobuf_dot_duration__pb2 + +from temporalio.api.enums.v1 import ( + stream_pb2 as temporal_dot_api_dot_enums_dot_v1_dot_stream__pb2, +) +from temporalio.api.stream.v1 import ( + message_pb2 as temporal_dot_api_dot_stream_dot_v1_dot_message__pb2, +) + +DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile( + b'\n4temporal/api/streamservice/v1/request_response.proto\x12\x1dtemporal.api.streamservice.v1\x1a\x1egoogle/protobuf/duration.proto\x1a"temporal/api/enums/v1/stream.proto\x1a$temporal/api/stream/v1/message.proto"\xbe\x01\n\x13\x43reateStreamRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12\x35\n\nstream_ref\x18\x02 \x01(\x0b\x32!.temporal.api.stream.v1.StreamRef\x12I\n\x12id_conflict_policy\x18\x03 \x01(\x0e\x32-.temporal.api.enums.v1.StreamIdConflictPolicy\x12\x12\n\nrequest_id\x18\x04 \x01(\t"W\n\x14\x43reateStreamResponse\x12.\n\x03ref\x18\x01 \x01(\x0b\x32!.temporal.api.stream.v1.StreamRef\x12\x0f\n\x07\x63reated\x18\x02 \x01(\x08"\xd6\x01\n\x14\x41ppendRecordsRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12.\n\x03ref\x18\x02 \x01(\x0b\x32!.temporal.api.stream.v1.StreamRef\x12\x32\n\x08producer\x18\x03 \x01(\x0b\x32 .temporal.api.stream.v1.Producer\x12\x10\n\x08sequence\x18\x04 \x01(\x03\x12\x35\n\x07records\x18\x05 \x03(\x0b\x32$.temporal.api.stream.v1.AppendRecord"B\n\x15\x41ppendRecordsResponse\x12\x14\n\x0c\x66irst_offset\x18\x01 \x01(\x0c\x12\x13\n\x0blast_offset\x18\x02 \x01(\x0c"\x8b\x01\n\x12\x43loseStreamRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12.\n\x03ref\x18\x02 \x01(\x0b\x32!.temporal.api.stream.v1.StreamRef\x12\x32\n\x08producer\x18\x03 \x01(\x0b\x32 .temporal.api.stream.v1.Producer"\x15\n\x13\x43loseStreamResponse"\xbd\x01\n\x12ReadRecordsRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12.\n\x03ref\x18\x02 \x01(\x0b\x32!.temporal.api.stream.v1.StreamRef\x12\x13\n\x0b\x66rom_offset\x18\x03 \x01(\x0c\x12\x13\n\x0bmax_records\x18\x04 \x01(\x05\x12\'\n\x04wait\x18\x05 \x01(\x0b\x32\x19.google.protobuf.Duration\x12\x11\n\tmax_bytes\x18\x06 \x01(\x03"o\n\x13ReadRecordsResponse\x12\x35\n\x07records\x18\x01 \x03(\x0b\x32$.temporal.api.stream.v1.StreamRecord\x12\x13\n\x0blast_offset\x18\x02 \x01(\x0c\x12\x0c\n\x04\x64one\x18\x03 \x01(\x08"Z\n\x15\x44\x65scribeStreamRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12.\n\x03ref\x18\x02 \x01(\x0b\x32!.temporal.api.stream.v1.StreamRef"b\n\x16\x44\x65scribeStreamResponse\x12\x33\n\x06status\x18\x01 \x01(\x0e\x32#.temporal.api.enums.v1.StreamStatus\x12\x13\n\x0blast_offset\x18\x02 \x01(\x0c\x42\xb4\x01\n io.temporal.api.streamservice.v1B\x14RequestResponseProtoP\x01Z1go.temporal.io/api/streamservice/v1;streamservice\xaa\x02\x1fTemporalio.Api.StreamService.V1\xea\x02"Temporalio::Api::StreamService::V1b\x06proto3' +) + +_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, globals()) +_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, __name__, globals()) +if _descriptor._USE_C_DESCRIPTORS == False: + DESCRIPTOR._options = None + DESCRIPTOR._serialized_options = b'\n io.temporal.api.streamservice.v1B\024RequestResponseProtoP\001Z1go.temporal.io/api/streamservice/v1;streamservice\252\002\037Temporalio.Api.StreamService.V1\352\002"Temporalio::Api::StreamService::V1' + _CREATESTREAMREQUEST._serialized_start = 194 + _CREATESTREAMREQUEST._serialized_end = 384 + _CREATESTREAMRESPONSE._serialized_start = 386 + _CREATESTREAMRESPONSE._serialized_end = 473 + _APPENDRECORDSREQUEST._serialized_start = 476 + _APPENDRECORDSREQUEST._serialized_end = 690 + _APPENDRECORDSRESPONSE._serialized_start = 692 + _APPENDRECORDSRESPONSE._serialized_end = 758 + _CLOSESTREAMREQUEST._serialized_start = 761 + _CLOSESTREAMREQUEST._serialized_end = 900 + _CLOSESTREAMRESPONSE._serialized_start = 902 + _CLOSESTREAMRESPONSE._serialized_end = 923 + _READRECORDSREQUEST._serialized_start = 926 + _READRECORDSREQUEST._serialized_end = 1115 + _READRECORDSRESPONSE._serialized_start = 1117 + _READRECORDSRESPONSE._serialized_end = 1228 + _DESCRIBESTREAMREQUEST._serialized_start = 1230 + _DESCRIBESTREAMREQUEST._serialized_end = 1320 + _DESCRIBESTREAMRESPONSE._serialized_start = 1322 + _DESCRIBESTREAMRESPONSE._serialized_end = 1420 +# @@protoc_insertion_point(module_scope) diff --git a/temporalio/api/streamservice/v1/request_response_pb2.pyi b/temporalio/api/streamservice/v1/request_response_pb2.pyi new file mode 100644 index 0000000000..5ae5b08e50 --- /dev/null +++ b/temporalio/api/streamservice/v1/request_response_pb2.pyi @@ -0,0 +1,446 @@ +""" +@generated by mypy-protobuf. Do not edit manually! +isort:skip_file +""" + +import builtins +import collections.abc +import typing + +import google.protobuf.descriptor +import google.protobuf.duration_pb2 +import google.protobuf.internal.containers +import google.protobuf.message + +import temporalio.api.enums.v1.stream_pb2 +import temporalio.api.stream.v1.message_pb2 + +DESCRIPTOR: google.protobuf.descriptor.FileDescriptor + +@typing.final +class CreateStreamRequest(google.protobuf.message.Message): + """Experimental. Might change in the future. + Creates the Stream. When the Stream exists, FAIL fails with ALREADY_EXISTS, and USE_EXISTING + gets it with created false. A retry of the create that made the Stream, with the same + request_id, gets created true. Each policy fails with CLOSED on a closed Stream and EXPIRED on + an expired one. A create takes no Retention. The Store config holds it, as a namespace holds + the Retention of its Workflows. + (-- api-linter: core::0133=disabled + aip.dev/not-precedent: The request identifies a Stream by namespace and ref. --) + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + NAMESPACE_FIELD_NUMBER: builtins.int + STREAM_REF_FIELD_NUMBER: builtins.int + ID_CONFLICT_POLICY_FIELD_NUMBER: builtins.int + REQUEST_ID_FIELD_NUMBER: builtins.int + namespace: builtins.str + """The caller's namespace. Only read by the server. The SDK sets it, and the SDK Stores don't + read it. + """ + id_conflict_policy: ( + temporalio.api.enums.v1.stream_pb2.StreamIdConflictPolicy.ValueType + ) + """What the create does when the Stream exists. Unspecified is FAIL.""" + request_id: builtins.str + """Unique for each create call, and the same on its retries. The Store keeps the request_id of + the create that made the Stream, so a retry after a lost answer doesn't get ALREADY_EXISTS. + """ + @property + def stream_ref(self) -> temporalio.api.stream.v1.message_pb2.StreamRef: + """Required. The ref of the Stream to create. Its id is the Stream id, and is never empty.""" + + def __init__( + self, + *, + namespace: builtins.str = ..., + stream_ref: temporalio.api.stream.v1.message_pb2.StreamRef | None = ..., + id_conflict_policy: temporalio.api.enums.v1.stream_pb2.StreamIdConflictPolicy.ValueType = ..., + request_id: builtins.str = ..., + ) -> None: ... + def HasField( + self, field_name: typing.Literal["stream_ref", b"stream_ref"] + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing.Literal[ + "id_conflict_policy", + b"id_conflict_policy", + "namespace", + b"namespace", + "request_id", + b"request_id", + "stream_ref", + b"stream_ref", + ], + ) -> None: ... + +global___CreateStreamRequest = CreateStreamRequest + +@typing.final +class CreateStreamResponse(google.protobuf.message.Message): + """Experimental. Might change in the future. + The Stream that a create made or got. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + REF_FIELD_NUMBER: builtins.int + CREATED_FIELD_NUMBER: builtins.int + created: builtins.bool + """False when USE_EXISTING got a Stream that existed.""" + @property + def ref(self) -> temporalio.api.stream.v1.message_pb2.StreamRef: + """The ref of the Stream, which the caller passes on as data. The SDK sets the name of the + Store that it picked, also when the request left the name empty. The server gives the ref + back as the request had it, since it doesn't know that name. + """ + + def __init__( + self, + *, + ref: temporalio.api.stream.v1.message_pb2.StreamRef | None = ..., + created: builtins.bool = ..., + ) -> None: ... + def HasField(self, field_name: typing.Literal["ref", b"ref"]) -> builtins.bool: ... + def ClearField( + self, field_name: typing.Literal["created", b"created", "ref", b"ref"] + ) -> None: ... + +global___CreateStreamResponse = CreateStreamResponse + +@typing.final +class AppendRecordsRequest(google.protobuf.message.Message): + """Experimental. Might change in the future. + Appends one batch of Records as one Producer. The Store writes a resent batch, with the same + Producer, epoch and sequence, only once. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + NAMESPACE_FIELD_NUMBER: builtins.int + REF_FIELD_NUMBER: builtins.int + PRODUCER_FIELD_NUMBER: builtins.int + SEQUENCE_FIELD_NUMBER: builtins.int + RECORDS_FIELD_NUMBER: builtins.int + namespace: builtins.str + """The caller's namespace. Only read by the server. The SDK sets it, and the SDK Stores don't + read it. + """ + sequence: builtins.int + """The batch number in this epoch. With the Producer and epoch it is the dedupe key. Zero takes + the next number of the epoch. Such a batch has no dedupe, so a retry writes it again. + """ + @property + def ref(self) -> temporalio.api.stream.v1.message_pb2.StreamRef: + """Required.""" + + @property + def producer(self) -> temporalio.api.stream.v1.message_pb2.Producer: + """Required. The Producer of the batch. An append and a close of one Producer carry the same + Producer, so the Store orders each write the same way. + """ + + @property + def records( + self, + ) -> google.protobuf.internal.containers.RepeatedCompositeFieldContainer[ + temporalio.api.stream.v1.message_pb2.AppendRecord + ]: ... + def __init__( + self, + *, + namespace: builtins.str = ..., + ref: temporalio.api.stream.v1.message_pb2.StreamRef | None = ..., + producer: temporalio.api.stream.v1.message_pb2.Producer | None = ..., + sequence: builtins.int = ..., + records: collections.abc.Iterable[ + temporalio.api.stream.v1.message_pb2.AppendRecord + ] + | None = ..., + ) -> None: ... + def HasField( + self, field_name: typing.Literal["producer", b"producer", "ref", b"ref"] + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing.Literal[ + "namespace", + b"namespace", + "producer", + b"producer", + "records", + b"records", + "ref", + b"ref", + "sequence", + b"sequence", + ], + ) -> None: ... + +global___AppendRecordsRequest = AppendRecordsRequest + +@typing.final +class AppendRecordsResponse(google.protobuf.message.Message): + """Experimental. Might change in the future. + Where the batch's Records are in the Stream. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + FIRST_OFFSET_FIELD_NUMBER: builtins.int + LAST_OFFSET_FIELD_NUMBER: builtins.int + first_offset: builtins.bytes + last_offset: builtins.bytes + def __init__( + self, + *, + first_offset: builtins.bytes = ..., + last_offset: builtins.bytes = ..., + ) -> None: ... + def ClearField( + self, + field_name: typing.Literal[ + "first_offset", b"first_offset", "last_offset", b"last_offset" + ], + ) -> None: ... + +global___AppendRecordsResponse = AppendRecordsResponse + +@typing.final +class CloseStreamRequest(google.protobuf.message.Message): + """Experimental. Might change in the future. + Closes a Stream. Readers drain what is left, then end, and later appends fail with CLOSED. A + close writes no Record and carries no value, so code that has a last value publishes it first. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + NAMESPACE_FIELD_NUMBER: builtins.int + REF_FIELD_NUMBER: builtins.int + PRODUCER_FIELD_NUMBER: builtins.int + namespace: builtins.str + """The caller's namespace. Only read by the server. The SDK sets it, and the SDK Stores don't + read it. + """ + @property + def ref(self) -> temporalio.api.stream.v1.message_pb2.StreamRef: + """Required.""" + + @property + def producer(self) -> temporalio.api.stream.v1.message_pb2.Producer: + """Required. The Producer that closes the Stream.""" + + def __init__( + self, + *, + namespace: builtins.str = ..., + ref: temporalio.api.stream.v1.message_pb2.StreamRef | None = ..., + producer: temporalio.api.stream.v1.message_pb2.Producer | None = ..., + ) -> None: ... + def HasField( + self, field_name: typing.Literal["producer", b"producer", "ref", b"ref"] + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing.Literal[ + "namespace", b"namespace", "producer", b"producer", "ref", b"ref" + ], + ) -> None: ... + +global___CloseStreamRequest = CloseStreamRequest + +@typing.final +class CloseStreamResponse(google.protobuf.message.Message): + """Experimental. Might change in the future. + A close answers when the Store has the close. It gives nothing back. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + def __init__( + self, + ) -> None: ... + +global___CloseStreamResponse = CloseStreamResponse + +@typing.final +class ReadRecordsRequest(google.protobuf.message.Message): + """Experimental. Might change in the future. + Reads one page of Records after an offset. At the end of the Stream, the read waits for a new + Record, so readers long poll. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + NAMESPACE_FIELD_NUMBER: builtins.int + REF_FIELD_NUMBER: builtins.int + FROM_OFFSET_FIELD_NUMBER: builtins.int + MAX_RECORDS_FIELD_NUMBER: builtins.int + WAIT_FIELD_NUMBER: builtins.int + MAX_BYTES_FIELD_NUMBER: builtins.int + namespace: builtins.str + """The caller's namespace. Only read by the server. The SDK sets it, and the SDK Stores don't + read it. + """ + from_offset: builtins.bytes + """Read after this offset. It fails with OFFSET_TOO_HIGH if it is past the newest offset. + Leave empty to read from the beginning. To read only new Records, a caller gets last_offset + from a describe and reads after it. + (-- api-linter: core::0140::prepositions=disabled + aip.dev/not-precedent: "from" marks the start of a range of offsets. --) + """ + max_records: builtins.int + """The most Records in one page. Zero is 1000.""" + max_bytes: builtins.int + """The most Record bytes in one page, so a page fits the caller's limit, such as a History + marker. A larger Record still comes alone, so the read always moves on. Zero is no cap. + """ + @property + def ref(self) -> temporalio.api.stream.v1.message_pb2.StreamRef: + """Required.""" + + @property + def wait(self) -> google.protobuf.duration_pb2.Duration: + """How long the read waits for a new Record before it answers with an empty page. Empty answers + at once, and a wait longer than 60 s waits 60 s. + """ + + def __init__( + self, + *, + namespace: builtins.str = ..., + ref: temporalio.api.stream.v1.message_pb2.StreamRef | None = ..., + from_offset: builtins.bytes = ..., + max_records: builtins.int = ..., + wait: google.protobuf.duration_pb2.Duration | None = ..., + max_bytes: builtins.int = ..., + ) -> None: ... + def HasField( + self, field_name: typing.Literal["ref", b"ref", "wait", b"wait"] + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing.Literal[ + "from_offset", + b"from_offset", + "max_bytes", + b"max_bytes", + "max_records", + b"max_records", + "namespace", + b"namespace", + "ref", + b"ref", + "wait", + b"wait", + ], + ) -> None: ... + +global___ReadRecordsRequest = ReadRecordsRequest + +@typing.final +class ReadRecordsResponse(google.protobuf.message.Message): + """Experimental. Might change in the future. + One page of a read. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + RECORDS_FIELD_NUMBER: builtins.int + LAST_OFFSET_FIELD_NUMBER: builtins.int + DONE_FIELD_NUMBER: builtins.int + last_offset: builtins.bytes + """The last offset that this read covered, which is the offset of the newest Record of the + page. The next read gives it as from_offset and starts after it. An empty page gives the + request's from_offset, and that is empty when the request had none. So it is always + known, and an empty value means that the next read starts as this one did. + """ + done: builtins.bool + """True when the Stream is closed and this read gave its last Record. It is the only sign of a + close in a read, since a close stores no Record. A read of a closed Stream with no Record + left answers at once with done. + """ + @property + def records( + self, + ) -> google.protobuf.internal.containers.RepeatedCompositeFieldContainer[ + temporalio.api.stream.v1.message_pb2.StreamRecord + ]: ... + def __init__( + self, + *, + records: collections.abc.Iterable[ + temporalio.api.stream.v1.message_pb2.StreamRecord + ] + | None = ..., + last_offset: builtins.bytes = ..., + done: builtins.bool = ..., + ) -> None: ... + def ClearField( + self, + field_name: typing.Literal[ + "done", b"done", "last_offset", b"last_offset", "records", b"records" + ], + ) -> None: ... + +global___ReadRecordsResponse = ReadRecordsResponse + +@typing.final +class DescribeStreamRequest(google.protobuf.message.Message): + """Experimental. Might change in the future. + Gets the status and the newest offset of a Stream. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + NAMESPACE_FIELD_NUMBER: builtins.int + REF_FIELD_NUMBER: builtins.int + namespace: builtins.str + """The caller's namespace. Only read by the server. The SDK sets it, and the SDK Stores don't + read it. + """ + @property + def ref(self) -> temporalio.api.stream.v1.message_pb2.StreamRef: + """Required.""" + + def __init__( + self, + *, + namespace: builtins.str = ..., + ref: temporalio.api.stream.v1.message_pb2.StreamRef | None = ..., + ) -> None: ... + def HasField(self, field_name: typing.Literal["ref", b"ref"]) -> builtins.bool: ... + def ClearField( + self, field_name: typing.Literal["namespace", b"namespace", "ref", b"ref"] + ) -> None: ... + +global___DescribeStreamRequest = DescribeStreamRequest + +@typing.final +class DescribeStreamResponse(google.protobuf.message.Message): + """Experimental. Might change in the future. + The state of a Stream. It has no Retention, since the Retention is a setting of the Store and + not of one Stream. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + STATUS_FIELD_NUMBER: builtins.int + LAST_OFFSET_FIELD_NUMBER: builtins.int + status: temporalio.api.enums.v1.stream_pb2.StreamStatus.ValueType + last_offset: builtins.bytes + """The newest offset, of a Record. Empty when the Stream has none.""" + def __init__( + self, + *, + status: temporalio.api.enums.v1.stream_pb2.StreamStatus.ValueType = ..., + last_offset: builtins.bytes = ..., + ) -> None: ... + def ClearField( + self, + field_name: typing.Literal["last_offset", b"last_offset", "status", b"status"], + ) -> None: ... + +global___DescribeStreamResponse = DescribeStreamResponse diff --git a/temporalio/api/streamservice/v1/request_response_pb2_grpc.py b/temporalio/api/streamservice/v1/request_response_pb2_grpc.py new file mode 100644 index 0000000000..bf947056a2 --- /dev/null +++ b/temporalio/api/streamservice/v1/request_response_pb2_grpc.py @@ -0,0 +1,4 @@ +# Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT! +"""Client and server classes corresponding to protobuf-defined services.""" + +import grpc diff --git a/temporalio/api/streamservice/v1/request_response_pb2_grpc.pyi b/temporalio/api/streamservice/v1/request_response_pb2_grpc.pyi new file mode 100644 index 0000000000..de236ac811 --- /dev/null +++ b/temporalio/api/streamservice/v1/request_response_pb2_grpc.pyi @@ -0,0 +1,21 @@ +""" +@generated by mypy-protobuf. Do not edit manually! +isort:skip_file +""" + +import abc +import collections.abc +import typing + +import grpc +import grpc.aio + +_T = typing.TypeVar("_T") + +class _MaybeAsyncIterator( + collections.abc.AsyncIterator[_T], + collections.abc.Iterator[_T], + metaclass=abc.ABCMeta, +): ... +class _ServicerContext(grpc.ServicerContext, grpc.aio.ServicerContext): # type: ignore[misc, type-arg] + ... diff --git a/temporalio/bridge/_visitor.py b/temporalio/bridge/_visitor.py index ead09a986f..8832b56bc4 100644 --- a/temporalio/bridge/_visitor.py +++ b/temporalio/bridge/_visitor.py @@ -639,6 +639,70 @@ async def _visit_temporal_api_notificationservice_v1_OnCompleteResponse( ) -> None: pass + async def _visit_temporal_api_streamservice_v1_CreateStreamRequest( + self, fs: VisitorFunctions, o: Any + ) -> None: + pass + + async def _visit_temporal_api_streamservice_v1_CreateStreamResponse( + self, fs: VisitorFunctions, o: Any + ) -> None: + pass + + async def _visit_temporal_api_stream_v1_AppendRecord( + self, fs: VisitorFunctions, o: Any + ): + if o.HasField("body"): + await self._visit_temporal_api_common_v1_Payload(fs, o.body) + + async def _visit_temporal_api_streamservice_v1_AppendRecordsRequest( + self, fs: VisitorFunctions, o: Any + ): + for v in o.records: + await self._visit_temporal_api_stream_v1_AppendRecord(fs, v) + + async def _visit_temporal_api_streamservice_v1_AppendRecordsResponse( + self, fs: VisitorFunctions, o: Any + ) -> None: + pass + + async def _visit_temporal_api_streamservice_v1_CloseStreamRequest( + self, fs: VisitorFunctions, o: Any + ) -> None: + pass + + async def _visit_temporal_api_streamservice_v1_CloseStreamResponse( + self, fs: VisitorFunctions, o: Any + ) -> None: + pass + + async def _visit_temporal_api_streamservice_v1_ReadRecordsRequest( + self, fs: VisitorFunctions, o: Any + ) -> None: + pass + + async def _visit_temporal_api_stream_v1_StreamRecord( + self, fs: VisitorFunctions, o: Any + ): + if o.HasField("body"): + await self._visit_temporal_api_common_v1_Payload(fs, o.body) + + async def _visit_temporal_api_streamservice_v1_ReadRecordsResponse( + self, fs: VisitorFunctions, o: Any + ): + for v in o.records: + await self._visit_temporal_api_stream_v1_StreamRecord(fs, v) + + async def _visit_temporal_api_streamservice_v1_DescribeStreamRequest( + self, fs: VisitorFunctions, o: Any + ) -> None: + pass + + async def _visit_temporal_api_streamservice_v1_DescribeStreamResponse( + self, fs: VisitorFunctions, o: Any + ) -> None: + pass + async def _visit_temporal_api_common_v1_Header(self, fs: VisitorFunctions, o: Any): for v in o.fields.values(): await self._visit_temporal_api_common_v1_Payload(fs, v) diff --git a/temporalio/nexus/system/stream_service/__init__.py b/temporalio/nexus/system/stream_service/__init__.py new file mode 100644 index 0000000000..35b198843c --- /dev/null +++ b/temporalio/nexus/system/stream_service/__init__.py @@ -0,0 +1,71 @@ +# Generated by nexgen v0.2.12. DO NOT EDIT! + +from __future__ import annotations + +from .models import ( + ActivityEpoch, + ActivityProducer, + AppendRecord, + AppendRecordsRequest, + AppendRecordsResponse, + ClientProducer, + CloseStreamRequest, + CloseStreamResponse, + CreateStreamRequest, + CreateStreamResult, + DescribeStreamRequest, + DescribeStreamResponse, + LocalActivityEpoch, + LocalActivityProducer, + Producer, + ProducerKind, + ProducerKindClient, + ProducerKindLocalActivity, + ProducerKindStandaloneActivity, + ProducerKindWorkflow, + ProducerKindWorkflowActivity, + ReadRecordsRequest, + ReadRecordsResponse, + StreamIdConflictPolicy, + StreamRecord, + StreamRef, + StreamStatus, + Timestamp, + WorkflowProducer, + WorkflowTaskEpoch, +) +from .services import StreamService + +__all__ = [ + "ActivityEpoch", + "ActivityProducer", + "AppendRecord", + "AppendRecordsRequest", + "AppendRecordsResponse", + "ClientProducer", + "CloseStreamRequest", + "CloseStreamResponse", + "CreateStreamRequest", + "CreateStreamResult", + "DescribeStreamRequest", + "DescribeStreamResponse", + "LocalActivityEpoch", + "LocalActivityProducer", + "Producer", + "ProducerKind", + "ProducerKindWorkflow", + "ProducerKindLocalActivity", + "ProducerKindWorkflowActivity", + "ProducerKindStandaloneActivity", + "ProducerKindClient", + "ReadRecordsRequest", + "ReadRecordsResponse", + "StreamIdConflictPolicy", + "StreamRecord", + "StreamRef", + "StreamStatus", + "Timestamp", + "WorkflowProducer", + "WorkflowTaskEpoch", + "StreamService", +] diff --git a/temporalio/nexus/system/stream_service/_resource_context_manager.py b/temporalio/nexus/system/stream_service/_resource_context_manager.py new file mode 100644 index 0000000000..86dd499669 --- /dev/null +++ b/temporalio/nexus/system/stream_service/_resource_context_manager.py @@ -0,0 +1,93 @@ +# Generated by nexgen v0.2.12. DO NOT EDIT! + +from __future__ import annotations + +import collections.abc +import functools +import types +import typing + +import typing_extensions + + +class _Closable(typing.Protocol): + async def __aenter__(self) -> object: ... + + async def __aexit__( + self, + exc_type: type[BaseException] | None, + exc: BaseException | None, + tb: types.TracebackType | None, + ) -> None: ... + + +_ResourceT = typing.TypeVar("_ResourceT", bound=_Closable) +_P = typing_extensions.ParamSpec("_P") + + +@typing.final +class ResourceContextManager( + collections.abc.Coroutine[typing.Any, typing.Any, _ResourceT], + typing.Generic[_ResourceT], +): + """The create call. You can use it two ways. Await it to get the resource and manage it + yourself. Or open it with ``async with`` to get the resource and close it at the end of the + block. The ``async with`` form does the create for you, so you do not add a separate await.""" + + def __init__( + self, call: collections.abc.Coroutine[typing.Any, typing.Any, _ResourceT] + ) -> None: + self._call: collections.abc.Coroutine[typing.Any, typing.Any, _ResourceT] = call + self._resource: _ResourceT | None = None + self.__name__: str = getattr(call, "__name__", type(self).__name__) + self.__qualname__: str = getattr(call, "__qualname__", self.__name__) + + @typing_extensions.override + def __await__(self) -> collections.abc.Generator[typing.Any, None, _ResourceT]: + return self._call.__await__() + + @typing_extensions.override + def send(self, value: typing.Any) -> typing.Any: + return self._call.send(value) + + @typing_extensions.override + def throw( + self, + typ: typing.Any, + val: typing.Any = None, + tb: types.TracebackType | None = None, + ) -> typing.Any: + return self._call.throw(typ, val, tb) + + @typing_extensions.override + def close(self) -> None: + self._call.close() + + async def __aenter__(self) -> _ResourceT: + resource = await self._call + _ = await resource.__aenter__() + self._resource = resource + return resource + + async def __aexit__( + self, + exc_type: type[BaseException] | None, + exc: BaseException | None, + tb: types.TracebackType | None, + ) -> None: + if self._resource is not None: + await self._resource.__aexit__(exc_type, exc, tb) + + +def resource_context_manager( + function: collections.abc.Callable[ + _P, collections.abc.Coroutine[typing.Any, typing.Any, _ResourceT] + ], +) -> collections.abc.Callable[_P, ResourceContextManager[_ResourceT]]: + """Wraps a call that returns a closable resource in a ``ResourceContextManager``.""" + + @functools.wraps(function) + def call(*args: _P.args, **kwargs: _P.kwargs) -> ResourceContextManager[_ResourceT]: + return ResourceContextManager(function(*args, **kwargs)) + + return call diff --git a/temporalio/nexus/system/stream_service/client/__init__.py b/temporalio/nexus/system/stream_service/client/__init__.py new file mode 100644 index 0000000000..d2df633d4e --- /dev/null +++ b/temporalio/nexus/system/stream_service/client/__init__.py @@ -0,0 +1,11 @@ +# Generated by nexgen v0.2.12. DO NOT EDIT! + +from __future__ import annotations + +from ._extensions import StreamServiceClientExtension +from ._resources.stream import Stream + +__all__ = [ + "StreamServiceClientExtension", + "Stream", +] diff --git a/temporalio/nexus/system/stream_service/client/_extensions.py b/temporalio/nexus/system/stream_service/client/_extensions.py new file mode 100644 index 0000000000..c96bb985f8 --- /dev/null +++ b/temporalio/nexus/system/stream_service/client/_extensions.py @@ -0,0 +1,180 @@ +# Generated by nexgen v0.2.12. DO NOT EDIT! + +from __future__ import annotations + +import abc +import collections.abc +import datetime +import typing +import uuid as _uuid + +import temporalio.nexus.system + +if typing.TYPE_CHECKING: + import temporalio.client + +import temporalio.nexus.system._support as _support + +from .._resource_context_manager import resource_context_manager +from ..models import ( + CreateStreamRequest, + CreateStreamResult, + StreamIdConflictPolicy, + StreamRef, +) +from ..services import StreamService +from ._resources.stream import Stream + +_NexusServiceT = typing.TypeVar("_NexusServiceT") +BodyT = typing.TypeVar("BodyT") + + +class StreamServiceClientExtension(abc.ABC): + """Starts the operations of ``StreamService`` from code outside a Workflow. + + A client class inherits this class and provides the SDK's ``create_nexus_client`` + method. + + .. warning:: + This API is experimental and subject to change. + """ + + @abc.abstractmethod + def create_nexus_client( + self, + service: type[_NexusServiceT] | str, + endpoint: str, + ) -> temporalio.client.NexusClient[_NexusServiceT]: + """Return the Nexus client that starts the operations of ``service``.""" + ... + + @typing.overload + def get_stream( + self, + ref: StreamRef[BodyT], + *, + created: bool | None = None, + body_type: None = None, + ) -> Stream[BodyT]: ... + @typing.overload + def get_stream( + self, + ref: StreamRef[BodyT] | StreamRef[object], + *, + created: bool | None = None, + body_type: type[BodyT], + ) -> Stream[BodyT]: ... + def get_stream( + self, + ref: StreamRef[typing.Any], + *, + created: bool | None = None, + body_type: typing.Any = None, + ) -> Stream[typing.Any]: + """Return the ``Stream`` handle for this ref.""" + arguments = typing.get_args(getattr(ref, "__orig_class__", None)) + # None means no explicit type, so use the ref's type argument, else object. + if body_type is None: + body_type = arguments[0] if arguments else object + elif arguments and arguments[0] is not object and arguments[0] != body_type: + raise TypeError("The decoding type must match the typed ref") + if arguments != (body_type,): + ref = StreamRef[body_type](id=ref.id, store=ref.store) + return Stream[body_type]( + self.create_nexus_client( + service=StreamService, + endpoint=temporalio.nexus.system.TEMPORAL_SYSTEM_ENDPOINT, + ), + ref=ref, + created=created, + ) + + async def _stream_service_create_stream( + self, + request: CreateStreamRequest[BodyT], + body_type: type[BodyT], + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, + ) -> Stream[BodyT]: + nexus_client = self.create_nexus_client( + service=StreamService, + endpoint=temporalio.nexus.system.TEMPORAL_SYSTEM_ENDPOINT, + ) + handle = await nexus_client.start_operation( + operation="CreateStream", + arg=request, + id=f"{_uuid.uuid4()}", + result_type=CreateStreamResult[body_type], + rpc_metadata=rpc_metadata or {}, + rpc_timeout=rpc_timeout, + ) + result = await handle.result() + if result.ref is None: + raise RuntimeError("CreateStream answered without `ref`") + resource = Stream[body_type]( + nexus_client, + ref=result.ref, + created=result.created or None, + ) + return resource + + @typing.overload + @resource_context_manager + async def create_stream( + self, + *, + id: str, + store: str | None = None, + id_conflict_policy: StreamIdConflictPolicy | None = None, + body_type: type[BodyT], + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, + ) -> Stream[BodyT]: ... + + @typing.overload + @resource_context_manager + async def create_stream( + self, + *, + id: str, + store: str | None = None, + id_conflict_policy: StreamIdConflictPolicy | None = None, + body_type: None = None, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, + ) -> Stream[object]: ... + + @resource_context_manager + async def create_stream( + self, + *, + id: str, + store: str | None = None, + id_conflict_policy: StreamIdConflictPolicy | None = None, + body_type: typing.Any = None, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, + ) -> Stream[typing.Any]: + """ + Args: + id: The Stream id, 1 to 4096 bytes. + store: Registered Store name. With none, the SDK picks one. Only an SDK + reads it. + id_conflict_policy: What the create does when the Stream exists. Unspecified + is fail. + """ + if body_type is None: + body_type = typing.Any + stream_ref = StreamRef( + id=id, + store=store, + ) + request = CreateStreamRequest( + stream_ref=stream_ref, + id_conflict_policy=id_conflict_policy, + request_id=_support.request_id(), + ) + return await self._stream_service_create_stream( + request, body_type, rpc_metadata=rpc_metadata, rpc_timeout=rpc_timeout + ) diff --git a/temporalio/nexus/system/stream_service/client/_poll.py b/temporalio/nexus/system/stream_service/client/_poll.py new file mode 100644 index 0000000000..a653d84f93 --- /dev/null +++ b/temporalio/nexus/system/stream_service/client/_poll.py @@ -0,0 +1,77 @@ +# Generated by nexgen v0.2.12. DO NOT EDIT! + +from __future__ import annotations + +import collections +import collections.abc +import typing + +_PollRequestT = typing.TypeVar("_PollRequestT") +_PollResponseT = typing.TypeVar("_PollResponseT") +_PollItemT = typing.TypeVar("_PollItemT") + + +@typing.final +class PollIterator(typing.Generic[_PollRequestT, _PollItemT]): + """Yields each item of the pages of a poll. + + ``next_request`` is the request to resume with. After an item, the resume helper gives + it. Without a resume helper, it reads the item's page again. After a page's last item, + it is the page helper's request, which is ``None`` after the last page. + """ + + def __init__( + self, + request: _PollRequestT, + pages: collections.abc.AsyncIterator[_PollResponseT], + items: collections.abc.Callable[ + [_PollResponseT], collections.abc.Sequence[_PollItemT] | None + ], + next_page: collections.abc.Callable[ + [_PollRequestT, _PollResponseT], _PollRequestT | None + ], + resume: collections.abc.Callable[[_PollRequestT, _PollItemT], _PollRequestT] + | None, + ) -> None: + async def read( + page_request: _PollRequestT, + ) -> tuple[_PollRequestT | None, collections.deque[_PollItemT]]: + response = await anext(pages) + # The page iterator keeps its requests to itself, so the page helper runs here + # again. It runs before the caller gets an item, so a changed item can't change + # the next request. + return next_page(page_request, response), collections.deque( + items(response) or () + ) + + self.next_request: _PollRequestT | None = request + self._read: collections.abc.Callable[ + [_PollRequestT], + collections.abc.Awaitable[ + tuple[_PollRequestT | None, collections.deque[_PollItemT]] + ], + ] = read + self._resume: ( + collections.abc.Callable[[_PollRequestT, _PollItemT], _PollRequestT] | None + ) = resume + self._page: collections.deque[_PollItemT] = collections.deque() + self._page_request: _PollRequestT = request + self._page_next: _PollRequestT | None = None + + def __aiter__(self) -> PollIterator[_PollRequestT, _PollItemT]: + return self + + async def __anext__(self) -> _PollItemT: + while not self._page: + if self.next_request is None: + raise StopAsyncIteration + self._page_request = self.next_request + self._page_next, self._page = await self._read(self._page_request) + if not self._page: + self.next_request = self._page_next + item = self._page.popleft() + if not self._page: + self.next_request = self._page_next + elif self._resume is not None: + self.next_request = self._resume(self._page_request, item) + return item diff --git a/temporalio/nexus/system/stream_service/client/_resources/__init__.py b/temporalio/nexus/system/stream_service/client/_resources/__init__.py new file mode 100644 index 0000000000..250b6a42b7 --- /dev/null +++ b/temporalio/nexus/system/stream_service/client/_resources/__init__.py @@ -0,0 +1,3 @@ +# Generated by nexgen v0.2.12. DO NOT EDIT! + +from __future__ import annotations diff --git a/temporalio/nexus/system/stream_service/client/_resources/stream.py b/temporalio/nexus/system/stream_service/client/_resources/stream.py new file mode 100644 index 0000000000..c3fa617b6d --- /dev/null +++ b/temporalio/nexus/system/stream_service/client/_resources/stream.py @@ -0,0 +1,261 @@ +# Generated by nexgen v0.2.12. DO NOT EDIT! + +from __future__ import annotations + +import collections.abc +import datetime +import types +import typing +import uuid as _uuid + +import typing_extensions + +if typing.TYPE_CHECKING: + import temporalio.client + +import temporalio.nexus.system._support as _support + +from ...models import ( + AppendRecord, + AppendRecordsRequest, + AppendRecordsResponse, + CloseStreamRequest, + DescribeStreamRequest, + DescribeStreamResponse, + Producer, + ReadRecordsRequest, + ReadRecordsResponse, + StreamRecord, + StreamRef, +) +from ...services import StreamService +from .._poll import PollIterator + +BodyT = typing.TypeVar("BodyT") + + +class Stream(typing.Generic[BodyT]): + """One Stream. Build it from a stream-ref, or get it from a create.""" + + ref: StreamRef[BodyT] + created: bool | None + + def __init__( + self, + nexus_client: temporalio.client.NexusClient[typing.Any], + /, + *, + ref: StreamRef[BodyT], + created: bool | None, + ) -> None: + self._nexus_client: temporalio.client.NexusClient[typing.Any] = nexus_client + if type(self) is not Stream: + raise TypeError("Generic resource subclasses are not supported") + self.ref = ref + self.created = created + + @property + def body_type(self) -> type[BodyT]: + arguments = typing.get_args(getattr(self.ref, "__orig_class__", None)) + return typing.cast("type[BodyT]", arguments[0] if arguments else object) + + async def __aenter__(self) -> typing_extensions.Self: + return self + + async def __aexit__( + self, + exc_type: type[BaseException] | None, + exc: BaseException | None, + tb: types.TracebackType | None, + ) -> None: + await self.close() + + async def append( + self, + *records: BodyT, + producer: Producer | None = None, + sequence: int | None = None, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, + ) -> AppendRecordsResponse: + """Append values, and return the offsets of the batch.""" + request = AppendRecordsRequest( + ref=self.ref, + producer=((_support.default_producer()) if producer is None else producer), + sequence=sequence, + records=[AppendRecord(body=value) for value in records], + ) + handle = await self._nexus_client.start_operation( + operation=StreamService.append_records, + arg=request, + id=f"{_uuid.uuid4()}", + rpc_metadata=rpc_metadata or {}, + rpc_timeout=rpc_timeout, + ) + return await handle.result() + + async def close( + self, + producer: Producer | None = None, + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, + ) -> None: + """Close the Stream. Readers drain what is left, then end.""" + request = CloseStreamRequest( + ref=self.ref, + producer=((_support.default_producer()) if producer is None else producer), + ) + handle = await self._nexus_client.start_operation( + operation=StreamService.close_stream, + arg=request, + id=f"{_uuid.uuid4()}", + rpc_metadata=rpc_metadata or {}, + rpc_timeout=rpc_timeout, + ) + _ = await handle.result() + + async def read( + self, + from_offset: bytes | None = None, + max_records: int | None = None, + wait: datetime.timedelta | None = None, + max_bytes: int | None = None, + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, + ) -> ReadRecordsResponse[BodyT]: + """Read one page of Records. The poll iterator reads them all.""" + request = ReadRecordsRequest( + ref=self.ref, + from_offset=from_offset, + max_records=max_records, + wait=wait, + max_bytes=max_bytes, + ) + handle = await self._nexus_client.start_operation( + operation="ReadRecords", + arg=request, + id=f"{_uuid.uuid4()}", + result_type=ReadRecordsResponse[self.body_type], + rpc_metadata=rpc_metadata or {}, + rpc_timeout=rpc_timeout, + ) + return await handle.result() + + async def iter_read( + self, + from_offset: bytes | None = None, + max_records: int | None = None, + wait: datetime.timedelta | None = None, + max_bytes: int | None = None, + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, + ) -> collections.abc.AsyncGenerator[ReadRecordsResponse[BodyT], None]: + """Yield each response of ``read`` until the poll helper stops. + + Read one page of Records. The poll iterator reads them all. + + Args: + from_offset: Read after this offset. Empty is the beginning. A too high + offset fails. + max_records: About this many Records per page, 1 to 1000. + wait: How long to wait for a new Record before an empty answer. At most 60 + s. + max_bytes: The maximum Record bytes in a page. A larger Record is still + returned alone. + """ + request = ReadRecordsRequest( + ref=self.ref, + from_offset=from_offset, + max_records=max_records, + wait=wait, + max_bytes=max_bytes, + ) + nexus_client = self._nexus_client + while True: + operation_handle = await nexus_client.start_operation( + operation="ReadRecords", + arg=request, + id=f"{_uuid.uuid4()}", + result_type=ReadRecordsResponse[self.body_type], + rpc_metadata=rpc_metadata or {}, + rpc_timeout=rpc_timeout, + ) + response = await operation_handle.result() + next_request = _support.next_read_request(request, response) + yield response + if next_request is None: + return + request = next_request + + def iter_read_items( + self, + from_offset: bytes | None = None, + max_records: int | None = None, + wait: datetime.timedelta | None = None, + max_bytes: int | None = None, + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, + ) -> PollIterator[ReadRecordsRequest[BodyT], StreamRecord[BodyT]]: + """Yield each item of ``iter_read``, page by page. + + ``next_request`` resumes after the last item that the iterator yielded. It is + ``None`` after the last page. + + Read one page of Records. The poll iterator reads them all. + + Args: + from_offset: Read after this offset. Empty is the beginning. A too high + offset fails. + max_records: About this many Records per page, 1 to 1000. + wait: How long to wait for a new Record before an empty answer. At most 60 + s. + max_bytes: The maximum Record bytes in a page. A larger Record is still + returned alone. + """ + request = ReadRecordsRequest( + ref=self.ref, + from_offset=from_offset, + max_records=max_records, + wait=wait, + max_bytes=max_bytes, + ) + return PollIterator( + request, + self.iter_read( + from_offset=from_offset, + max_records=max_records, + wait=wait, + max_bytes=max_bytes, + rpc_metadata=rpc_metadata, + rpc_timeout=rpc_timeout, + ), + lambda response: response.records, + _support.next_read_request, + _support.resume_read_request, + ) + + async def describe( + self, + *, + rpc_metadata: collections.abc.Mapping[str, str | bytes] | None = None, + rpc_timeout: datetime.timedelta | None = None, + ) -> DescribeStreamResponse: + """Describe the Stream's status and newest offset.""" + request = DescribeStreamRequest( + ref=self.ref, + ) + handle = await self._nexus_client.start_operation( + operation=StreamService.describe_stream, + arg=request, + id=f"{_uuid.uuid4()}", + rpc_metadata=rpc_metadata or {}, + rpc_timeout=rpc_timeout, + ) + return await handle.result() + + +__all__ = ["Stream"] diff --git a/temporalio/nexus/system/stream_service/models.py b/temporalio/nexus/system/stream_service/models.py new file mode 100644 index 0000000000..4848a2ea58 --- /dev/null +++ b/temporalio/nexus/system/stream_service/models.py @@ -0,0 +1,1371 @@ +# Generated by nexgen v0.2.12. DO NOT EDIT! + +from __future__ import annotations + +import dataclasses +import datetime +import enum +import typing + +import google.protobuf.timestamp_pb2 +import typing_extensions + +import temporalio.api.enums.v1.stream_pb2 +import temporalio.api.stream.v1.message_pb2 +import temporalio.api.streamservice.v1.request_response_pb2 +import temporalio.converter +import temporalio.nexus.system._support as _support + + +class StreamIdConflictPolicy(enum.IntEnum): + StreamIdConflictPolicyUnspecified = 0 + StreamIdConflictPolicyFail = 1 + StreamIdConflictPolicyUseExisting = 2 + + +class StreamStatus(enum.IntEnum): + StreamStatusUnspecified = 0 + StreamStatusOpen = 1 + StreamStatusClosed = 2 + StreamStatusExpired = 3 + + +def _stream_status_from_proto(value: int) -> StreamStatus: + try: + return StreamStatus(value) + except ValueError: + # A newer peer can send a case that this enum doesn't know. + return StreamStatus.StreamStatusUnspecified + + +BodyT = typing.TypeVar("BodyT") +_BodyT = typing_extensions.TypeVar("_BodyT", default=object) + + +@dataclasses.dataclass(slots=True) +class CreateStreamRequest(typing.Generic[BodyT]): + """Create a Stream. If it exists, fail, or with use-existing get the open one.""" + + stream_ref: StreamRef[BodyT] + id_conflict_policy: StreamIdConflictPolicy | None = None + request_id: str | None = None + + +class _CreateStreamRequestTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + CreateStreamRequest[BodyT], + temporalio.api.streamservice.v1.request_response_pb2.CreateStreamRequest, + ] +): + transfer_type: ( + type[temporalio.api.streamservice.v1.request_response_pb2.CreateStreamRequest] + | None + ) = temporalio.api.streamservice.v1.request_response_pb2.CreateStreamRequest + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.streamservice.v1.request_response_pb2.CreateStreamRequest, + type_hint: type[CreateStreamRequest[BodyT]], + ) -> CreateStreamRequest[BodyT]: + (body_type,) = typing.get_args(type_hint) or (typing.Any,) + if not value.HasField("stream_ref"): + raise ValueError("missing required field CreateStreamRequest.stream_ref") + stream_ref = typing.cast( + StreamRef[BodyT], + _StreamRefTransferTypeConverter().from_transfer_type( + value.stream_ref, StreamRef[body_type] + ), + ) + return CreateStreamRequest( + stream_ref=stream_ref, + id_conflict_policy=StreamIdConflictPolicy(value.id_conflict_policy) + if value.id_conflict_policy != 0 + else None, + request_id=value.request_id if bool(value.request_id) else None, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: CreateStreamRequest[BodyT], + ) -> temporalio.api.streamservice.v1.request_response_pb2.CreateStreamRequest: + message = ( + temporalio.api.streamservice.v1.request_response_pb2.CreateStreamRequest() + ) + message.stream_ref.CopyFrom( + _StreamRefTransferTypeConverter().to_transfer_type(value.stream_ref) + ) + if value.id_conflict_policy is not None: + message.id_conflict_policy = ( + temporalio.api.enums.v1.stream_pb2.StreamIdConflictPolicy.ValueType( + value.id_conflict_policy + ) + ) + if value.request_id is not None: + message.request_id = value.request_id + return message + + +temporalio.converter.transfer_type_convertible( + _CreateStreamRequestTransferTypeConverter[typing.Any] +)(CreateStreamRequest) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass +class StreamRef(typing.Generic[_BodyT]): + """Identifies one Stream. The id is opaque to callers.""" + + id: str + store: str | None = None + + +class _StreamRefTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + StreamRef[BodyT], temporalio.api.stream.v1.message_pb2.StreamRef + ] +): + transfer_type: type[temporalio.api.stream.v1.message_pb2.StreamRef] | None = ( + temporalio.api.stream.v1.message_pb2.StreamRef + ) + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.stream.v1.message_pb2.StreamRef, + type_hint: type[StreamRef[BodyT]], + ) -> StreamRef[BodyT]: + (_body_type,) = typing.get_args(type_hint) or (typing.Any,) + if not value.id: + raise ValueError("missing required field StreamRef.id") + id = value.id + return type_hint( + id=id, + store=value.store if bool(value.store) else None, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: StreamRef[BodyT], + ) -> temporalio.api.stream.v1.message_pb2.StreamRef: + message = temporalio.api.stream.v1.message_pb2.StreamRef() + message.id = value.id + if value.store is not None: + message.store = value.store + return message + + +temporalio.converter.transfer_type_convertible( + _StreamRefTransferTypeConverter[typing.Any] +)(StreamRef) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class CreateStreamResult(typing.Generic[BodyT]): + ref: StreamRef[BodyT] | None = None + created: bool | None = None + + +class _CreateStreamResultTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + CreateStreamResult[BodyT], + temporalio.api.streamservice.v1.request_response_pb2.CreateStreamResponse, + ] +): + transfer_type: ( + type[temporalio.api.streamservice.v1.request_response_pb2.CreateStreamResponse] + | None + ) = temporalio.api.streamservice.v1.request_response_pb2.CreateStreamResponse + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.streamservice.v1.request_response_pb2.CreateStreamResponse, + type_hint: type[CreateStreamResult[BodyT]], + ) -> CreateStreamResult[BodyT]: + (body_type,) = typing.get_args(type_hint) or (typing.Any,) + return CreateStreamResult( + ref=typing.cast( + StreamRef[BodyT], + _StreamRefTransferTypeConverter().from_transfer_type( + value.ref, StreamRef[body_type] + ), + ) + if value.HasField("ref") + else None, + created=value.created if bool(value.created) else None, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: CreateStreamResult[BodyT], + ) -> temporalio.api.streamservice.v1.request_response_pb2.CreateStreamResponse: + message = ( + temporalio.api.streamservice.v1.request_response_pb2.CreateStreamResponse() + ) + if value.ref is not None: + message.ref.CopyFrom( + _StreamRefTransferTypeConverter().to_transfer_type(value.ref) + ) + if value.created is not None: + message.created = value.created + return message + + +temporalio.converter.transfer_type_convertible( + _CreateStreamResultTransferTypeConverter[typing.Any] +)(CreateStreamResult) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class AppendRecordsRequest(typing.Generic[BodyT]): + """Append one batch. The Store keeps a resent batch once, by Producer, epoch and + sequence. + """ + + ref: StreamRef[BodyT] + records: list[AppendRecord[BodyT]] + producer: Producer + sequence: int | None = None + + +class _AppendRecordsRequestTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + AppendRecordsRequest[BodyT], + temporalio.api.streamservice.v1.request_response_pb2.AppendRecordsRequest, + ] +): + transfer_type: ( + type[temporalio.api.streamservice.v1.request_response_pb2.AppendRecordsRequest] + | None + ) = temporalio.api.streamservice.v1.request_response_pb2.AppendRecordsRequest + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.streamservice.v1.request_response_pb2.AppendRecordsRequest, + type_hint: type[AppendRecordsRequest[BodyT]], + ) -> AppendRecordsRequest[BodyT]: + (body_type,) = typing.get_args(type_hint) or (typing.Any,) + if not value.HasField("ref"): + raise ValueError("missing required field AppendRecordsRequest.ref") + ref = typing.cast( + StreamRef[BodyT], + _StreamRefTransferTypeConverter().from_transfer_type( + value.ref, StreamRef[body_type] + ), + ) + records = [ + typing.cast( + AppendRecord[BodyT], + _AppendRecordTransferTypeConverter().from_transfer_type( + item, AppendRecord[body_type] + ), + ) + for item in value.records + ] + if not value.HasField("producer"): + raise ValueError("missing required field AppendRecordsRequest.producer") + producer = _ProducerTransferTypeConverter().from_transfer_type( + value.producer, Producer + ) + return AppendRecordsRequest( + ref=ref, + records=records, + producer=producer, + sequence=value.sequence if value.sequence != 0 else None, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: AppendRecordsRequest[BodyT], + ) -> temporalio.api.streamservice.v1.request_response_pb2.AppendRecordsRequest: + message = ( + temporalio.api.streamservice.v1.request_response_pb2.AppendRecordsRequest() + ) + message.ref.CopyFrom( + _StreamRefTransferTypeConverter().to_transfer_type(value.ref) + ) + for element in value.records: + item = message.records.add() + item.CopyFrom( + _AppendRecordTransferTypeConverter().to_transfer_type(element) + ) + message.producer.CopyFrom( + _ProducerTransferTypeConverter().to_transfer_type(value.producer) + ) + if value.sequence is not None: + message.sequence = value.sequence + return message + + +temporalio.converter.transfer_type_convertible( + _AppendRecordsRequestTransferTypeConverter[typing.Any] +)(AppendRecordsRequest) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class AppendRecord(typing.Generic[BodyT]): + """One Record to append.""" + + body: BodyT + + +class _AppendRecordTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + AppendRecord[BodyT], temporalio.api.stream.v1.message_pb2.AppendRecord + ] +): + transfer_type: type[temporalio.api.stream.v1.message_pb2.AppendRecord] | None = ( + temporalio.api.stream.v1.message_pb2.AppendRecord + ) + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.stream.v1.message_pb2.AppendRecord, + type_hint: type[AppendRecord[BodyT]], + ) -> AppendRecord[BodyT]: + (body_type,) = typing.get_args(type_hint) or (typing.Any,) + if not value.HasField("body"): + raise ValueError("missing required field AppendRecord.body") + body = typing.cast(BodyT, _support.payload_from_proto(value.body, body_type)) + return AppendRecord( + body=body, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: AppendRecord[BodyT], + ) -> temporalio.api.stream.v1.message_pb2.AppendRecord: + message = temporalio.api.stream.v1.message_pb2.AppendRecord() + message.body.CopyFrom(_support.payload_to_proto(value.body)) + return message + + +temporalio.converter.transfer_type_convertible( + _AppendRecordTransferTypeConverter[typing.Any] +)(AppendRecord) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class Producer: + """The code that wrote a batch. A higher epoch fences and supersedes a lower one.""" + + id: str + kind: ProducerKind | None = None + + +class _ProducerTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + Producer, temporalio.api.stream.v1.message_pb2.Producer + ] +): + transfer_type: type[temporalio.api.stream.v1.message_pb2.Producer] | None = ( + temporalio.api.stream.v1.message_pb2.Producer + ) + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.stream.v1.message_pb2.Producer, + type_hint: type[Producer], + ) -> Producer: + if not value.id: + raise ValueError("missing required field Producer.id") + id = value.id + _oneof_kind: ( + ProducerKindWorkflow + | ProducerKindLocalActivity + | ProducerKindWorkflowActivity + | ProducerKindStandaloneActivity + | ProducerKindClient + | None + ) + match value.WhichOneof("kind"): + case "workflow": + _oneof_kind = ProducerKindWorkflow( + _WorkflowProducerTransferTypeConverter().from_transfer_type( + value.workflow, WorkflowProducer + ) + ) + case "local_activity": + _oneof_kind = ProducerKindLocalActivity( + _LocalActivityProducerTransferTypeConverter().from_transfer_type( + value.local_activity, LocalActivityProducer + ) + ) + case "workflow_activity": + _oneof_kind = ProducerKindWorkflowActivity( + _ActivityProducerTransferTypeConverter().from_transfer_type( + value.workflow_activity, ActivityProducer + ) + ) + case "standalone_activity": + _oneof_kind = ProducerKindStandaloneActivity( + _ActivityProducerTransferTypeConverter().from_transfer_type( + value.standalone_activity, ActivityProducer + ) + ) + case "client": + _oneof_kind = ProducerKindClient( + _ClientProducerTransferTypeConverter().from_transfer_type( + value.client, ClientProducer + ) + ) + case None: + _oneof_kind = None + return Producer( + id=id, + kind=_oneof_kind, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: Producer, + ) -> temporalio.api.stream.v1.message_pb2.Producer: + message = temporalio.api.stream.v1.message_pb2.Producer() + message.id = value.id + if value.kind is not None: + match value.kind: + case ProducerKindWorkflow(): + message.workflow.CopyFrom( + _WorkflowProducerTransferTypeConverter().to_transfer_type( + value.kind.value + ) + ) + case ProducerKindLocalActivity(): + message.local_activity.CopyFrom( + _LocalActivityProducerTransferTypeConverter().to_transfer_type( + value.kind.value + ) + ) + case ProducerKindWorkflowActivity(): + message.workflow_activity.CopyFrom( + _ActivityProducerTransferTypeConverter().to_transfer_type( + value.kind.value + ) + ) + case ProducerKindStandaloneActivity(): + message.standalone_activity.CopyFrom( + _ActivityProducerTransferTypeConverter().to_transfer_type( + value.kind.value + ) + ) + case ProducerKindClient(): + message.client.CopyFrom( + _ClientProducerTransferTypeConverter().to_transfer_type( + value.kind.value + ) + ) + return message + + +temporalio.converter.transfer_type_convertible(_ProducerTransferTypeConverter)(Producer) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class WorkflowProducer: + """A Workflow run Producer.""" + + run_id: str + epoch: WorkflowTaskEpoch | None = None + + +class _WorkflowProducerTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + WorkflowProducer, temporalio.api.stream.v1.message_pb2.WorkflowProducer + ] +): + transfer_type: ( + type[temporalio.api.stream.v1.message_pb2.WorkflowProducer] | None + ) = temporalio.api.stream.v1.message_pb2.WorkflowProducer + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.stream.v1.message_pb2.WorkflowProducer, + type_hint: type[WorkflowProducer], + ) -> WorkflowProducer: + if not value.run_id: + raise ValueError("missing required field WorkflowProducer.run_id") + run_id = value.run_id + return WorkflowProducer( + run_id=run_id, + epoch=_WorkflowTaskEpochTransferTypeConverter().from_transfer_type( + value.epoch, WorkflowTaskEpoch + ) + if value.HasField("epoch") + else None, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: WorkflowProducer, + ) -> temporalio.api.stream.v1.message_pb2.WorkflowProducer: + message = temporalio.api.stream.v1.message_pb2.WorkflowProducer() + message.run_id = value.run_id + if value.epoch is not None: + message.epoch.CopyFrom( + _WorkflowTaskEpochTransferTypeConverter().to_transfer_type(value.epoch) + ) + return message + + +temporalio.converter.transfer_type_convertible(_WorkflowProducerTransferTypeConverter)( + WorkflowProducer +) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class WorkflowTaskEpoch: + """A Workflow Task attempt of the run.""" + + started_event_id: int + attempt: int + started_time: Timestamp | None = None + + +class _WorkflowTaskEpochTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + WorkflowTaskEpoch, temporalio.api.stream.v1.message_pb2.WorkflowTaskEpoch + ] +): + transfer_type: ( + type[temporalio.api.stream.v1.message_pb2.WorkflowTaskEpoch] | None + ) = temporalio.api.stream.v1.message_pb2.WorkflowTaskEpoch + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.stream.v1.message_pb2.WorkflowTaskEpoch, + type_hint: type[WorkflowTaskEpoch], + ) -> WorkflowTaskEpoch: + started_event_id = value.started_event_id + attempt = value.attempt + return WorkflowTaskEpoch( + started_event_id=started_event_id, + attempt=attempt, + started_time=_TimestampTransferTypeConverter().from_transfer_type( + value.started_time, Timestamp + ) + if value.HasField("started_time") + else None, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: WorkflowTaskEpoch, + ) -> temporalio.api.stream.v1.message_pb2.WorkflowTaskEpoch: + message = temporalio.api.stream.v1.message_pb2.WorkflowTaskEpoch() + message.started_event_id = value.started_event_id + message.attempt = value.attempt + if value.started_time is not None: + message.started_time.CopyFrom( + _TimestampTransferTypeConverter().to_transfer_type(value.started_time) + ) + return message + + +temporalio.converter.transfer_type_convertible(_WorkflowTaskEpochTransferTypeConverter)( + WorkflowTaskEpoch +) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class Timestamp: + seconds: int + nanos: int + + +class _TimestampTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + Timestamp, google.protobuf.timestamp_pb2.Timestamp + ] +): + transfer_type: type[google.protobuf.timestamp_pb2.Timestamp] | None = ( + google.protobuf.timestamp_pb2.Timestamp + ) + + @typing_extensions.override + def from_transfer_type( + self, + value: google.protobuf.timestamp_pb2.Timestamp, + type_hint: type[Timestamp], + ) -> Timestamp: + seconds = value.seconds + nanos = value.nanos + return Timestamp( + seconds=seconds, + nanos=nanos, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: Timestamp, + ) -> google.protobuf.timestamp_pb2.Timestamp: + message = google.protobuf.timestamp_pb2.Timestamp() + message.seconds = value.seconds + message.nanos = value.nanos + return message + + +temporalio.converter.transfer_type_convertible(_TimestampTransferTypeConverter)( + Timestamp +) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class LocalActivityProducer: + """A Local Activity Producer.""" + + run_id: str + epoch: LocalActivityEpoch + + +class _LocalActivityProducerTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + LocalActivityProducer, + temporalio.api.stream.v1.message_pb2.LocalActivityProducer, + ] +): + transfer_type: ( + type[temporalio.api.stream.v1.message_pb2.LocalActivityProducer] | None + ) = temporalio.api.stream.v1.message_pb2.LocalActivityProducer + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.stream.v1.message_pb2.LocalActivityProducer, + type_hint: type[LocalActivityProducer], + ) -> LocalActivityProducer: + if not value.run_id: + raise ValueError("missing required field LocalActivityProducer.run_id") + run_id = value.run_id + if not value.HasField("epoch"): + raise ValueError("missing required field LocalActivityProducer.epoch") + epoch = _LocalActivityEpochTransferTypeConverter().from_transfer_type( + value.epoch, LocalActivityEpoch + ) + return LocalActivityProducer( + run_id=run_id, + epoch=epoch, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: LocalActivityProducer, + ) -> temporalio.api.stream.v1.message_pb2.LocalActivityProducer: + message = temporalio.api.stream.v1.message_pb2.LocalActivityProducer() + message.run_id = value.run_id + message.epoch.CopyFrom( + _LocalActivityEpochTransferTypeConverter().to_transfer_type(value.epoch) + ) + return message + + +temporalio.converter.transfer_type_convertible( + _LocalActivityProducerTransferTypeConverter +)(LocalActivityProducer) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class LocalActivityEpoch: + """A Local Activity attempt. Its Workflow Task orders it first.""" + + task: WorkflowTaskEpoch + attempt: int + + +class _LocalActivityEpochTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + LocalActivityEpoch, temporalio.api.stream.v1.message_pb2.LocalActivityEpoch + ] +): + transfer_type: ( + type[temporalio.api.stream.v1.message_pb2.LocalActivityEpoch] | None + ) = temporalio.api.stream.v1.message_pb2.LocalActivityEpoch + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.stream.v1.message_pb2.LocalActivityEpoch, + type_hint: type[LocalActivityEpoch], + ) -> LocalActivityEpoch: + if not value.HasField("task"): + raise ValueError("missing required field LocalActivityEpoch.task") + task = _WorkflowTaskEpochTransferTypeConverter().from_transfer_type( + value.task, WorkflowTaskEpoch + ) + attempt = value.attempt + return LocalActivityEpoch( + task=task, + attempt=attempt, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: LocalActivityEpoch, + ) -> temporalio.api.stream.v1.message_pb2.LocalActivityEpoch: + message = temporalio.api.stream.v1.message_pb2.LocalActivityEpoch() + message.task.CopyFrom( + _WorkflowTaskEpochTransferTypeConverter().to_transfer_type(value.task) + ) + message.attempt = value.attempt + return message + + +temporalio.converter.transfer_type_convertible( + _LocalActivityEpochTransferTypeConverter +)(LocalActivityEpoch) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True, kw_only=True) +class ActivityProducer: + """An Activity Producer.""" + + run_id: str | None = None + epoch: ActivityEpoch + + +class _ActivityProducerTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + ActivityProducer, temporalio.api.stream.v1.message_pb2.ActivityProducer + ] +): + transfer_type: ( + type[temporalio.api.stream.v1.message_pb2.ActivityProducer] | None + ) = temporalio.api.stream.v1.message_pb2.ActivityProducer + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.stream.v1.message_pb2.ActivityProducer, + type_hint: type[ActivityProducer], + ) -> ActivityProducer: + if not value.HasField("epoch"): + raise ValueError("missing required field ActivityProducer.epoch") + epoch = _ActivityEpochTransferTypeConverter().from_transfer_type( + value.epoch, ActivityEpoch + ) + return ActivityProducer( + run_id=value.run_id if bool(value.run_id) else None, + epoch=epoch, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: ActivityProducer, + ) -> temporalio.api.stream.v1.message_pb2.ActivityProducer: + message = temporalio.api.stream.v1.message_pb2.ActivityProducer() + if value.run_id is not None: + message.run_id = value.run_id + message.epoch.CopyFrom( + _ActivityEpochTransferTypeConverter().to_transfer_type(value.epoch) + ) + return message + + +temporalio.converter.transfer_type_convertible(_ActivityProducerTransferTypeConverter)( + ActivityProducer +) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class ActivityEpoch: + """An Activity or Standalone Activity attempt.""" + + attempt_scheduled_time: Timestamp + attempt: int + + +class _ActivityEpochTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + ActivityEpoch, temporalio.api.stream.v1.message_pb2.ActivityEpoch + ] +): + transfer_type: type[temporalio.api.stream.v1.message_pb2.ActivityEpoch] | None = ( + temporalio.api.stream.v1.message_pb2.ActivityEpoch + ) + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.stream.v1.message_pb2.ActivityEpoch, + type_hint: type[ActivityEpoch], + ) -> ActivityEpoch: + if not value.HasField("attempt_scheduled_time"): + raise ValueError( + "missing required field ActivityEpoch.attempt_scheduled_time" + ) + attempt_scheduled_time = _TimestampTransferTypeConverter().from_transfer_type( + value.attempt_scheduled_time, Timestamp + ) + attempt = value.attempt + return ActivityEpoch( + attempt_scheduled_time=attempt_scheduled_time, + attempt=attempt, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: ActivityEpoch, + ) -> temporalio.api.stream.v1.message_pb2.ActivityEpoch: + message = temporalio.api.stream.v1.message_pb2.ActivityEpoch() + message.attempt_scheduled_time.CopyFrom( + _TimestampTransferTypeConverter().to_transfer_type( + value.attempt_scheduled_time + ) + ) + message.attempt = value.attempt + return message + + +temporalio.converter.transfer_type_convertible(_ActivityEpochTransferTypeConverter)( + ActivityEpoch +) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class ClientProducer: + """A client Producer. It has no run and no epoch, so nothing fences it.""" + + pass + + +class _ClientProducerTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + ClientProducer, temporalio.api.stream.v1.message_pb2.ClientProducer + ] +): + transfer_type: type[temporalio.api.stream.v1.message_pb2.ClientProducer] | None = ( + temporalio.api.stream.v1.message_pb2.ClientProducer + ) + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.stream.v1.message_pb2.ClientProducer, + type_hint: type[ClientProducer], + ) -> ClientProducer: + return ClientProducer() + + @typing_extensions.override + def to_transfer_type( + self, + value: ClientProducer, + ) -> temporalio.api.stream.v1.message_pb2.ClientProducer: + message = temporalio.api.stream.v1.message_pb2.ClientProducer() + return message + + +temporalio.converter.transfer_type_convertible(_ClientProducerTransferTypeConverter)( + ClientProducer +) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class AppendRecordsResponse: + first_offset: bytes | None = None + last_offset: bytes | None = None + + +class _AppendRecordsResponseTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + AppendRecordsResponse, + temporalio.api.streamservice.v1.request_response_pb2.AppendRecordsResponse, + ] +): + transfer_type: ( + type[temporalio.api.streamservice.v1.request_response_pb2.AppendRecordsResponse] + | None + ) = temporalio.api.streamservice.v1.request_response_pb2.AppendRecordsResponse + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.streamservice.v1.request_response_pb2.AppendRecordsResponse, + type_hint: type[AppendRecordsResponse], + ) -> AppendRecordsResponse: + return AppendRecordsResponse( + first_offset=value.first_offset if bool(value.first_offset) else None, + last_offset=value.last_offset if bool(value.last_offset) else None, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: AppendRecordsResponse, + ) -> temporalio.api.streamservice.v1.request_response_pb2.AppendRecordsResponse: + message = ( + temporalio.api.streamservice.v1.request_response_pb2.AppendRecordsResponse() + ) + if value.first_offset is not None: + message.first_offset = value.first_offset + if value.last_offset is not None: + message.last_offset = value.last_offset + return message + + +temporalio.converter.transfer_type_convertible( + _AppendRecordsResponseTransferTypeConverter +)(AppendRecordsResponse) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class CloseStreamRequest(typing.Generic[BodyT]): + """Close a Stream. Readers drain what is left, then end. Later appends are refused.""" + + ref: StreamRef[BodyT] + producer: Producer + + +class _CloseStreamRequestTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + CloseStreamRequest[BodyT], + temporalio.api.streamservice.v1.request_response_pb2.CloseStreamRequest, + ] +): + transfer_type: ( + type[temporalio.api.streamservice.v1.request_response_pb2.CloseStreamRequest] + | None + ) = temporalio.api.streamservice.v1.request_response_pb2.CloseStreamRequest + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.streamservice.v1.request_response_pb2.CloseStreamRequest, + type_hint: type[CloseStreamRequest[BodyT]], + ) -> CloseStreamRequest[BodyT]: + (body_type,) = typing.get_args(type_hint) or (typing.Any,) + if not value.HasField("ref"): + raise ValueError("missing required field CloseStreamRequest.ref") + ref = typing.cast( + StreamRef[BodyT], + _StreamRefTransferTypeConverter().from_transfer_type( + value.ref, StreamRef[body_type] + ), + ) + if not value.HasField("producer"): + raise ValueError("missing required field CloseStreamRequest.producer") + producer = _ProducerTransferTypeConverter().from_transfer_type( + value.producer, Producer + ) + return CloseStreamRequest( + ref=ref, + producer=producer, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: CloseStreamRequest[BodyT], + ) -> temporalio.api.streamservice.v1.request_response_pb2.CloseStreamRequest: + message = ( + temporalio.api.streamservice.v1.request_response_pb2.CloseStreamRequest() + ) + message.ref.CopyFrom( + _StreamRefTransferTypeConverter().to_transfer_type(value.ref) + ) + message.producer.CopyFrom( + _ProducerTransferTypeConverter().to_transfer_type(value.producer) + ) + return message + + +temporalio.converter.transfer_type_convertible( + _CloseStreamRequestTransferTypeConverter[typing.Any] +)(CloseStreamRequest) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class CloseStreamResponse: + pass + + +class _CloseStreamResponseTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + CloseStreamResponse, + temporalio.api.streamservice.v1.request_response_pb2.CloseStreamResponse, + ] +): + transfer_type: ( + type[temporalio.api.streamservice.v1.request_response_pb2.CloseStreamResponse] + | None + ) = temporalio.api.streamservice.v1.request_response_pb2.CloseStreamResponse + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.streamservice.v1.request_response_pb2.CloseStreamResponse, + type_hint: type[CloseStreamResponse], + ) -> CloseStreamResponse: + return CloseStreamResponse() + + @typing_extensions.override + def to_transfer_type( + self, + value: CloseStreamResponse, + ) -> temporalio.api.streamservice.v1.request_response_pb2.CloseStreamResponse: + message = ( + temporalio.api.streamservice.v1.request_response_pb2.CloseStreamResponse() + ) + return message + + +temporalio.converter.transfer_type_convertible( + _CloseStreamResponseTransferTypeConverter +)(CloseStreamResponse) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class ReadRecordsRequest(typing.Generic[BodyT]): + """Read Records after an offset.""" + + ref: StreamRef[BodyT] + from_offset: bytes | None = None + max_records: int | None = None + wait: datetime.timedelta | None = None + max_bytes: int | None = None + + +class _ReadRecordsRequestTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + ReadRecordsRequest[BodyT], + temporalio.api.streamservice.v1.request_response_pb2.ReadRecordsRequest, + ] +): + transfer_type: ( + type[temporalio.api.streamservice.v1.request_response_pb2.ReadRecordsRequest] + | None + ) = temporalio.api.streamservice.v1.request_response_pb2.ReadRecordsRequest + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.streamservice.v1.request_response_pb2.ReadRecordsRequest, + type_hint: type[ReadRecordsRequest[BodyT]], + ) -> ReadRecordsRequest[BodyT]: + (body_type,) = typing.get_args(type_hint) or (typing.Any,) + if not value.HasField("ref"): + raise ValueError("missing required field ReadRecordsRequest.ref") + ref = typing.cast( + StreamRef[BodyT], + _StreamRefTransferTypeConverter().from_transfer_type( + value.ref, StreamRef[body_type] + ), + ) + return ReadRecordsRequest( + ref=ref, + from_offset=value.from_offset if bool(value.from_offset) else None, + max_records=value.max_records if value.max_records != 0 else None, + wait=_support.duration_from_proto(value.wait) + if value.HasField("wait") + else None, + max_bytes=value.max_bytes if value.max_bytes != 0 else None, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: ReadRecordsRequest[BodyT], + ) -> temporalio.api.streamservice.v1.request_response_pb2.ReadRecordsRequest: + message = ( + temporalio.api.streamservice.v1.request_response_pb2.ReadRecordsRequest() + ) + message.ref.CopyFrom( + _StreamRefTransferTypeConverter().to_transfer_type(value.ref) + ) + if value.from_offset is not None: + message.from_offset = value.from_offset + if value.max_records is not None: + message.max_records = value.max_records + if value.wait is not None: + message.wait.CopyFrom(_support.duration_to_proto(value.wait)) + if value.max_bytes is not None: + message.max_bytes = value.max_bytes + return message + + +temporalio.converter.transfer_type_convertible( + _ReadRecordsRequestTransferTypeConverter[typing.Any] +)(ReadRecordsRequest) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True, kw_only=True) +class ReadRecordsResponse(typing.Generic[BodyT]): + records: list[StreamRecord[BodyT]] + last_offset: bytes | None = None + done: bool + + +class _ReadRecordsResponseTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + ReadRecordsResponse[BodyT], + temporalio.api.streamservice.v1.request_response_pb2.ReadRecordsResponse, + ] +): + transfer_type: ( + type[temporalio.api.streamservice.v1.request_response_pb2.ReadRecordsResponse] + | None + ) = temporalio.api.streamservice.v1.request_response_pb2.ReadRecordsResponse + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.streamservice.v1.request_response_pb2.ReadRecordsResponse, + type_hint: type[ReadRecordsResponse[BodyT]], + ) -> ReadRecordsResponse[BodyT]: + (body_type,) = typing.get_args(type_hint) or (typing.Any,) + records = [ + typing.cast( + StreamRecord[BodyT], + _StreamRecordTransferTypeConverter().from_transfer_type( + item, StreamRecord[body_type] + ), + ) + for item in value.records + ] + done = value.done + return ReadRecordsResponse( + records=records, + last_offset=value.last_offset if bool(value.last_offset) else None, + done=done, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: ReadRecordsResponse[BodyT], + ) -> temporalio.api.streamservice.v1.request_response_pb2.ReadRecordsResponse: + message = ( + temporalio.api.streamservice.v1.request_response_pb2.ReadRecordsResponse() + ) + for element in value.records: + item = message.records.add() + item.CopyFrom( + _StreamRecordTransferTypeConverter().to_transfer_type(element) + ) + if value.last_offset is not None: + message.last_offset = value.last_offset + message.done = value.done + return message + + +temporalio.converter.transfer_type_convertible( + _ReadRecordsResponseTransferTypeConverter[typing.Any] +)(ReadRecordsResponse) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True, kw_only=True) +class StreamRecord(typing.Generic[BodyT]): + """One Record that a Producer appended, as a read gives it.""" + + body: BodyT + producer: Producer | None = None + offset: bytes + + +class _StreamRecordTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + StreamRecord[BodyT], temporalio.api.stream.v1.message_pb2.StreamRecord + ] +): + transfer_type: type[temporalio.api.stream.v1.message_pb2.StreamRecord] | None = ( + temporalio.api.stream.v1.message_pb2.StreamRecord + ) + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.stream.v1.message_pb2.StreamRecord, + type_hint: type[StreamRecord[BodyT]], + ) -> StreamRecord[BodyT]: + (body_type,) = typing.get_args(type_hint) or (typing.Any,) + if not value.HasField("body"): + raise ValueError("missing required field StreamRecord.body") + body = typing.cast(BodyT, _support.payload_from_proto(value.body, body_type)) + if not value.offset: + raise ValueError("missing required field StreamRecord.offset") + offset = value.offset + return StreamRecord( + body=body, + producer=_ProducerTransferTypeConverter().from_transfer_type( + value.producer, Producer + ) + if value.HasField("producer") + else None, + offset=offset, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: StreamRecord[BodyT], + ) -> temporalio.api.stream.v1.message_pb2.StreamRecord: + message = temporalio.api.stream.v1.message_pb2.StreamRecord() + message.body.CopyFrom(_support.payload_to_proto(value.body)) + if value.producer is not None: + message.producer.CopyFrom( + _ProducerTransferTypeConverter().to_transfer_type(value.producer) + ) + message.offset = value.offset + return message + + +temporalio.converter.transfer_type_convertible( + _StreamRecordTransferTypeConverter[typing.Any] +)(StreamRecord) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class DescribeStreamRequest(typing.Generic[BodyT]): + ref: StreamRef[BodyT] + + +class _DescribeStreamRequestTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + DescribeStreamRequest[BodyT], + temporalio.api.streamservice.v1.request_response_pb2.DescribeStreamRequest, + ] +): + transfer_type: ( + type[temporalio.api.streamservice.v1.request_response_pb2.DescribeStreamRequest] + | None + ) = temporalio.api.streamservice.v1.request_response_pb2.DescribeStreamRequest + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.streamservice.v1.request_response_pb2.DescribeStreamRequest, + type_hint: type[DescribeStreamRequest[BodyT]], + ) -> DescribeStreamRequest[BodyT]: + (body_type,) = typing.get_args(type_hint) or (typing.Any,) + if not value.HasField("ref"): + raise ValueError("missing required field DescribeStreamRequest.ref") + ref = typing.cast( + StreamRef[BodyT], + _StreamRefTransferTypeConverter().from_transfer_type( + value.ref, StreamRef[body_type] + ), + ) + return DescribeStreamRequest( + ref=ref, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: DescribeStreamRequest[BodyT], + ) -> temporalio.api.streamservice.v1.request_response_pb2.DescribeStreamRequest: + message = ( + temporalio.api.streamservice.v1.request_response_pb2.DescribeStreamRequest() + ) + message.ref.CopyFrom( + _StreamRefTransferTypeConverter().to_transfer_type(value.ref) + ) + return message + + +temporalio.converter.transfer_type_convertible( + _DescribeStreamRequestTransferTypeConverter[typing.Any] +)(DescribeStreamRequest) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class DescribeStreamResponse: + status: StreamStatus | None = None + last_offset: bytes | None = None + + +class _DescribeStreamResponseTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + DescribeStreamResponse, + temporalio.api.streamservice.v1.request_response_pb2.DescribeStreamResponse, + ] +): + transfer_type: ( + type[ + temporalio.api.streamservice.v1.request_response_pb2.DescribeStreamResponse + ] + | None + ) = temporalio.api.streamservice.v1.request_response_pb2.DescribeStreamResponse + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.streamservice.v1.request_response_pb2.DescribeStreamResponse, + type_hint: type[DescribeStreamResponse], + ) -> DescribeStreamResponse: + return DescribeStreamResponse( + status=_stream_status_from_proto(value.status) + if value.status != 0 + else None, + last_offset=value.last_offset if bool(value.last_offset) else None, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: DescribeStreamResponse, + ) -> temporalio.api.streamservice.v1.request_response_pb2.DescribeStreamResponse: + message = temporalio.api.streamservice.v1.request_response_pb2.DescribeStreamResponse() + if value.status is not None: + message.status = temporalio.api.enums.v1.stream_pb2.StreamStatus.ValueType( + value.status + ) + if value.last_offset is not None: + message.last_offset = value.last_offset + return message + + +temporalio.converter.transfer_type_convertible( + _DescribeStreamResponseTransferTypeConverter +)(DescribeStreamResponse) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class ProducerKindWorkflow: + value: WorkflowProducer + + +@dataclasses.dataclass(slots=True) +class ProducerKindLocalActivity: + value: LocalActivityProducer + + +@dataclasses.dataclass(slots=True) +class ProducerKindWorkflowActivity: + value: ActivityProducer + + +@dataclasses.dataclass(slots=True) +class ProducerKindStandaloneActivity: + value: ActivityProducer + + +@dataclasses.dataclass(slots=True) +class ProducerKindClient: + value: ClientProducer + + +ProducerKind = ( + ProducerKindWorkflow + | ProducerKindLocalActivity + | ProducerKindWorkflowActivity + | ProducerKindStandaloneActivity + | ProducerKindClient +) +"""The kind of a Producer, with its run and epoch.""" diff --git a/temporalio/nexus/system/stream_service/services.py b/temporalio/nexus/system/stream_service/services.py new file mode 100644 index 0000000000..76231c5505 --- /dev/null +++ b/temporalio/nexus/system/stream_service/services.py @@ -0,0 +1,53 @@ +# Generated by nexgen v0.2.12. DO NOT EDIT! + +from __future__ import annotations + +import typing + +from nexusrpc import Operation, service + +from .models import ( + AppendRecordsRequest, + AppendRecordsResponse, + CloseStreamRequest, + CloseStreamResponse, + CreateStreamRequest, + CreateStreamResult, + DescribeStreamRequest, + DescribeStreamResponse, + ReadRecordsRequest, + ReadRecordsResponse, +) + + +@service(name="temporal.api.streamservice.v1.StreamService") +class StreamService: + """ + .. warning:: + This API is experimental and subject to change. + """ + + create_stream: Operation[ + CreateStreamRequest[typing.Any], + CreateStreamResult[typing.Any], + ] = Operation(name="CreateStream") + + append_records: Operation[ + AppendRecordsRequest[typing.Any], + AppendRecordsResponse, + ] = Operation(name="AppendRecords") + + close_stream: Operation[ + CloseStreamRequest[typing.Any], + CloseStreamResponse, + ] = Operation(name="CloseStream") + + read_records: Operation[ + ReadRecordsRequest[typing.Any], + ReadRecordsResponse[typing.Any], + ] = Operation(name="ReadRecords") + + describe_stream: Operation[ + DescribeStreamRequest[typing.Any], + DescribeStreamResponse, + ] = Operation(name="DescribeStream") diff --git a/temporalio/nexus/system/stream_service/workflow/__init__.py b/temporalio/nexus/system/stream_service/workflow/__init__.py new file mode 100644 index 0000000000..12b9baa97e --- /dev/null +++ b/temporalio/nexus/system/stream_service/workflow/__init__.py @@ -0,0 +1,77 @@ +# Generated by nexgen v0.2.12. DO NOT EDIT! + +from __future__ import annotations + +import collections.abc +import typing + +import nexusrpc + +import temporalio.converter + +from .. import services as _services +from ._resources.stream import get_stream +from .operations.create_stream import create_stream + +__all__ = [ + "create_stream", + "get_stream", +] + + +_InputT = typing.TypeVar("_InputT") +_OutputT = typing.TypeVar("_OutputT") + + +_SerializationContextFactory = collections.abc.Callable[ + [_InputT], temporalio.converter.SerializationContext +] + + +class _NexusOperationInfo(typing.Generic[_InputT, _OutputT]): + def __init__( + self, + *, + operation: nexusrpc.Operation[_InputT, _OutputT], + serialization_context: _SerializationContextFactory[_InputT] | None = None, + ) -> None: + self.operation: nexusrpc.Operation[_InputT, _OutputT] = operation + self.serialization_context: _SerializationContextFactory[_InputT] | None = ( + serialization_context + ) + + +__nexus_operation_registry__: dict[ + tuple[str, str], _NexusOperationInfo[typing.Any, typing.Any] +] = { + ( + "temporal.api.streamservice.v1.StreamService", + "CreateStream", + ): _NexusOperationInfo( + operation=_services.StreamService.create_stream, + ), + ( + "temporal.api.streamservice.v1.StreamService", + "AppendRecords", + ): _NexusOperationInfo( + operation=_services.StreamService.append_records, + ), + ( + "temporal.api.streamservice.v1.StreamService", + "CloseStream", + ): _NexusOperationInfo( + operation=_services.StreamService.close_stream, + ), + ( + "temporal.api.streamservice.v1.StreamService", + "ReadRecords", + ): _NexusOperationInfo( + operation=_services.StreamService.read_records, + ), + ( + "temporal.api.streamservice.v1.StreamService", + "DescribeStream", + ): _NexusOperationInfo( + operation=_services.StreamService.describe_stream, + ), +} diff --git a/temporalio/nexus/system/stream_service/workflow/_resources/__init__.py b/temporalio/nexus/system/stream_service/workflow/_resources/__init__.py new file mode 100644 index 0000000000..04fda4c2bc --- /dev/null +++ b/temporalio/nexus/system/stream_service/workflow/_resources/__init__.py @@ -0,0 +1,9 @@ +# Generated by nexgen v0.2.12. DO NOT EDIT! + +from __future__ import annotations + +from .stream import Stream + +__all__ = [ + "Stream", +] diff --git a/temporalio/nexus/system/stream_service/workflow/_resources/stream.py b/temporalio/nexus/system/stream_service/workflow/_resources/stream.py new file mode 100644 index 0000000000..4cf06445e5 --- /dev/null +++ b/temporalio/nexus/system/stream_service/workflow/_resources/stream.py @@ -0,0 +1,101 @@ +# Generated by nexgen v0.2.12. DO NOT EDIT! + +from __future__ import annotations + +import typing + +import temporalio.nexus.system +import temporalio.workflow + +from ...models import ( + DescribeStreamRequest, + DescribeStreamResponse, + StreamRef, +) + +BodyT = typing.TypeVar("BodyT") + + +class Stream(typing.Generic[BodyT]): + """One Stream. Build it from a stream-ref, or get it from a create.""" + + ref: StreamRef[BodyT] + created: bool | None + + def __init__( + self, + ref: StreamRef[BodyT], + created: bool | None, + ) -> None: + if type(self) is not Stream: + raise TypeError("Generic resource subclasses are not supported") + self.ref = ref + self.created = created + + @property + def body_type(self) -> type[BodyT]: + arguments = typing.get_args(getattr(self.ref, "__orig_class__", None)) + return typing.cast("type[BodyT]", arguments[0] if arguments else object) + + async def describe( + self, + ) -> DescribeStreamResponse: + """Describe the Stream's status and newest offset.""" + request = DescribeStreamRequest( + ref=self.ref, + ) + handle = await _describe_stream(request) + return await handle + + +@typing.overload +def get_stream( + ref: StreamRef[BodyT], + *, + created: bool | None = None, + body_type: None = None, +) -> Stream[BodyT]: ... + + +@typing.overload +def get_stream( + ref: StreamRef[BodyT] | StreamRef[object], + *, + created: bool | None = None, + body_type: type[BodyT], +) -> Stream[BodyT]: ... + + +def get_stream( + ref: StreamRef[typing.Any], + *, + created: bool | None = None, + body_type: typing.Any = None, +) -> Stream[typing.Any]: + """Return the ``Stream`` handle for this ref.""" + arguments = typing.get_args(getattr(ref, "__orig_class__", None)) + # None means no explicit type, so use the ref's type argument, else object. + if body_type is None: + body_type = arguments[0] if arguments else object + elif arguments and arguments[0] is not object and arguments[0] != body_type: + raise TypeError("The decoding type must match the typed ref") + if arguments != (body_type,): + ref = StreamRef[body_type](id=ref.id, store=ref.store) + return Stream[body_type]( + ref=ref, + created=created, + ) + + +async def _describe_stream( + request: DescribeStreamRequest[BodyT], +) -> temporalio.workflow.NexusOperationHandle[DescribeStreamResponse,]: + nexus_client = temporalio.workflow.create_nexus_client( + service="temporal.api.streamservice.v1.StreamService", + endpoint=temporalio.nexus.system.TEMPORAL_SYSTEM_ENDPOINT, + ) + return await nexus_client.start_operation( + operation="DescribeStream", + input=request, + output_type=DescribeStreamResponse, + ) diff --git a/temporalio/nexus/system/stream_service/workflow/_system_nexus_interceptor.py b/temporalio/nexus/system/stream_service/workflow/_system_nexus_interceptor.py new file mode 100644 index 0000000000..fa1196c89d --- /dev/null +++ b/temporalio/nexus/system/stream_service/workflow/_system_nexus_interceptor.py @@ -0,0 +1,299 @@ +# Generated by nexgen v0.2.12. DO NOT EDIT! + +from __future__ import annotations + +import abc +import typing + +from .. import models + +if typing.TYPE_CHECKING: + import temporalio.workflow + from temporalio.worker._interceptor import StartNexusOperationInput + + +__all__ = [ + "_start_system_nexus_operation", + "_SystemNexusWorkflowOutboundInterceptorBase", + "_SystemNexusWorkflowOutboundInterceptorTerminal", +] + + +_InputT = typing.TypeVar("_InputT") +_OutputT = typing.TypeVar("_OutputT") + + +async def _start_system_nexus_operation( + interceptor: _SystemNexusWorkflowOutboundInterceptorBase, + input: StartNexusOperationInput[_InputT, _OutputT], +) -> temporalio.workflow.NexusOperationHandle[_OutputT]: + if ( + input.service == "temporal.api.streamservice.v1.StreamService" + and input.operation_name == "CreateStream" + ): + typed_input = typing.cast( + "StartNexusOperationInput[models.CreateStreamRequest[typing.Any], models.CreateStreamResult[typing.Any]]", + input, + ) + # The dispatch check above establishes that this operation's response type is _OutputT. + return typing.cast( + "temporalio.workflow.NexusOperationHandle[_OutputT]", + await interceptor.start_create_stream( + typed_input.input, typed_input.output_type + ), + ) + if ( + input.service == "temporal.api.streamservice.v1.StreamService" + and input.operation_name == "AppendRecords" + ): + typed_input = typing.cast( + "StartNexusOperationInput[models.AppendRecordsRequest[typing.Any], models.AppendRecordsResponse]", + input, + ) + # The dispatch check above establishes that this operation's response type is _OutputT. + return typing.cast( + "temporalio.workflow.NexusOperationHandle[_OutputT]", + await interceptor.start_append_records(typed_input.input), + ) + if ( + input.service == "temporal.api.streamservice.v1.StreamService" + and input.operation_name == "CloseStream" + ): + typed_input = typing.cast( + "StartNexusOperationInput[models.CloseStreamRequest[typing.Any], models.CloseStreamResponse]", + input, + ) + # The dispatch check above establishes that this operation's response type is _OutputT. + return typing.cast( + "temporalio.workflow.NexusOperationHandle[_OutputT]", + await interceptor.start_close_stream(typed_input.input), + ) + if ( + input.service == "temporal.api.streamservice.v1.StreamService" + and input.operation_name == "ReadRecords" + ): + typed_input = typing.cast( + "StartNexusOperationInput[models.ReadRecordsRequest[typing.Any], models.ReadRecordsResponse[typing.Any]]", + input, + ) + # The dispatch check above establishes that this operation's response type is _OutputT. + return typing.cast( + "temporalio.workflow.NexusOperationHandle[_OutputT]", + await interceptor.start_read_records( + typed_input.input, typed_input.output_type + ), + ) + if ( + input.service == "temporal.api.streamservice.v1.StreamService" + and input.operation_name == "DescribeStream" + ): + typed_input = typing.cast( + "StartNexusOperationInput[models.DescribeStreamRequest[typing.Any], models.DescribeStreamResponse]", + input, + ) + # The dispatch check above establishes that this operation's response type is _OutputT. + return typing.cast( + "temporalio.workflow.NexusOperationHandle[_OutputT]", + await interceptor.start_describe_stream(typed_input.input), + ) + return await interceptor.start_system_nexus_operation(input) + + +class _SystemNexusWorkflowOutboundInterceptorBase(abc.ABC): + @abc.abstractmethod + def _next_system_nexus_interceptor( + self, + ) -> _SystemNexusWorkflowOutboundInterceptorBase: ... + + @abc.abstractmethod + async def start_system_nexus_operation( + self, + input: StartNexusOperationInput[_InputT, _OutputT], + ) -> temporalio.workflow.NexusOperationHandle[_OutputT]: + """Intercept a System Nexus operation through the generic hook.""" + ... + + async def start_create_stream( + self, + request: models.CreateStreamRequest[typing.Any], + output_type: type[models.CreateStreamResult[typing.Any]] | None = None, + ) -> temporalio.workflow.NexusOperationHandle[ + models.CreateStreamResult[typing.Any] + ]: + """Intercept the CreateStream operation.""" + return await self._next_system_nexus_interceptor().start_create_stream( + request, output_type + ) + + async def start_append_records( + self, request: models.AppendRecordsRequest[typing.Any] + ) -> temporalio.workflow.NexusOperationHandle[models.AppendRecordsResponse]: + """Intercept the AppendRecords operation.""" + return await self._next_system_nexus_interceptor().start_append_records(request) + + async def start_close_stream( + self, request: models.CloseStreamRequest[typing.Any] + ) -> temporalio.workflow.NexusOperationHandle[models.CloseStreamResponse]: + """Intercept the CloseStream operation.""" + return await self._next_system_nexus_interceptor().start_close_stream(request) + + async def start_read_records( + self, + request: models.ReadRecordsRequest[typing.Any], + output_type: type[models.ReadRecordsResponse[typing.Any]] | None = None, + ) -> temporalio.workflow.NexusOperationHandle[ + models.ReadRecordsResponse[typing.Any] + ]: + """Intercept the ReadRecords operation.""" + return await self._next_system_nexus_interceptor().start_read_records( + request, output_type + ) + + async def start_describe_stream( + self, request: models.DescribeStreamRequest[typing.Any] + ) -> temporalio.workflow.NexusOperationHandle[models.DescribeStreamResponse]: + """Intercept the DescribeStream operation.""" + return await self._next_system_nexus_interceptor().start_describe_stream( + request + ) + + +class _SystemNexusWorkflowOutboundInterceptorTerminal(abc.ABC): + @abc.abstractmethod + async def _intercept_system_nexus_operation( + self, + input: StartNexusOperationInput[_InputT, _OutputT], + ) -> temporalio.workflow.NexusOperationHandle[_OutputT]: ... + + async def start_create_stream( + self, + request: models.CreateStreamRequest[typing.Any], + output_type: type[models.CreateStreamResult[typing.Any]] | None = None, + ) -> temporalio.workflow.NexusOperationHandle[ + models.CreateStreamResult[typing.Any] + ]: + from temporalio.nexus.system import TEMPORAL_SYSTEM_ENDPOINT + from temporalio.worker._interceptor import StartNexusOperationInput + from temporalio.workflow import NexusOperationCancellationType + + return await self._intercept_system_nexus_operation( + StartNexusOperationInput( + endpoint=TEMPORAL_SYSTEM_ENDPOINT, + service="temporal.api.streamservice.v1.StreamService", + operation="CreateStream", + input=request, + output_type=typing.cast( + "type[models.CreateStreamResult[typing.Any]]", + models.CreateStreamResult, + ) + if output_type is None + else output_type, + schedule_to_close_timeout=None, + schedule_to_start_timeout=None, + start_to_close_timeout=None, + cancellation_type=NexusOperationCancellationType.WAIT_COMPLETED, + headers=None, + summary=None, + ) + ) + + async def start_append_records( + self, request: models.AppendRecordsRequest[typing.Any] + ) -> temporalio.workflow.NexusOperationHandle[models.AppendRecordsResponse]: + from temporalio.nexus.system import TEMPORAL_SYSTEM_ENDPOINT + from temporalio.worker._interceptor import StartNexusOperationInput + from temporalio.workflow import NexusOperationCancellationType + + return await self._intercept_system_nexus_operation( + StartNexusOperationInput( + endpoint=TEMPORAL_SYSTEM_ENDPOINT, + service="temporal.api.streamservice.v1.StreamService", + operation="AppendRecords", + input=request, + output_type=models.AppendRecordsResponse, + schedule_to_close_timeout=None, + schedule_to_start_timeout=None, + start_to_close_timeout=None, + cancellation_type=NexusOperationCancellationType.WAIT_COMPLETED, + headers=None, + summary=None, + ) + ) + + async def start_close_stream( + self, request: models.CloseStreamRequest[typing.Any] + ) -> temporalio.workflow.NexusOperationHandle[models.CloseStreamResponse]: + from temporalio.nexus.system import TEMPORAL_SYSTEM_ENDPOINT + from temporalio.worker._interceptor import StartNexusOperationInput + from temporalio.workflow import NexusOperationCancellationType + + return await self._intercept_system_nexus_operation( + StartNexusOperationInput( + endpoint=TEMPORAL_SYSTEM_ENDPOINT, + service="temporal.api.streamservice.v1.StreamService", + operation="CloseStream", + input=request, + output_type=models.CloseStreamResponse, + schedule_to_close_timeout=None, + schedule_to_start_timeout=None, + start_to_close_timeout=None, + cancellation_type=NexusOperationCancellationType.WAIT_COMPLETED, + headers=None, + summary=None, + ) + ) + + async def start_read_records( + self, + request: models.ReadRecordsRequest[typing.Any], + output_type: type[models.ReadRecordsResponse[typing.Any]] | None = None, + ) -> temporalio.workflow.NexusOperationHandle[ + models.ReadRecordsResponse[typing.Any] + ]: + from temporalio.nexus.system import TEMPORAL_SYSTEM_ENDPOINT + from temporalio.worker._interceptor import StartNexusOperationInput + from temporalio.workflow import NexusOperationCancellationType + + return await self._intercept_system_nexus_operation( + StartNexusOperationInput( + endpoint=TEMPORAL_SYSTEM_ENDPOINT, + service="temporal.api.streamservice.v1.StreamService", + operation="ReadRecords", + input=request, + output_type=typing.cast( + "type[models.ReadRecordsResponse[typing.Any]]", + models.ReadRecordsResponse, + ) + if output_type is None + else output_type, + schedule_to_close_timeout=None, + schedule_to_start_timeout=None, + start_to_close_timeout=None, + cancellation_type=NexusOperationCancellationType.WAIT_COMPLETED, + headers=None, + summary=None, + ) + ) + + async def start_describe_stream( + self, request: models.DescribeStreamRequest[typing.Any] + ) -> temporalio.workflow.NexusOperationHandle[models.DescribeStreamResponse]: + from temporalio.nexus.system import TEMPORAL_SYSTEM_ENDPOINT + from temporalio.worker._interceptor import StartNexusOperationInput + from temporalio.workflow import NexusOperationCancellationType + + return await self._intercept_system_nexus_operation( + StartNexusOperationInput( + endpoint=TEMPORAL_SYSTEM_ENDPOINT, + service="temporal.api.streamservice.v1.StreamService", + operation="DescribeStream", + input=request, + output_type=models.DescribeStreamResponse, + schedule_to_close_timeout=None, + schedule_to_start_timeout=None, + start_to_close_timeout=None, + cancellation_type=NexusOperationCancellationType.WAIT_COMPLETED, + headers=None, + summary=None, + ) + ) diff --git a/temporalio/nexus/system/stream_service/workflow/operations/__init__.py b/temporalio/nexus/system/stream_service/workflow/operations/__init__.py new file mode 100644 index 0000000000..250b6a42b7 --- /dev/null +++ b/temporalio/nexus/system/stream_service/workflow/operations/__init__.py @@ -0,0 +1,3 @@ +# Generated by nexgen v0.2.12. DO NOT EDIT! + +from __future__ import annotations diff --git a/temporalio/nexus/system/stream_service/workflow/operations/create_stream.py b/temporalio/nexus/system/stream_service/workflow/operations/create_stream.py new file mode 100644 index 0000000000..c06679d195 --- /dev/null +++ b/temporalio/nexus/system/stream_service/workflow/operations/create_stream.py @@ -0,0 +1,90 @@ +# Generated by nexgen v0.2.12. DO NOT EDIT! + +from __future__ import annotations + +import typing + +import temporalio.nexus.system +import temporalio.nexus.system._support as _support +import temporalio.workflow + +from ...models import ( + CreateStreamRequest, + CreateStreamResult, + StreamIdConflictPolicy, + StreamRef, +) +from .._resources import Stream + +BodyT = typing.TypeVar("BodyT") + + +async def _create_stream( + request: CreateStreamRequest[BodyT], + body_type: type[BodyT], +) -> Stream[BodyT]: + nexus_client = temporalio.workflow.create_nexus_client( + service="temporal.api.streamservice.v1.StreamService", + endpoint=temporalio.nexus.system.TEMPORAL_SYSTEM_ENDPOINT, + ) + handle = await nexus_client.start_operation( + operation="CreateStream", + input=request, + output_type=CreateStreamResult[body_type], + ) + result = await handle + if result.ref is None: + raise RuntimeError("CreateStream answered without `ref`") + return Stream[body_type]( + ref=result.ref, + created=result.created or None, + ) + + +@typing.overload +async def create_stream( + *, + id: str, + store: str | None = None, + id_conflict_policy: StreamIdConflictPolicy | None = None, + body_type: type[BodyT], +) -> Stream[BodyT]: ... + + +@typing.overload +async def create_stream( + *, + id: str, + store: str | None = None, + id_conflict_policy: StreamIdConflictPolicy | None = None, + body_type: None = None, +) -> Stream[object]: ... + + +async def create_stream( + *, + id: str, + store: str | None = None, + id_conflict_policy: StreamIdConflictPolicy | None = None, + body_type: typing.Any = None, +) -> Stream[typing.Any]: + """ + Args: + id: The Stream id, 1 to 4096 bytes. + store: Registered Store name. With none, the SDK picks one. Only an SDK reads + it. + id_conflict_policy: What the create does when the Stream exists. Unspecified is + fail. + """ + if body_type is None: + body_type = typing.Any + stream_ref = StreamRef( + id=id, + store=store, + ) + request = CreateStreamRequest( + stream_ref=stream_ref, + id_conflict_policy=id_conflict_policy, + request_id=_support.request_id(), + ) + return await _create_stream(request, body_type) diff --git a/temporalio/workflow/__init__.py b/temporalio/workflow/__init__.py index c631fd88e8..60b02cc543 100644 --- a/temporalio/workflow/__init__.py +++ b/temporalio/workflow/__init__.py @@ -3,6 +3,10 @@ from __future__ import annotations # BEGIN GENERATED NEXUS SYSTEM EXPORTS +from temporalio.nexus.system.stream_service.workflow import ( + create_stream, + get_stream, +) from temporalio.nexus.system.workflow_service import ( signal_with_start_workflow, ) @@ -329,6 +333,8 @@ "ReturnType", "SelfType", # BEGIN GENERATED NEXUS SYSTEM __ALL__ + "create_stream", + "get_stream", "signal_with_start_workflow", # END GENERATED NEXUS SYSTEM __ALL__ ] From 434ddd3e67e765be733a59fa423e5b9d6f82317c Mon Sep 17 00:00:00 2001 From: Moe Dashti Date: Fri, 13 Nov 2026 18:55:23 -0700 Subject: [PATCH 2/2] Regenerated the Streams API with append and close on the Workflow Stream. The new Stream WIT gives the Workflow target `append` and `close`, so the generated Workflow `Stream` has them and `async with` closes it. --- .../workflow/_resources/stream.py | 77 +++++++++++++++++++ .../workflow/operations/create_stream.py | 4 + 2 files changed, 81 insertions(+) diff --git a/temporalio/nexus/system/stream_service/workflow/_resources/stream.py b/temporalio/nexus/system/stream_service/workflow/_resources/stream.py index 4cf06445e5..df548d2528 100644 --- a/temporalio/nexus/system/stream_service/workflow/_resources/stream.py +++ b/temporalio/nexus/system/stream_service/workflow/_resources/stream.py @@ -2,14 +2,24 @@ from __future__ import annotations +import types import typing +import typing_extensions + import temporalio.nexus.system +import temporalio.nexus.system._support as _support import temporalio.workflow from ...models import ( + AppendRecord, + AppendRecordsRequest, + AppendRecordsResponse, + CloseStreamRequest, + CloseStreamResponse, DescribeStreamRequest, DescribeStreamResponse, + Producer, StreamRef, ) @@ -37,6 +47,45 @@ def body_type(self) -> type[BodyT]: arguments = typing.get_args(getattr(self.ref, "__orig_class__", None)) return typing.cast("type[BodyT]", arguments[0] if arguments else object) + async def __aenter__(self) -> typing_extensions.Self: + return self + + async def __aexit__( + self, + exc_type: type[BaseException] | None, + exc: BaseException | None, + tb: types.TracebackType | None, + ) -> None: + await self.close() + + async def append( + self, + *records: BodyT, + producer: Producer | None = None, + sequence: int | None = None, + ) -> AppendRecordsResponse: + """Append values, and return the offsets of the batch.""" + request = AppendRecordsRequest( + ref=self.ref, + producer=((_support.workflow_producer()) if producer is None else producer), + sequence=sequence, + records=[AppendRecord(body=value) for value in records], + ) + handle = await _append_records(request) + return await handle + + async def close( + self, + producer: Producer | None = None, + ) -> None: + """Close the Stream. Readers drain what is left, then end.""" + request = CloseStreamRequest( + ref=self.ref, + producer=((_support.workflow_producer()) if producer is None else producer), + ) + handle = await _close_stream(request) + await handle + async def describe( self, ) -> DescribeStreamResponse: @@ -87,6 +136,34 @@ def get_stream( ) +async def _append_records( + request: AppendRecordsRequest[BodyT], +) -> temporalio.workflow.NexusOperationHandle[AppendRecordsResponse,]: + nexus_client = temporalio.workflow.create_nexus_client( + service="temporal.api.streamservice.v1.StreamService", + endpoint=temporalio.nexus.system.TEMPORAL_SYSTEM_ENDPOINT, + ) + return await nexus_client.start_operation( + operation="AppendRecords", + input=request, + output_type=AppendRecordsResponse, + ) + + +async def _close_stream( + request: CloseStreamRequest[BodyT], +) -> temporalio.workflow.NexusOperationHandle[CloseStreamResponse,]: + nexus_client = temporalio.workflow.create_nexus_client( + service="temporal.api.streamservice.v1.StreamService", + endpoint=temporalio.nexus.system.TEMPORAL_SYSTEM_ENDPOINT, + ) + return await nexus_client.start_operation( + operation="CloseStream", + input=request, + output_type=CloseStreamResponse, + ) + + async def _describe_stream( request: DescribeStreamRequest[BodyT], ) -> temporalio.workflow.NexusOperationHandle[DescribeStreamResponse,]: diff --git a/temporalio/nexus/system/stream_service/workflow/operations/create_stream.py b/temporalio/nexus/system/stream_service/workflow/operations/create_stream.py index c06679d195..12e5f6b466 100644 --- a/temporalio/nexus/system/stream_service/workflow/operations/create_stream.py +++ b/temporalio/nexus/system/stream_service/workflow/operations/create_stream.py @@ -8,6 +8,7 @@ import temporalio.nexus.system._support as _support import temporalio.workflow +from ..._resource_context_manager import resource_context_manager from ...models import ( CreateStreamRequest, CreateStreamResult, @@ -42,6 +43,7 @@ async def _create_stream( @typing.overload +@resource_context_manager async def create_stream( *, id: str, @@ -52,6 +54,7 @@ async def create_stream( @typing.overload +@resource_context_manager async def create_stream( *, id: str, @@ -61,6 +64,7 @@ async def create_stream( ) -> Stream[object]: ... +@resource_context_manager async def create_stream( *, id: str,