From 7afb312c5b0ba60c20a14cf78d44f8063a36abe7 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Sun, 25 Oct 2026 14:39:45 -0700 Subject: [PATCH] Added the Stream service protos and their generated code. The Stream service messages come from the API repo, and the runtime messages from the `temporalio-streams` crate. `poe gen-protos-docker` copies the Streams output too. --- pyproject.toml | 8 + scripts/_proto/Dockerfile | 4 +- .../_proto/temporal/api/enums/v1/stream.proto | 88 +++ .../temporal/api/stream/v1/message.proto | 127 ++++ .../streamservice/v1/request_response.proto | 175 +++++ .../temporal/sdk/streams/v1/redis.proto | 28 + .../temporal/sdk/streams/v1/runtime.proto | 94 +++ scripts/gen_protos.py | 18 + scripts/gen_protos_docker.py | 3 + temporalio/api/enums/v1/__init__.py | 12 + temporalio/api/enums/v1/stream_pb2.py | 81 +++ temporalio/api/enums/v1/stream_pb2.pyi | 225 +++++++ temporalio/api/stream/__init__.py | 0 temporalio/api/stream/v1/__init__.py | 25 + temporalio/api/stream/v1/message_pb2.py | 177 +++++ temporalio/api/stream/v1/message_pb2.pyi | 454 +++++++++++++ temporalio/api/streamservice/__init__.py | 0 temporalio/api/streamservice/v1/__init__.py | 25 + .../streamservice/v1/request_response_pb2.py | 180 +++++ .../streamservice/v1/request_response_pb2.pyi | 619 ++++++++++++++++++ .../v1/request_response_pb2_grpc.py | 4 + .../v1/request_response_pb2_grpc.pyi | 4 + temporalio/common/_streams/__init__.py | 1 + temporalio/common/_streams/proto/__init__.py | 0 .../common/_streams/proto/v1/__init__.py | 29 + .../common/_streams/proto/v1/redis_pb2.py | 40 ++ .../common/_streams/proto/v1/redis_pb2.pyi | 106 +++ .../common/_streams/proto/v1/runtime_pb2.py | 184 ++++++ .../common/_streams/proto/v1/runtime_pb2.pyi | 335 ++++++++++ 29 files changed, 3044 insertions(+), 2 deletions(-) create mode 100644 scripts/_proto/temporal/api/enums/v1/stream.proto create mode 100644 scripts/_proto/temporal/api/stream/v1/message.proto create mode 100644 scripts/_proto/temporal/api/streamservice/v1/request_response.proto create mode 100644 scripts/_proto/temporal/sdk/streams/v1/redis.proto create mode 100644 scripts/_proto/temporal/sdk/streams/v1/runtime.proto 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/common/_streams/__init__.py create mode 100644 temporalio/common/_streams/proto/__init__.py create mode 100644 temporalio/common/_streams/proto/v1/__init__.py create mode 100644 temporalio/common/_streams/proto/v1/redis_pb2.py create mode 100644 temporalio/common/_streams/proto/v1/redis_pb2.pyi create mode 100644 temporalio/common/_streams/proto/v1/runtime_pb2.py create mode 100644 temporalio/common/_streams/proto/v1/runtime_pb2.pyi diff --git a/pyproject.toml b/pyproject.toml index f803404db6..1300699a27 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -183,12 +183,19 @@ exclude = [ 'temporalio/api', 'temporalio/bridge/proto', 'temporalio/nexus/system/workflow_service', + 'temporalio/common/_streams/proto', ] [[tool.mypy.overrides]] module = "temporalio.nexus.system.workflow_service.*" ignore_errors = true +[[tool.mypy.overrides]] +module = [ + "temporalio.common._streams.proto.*", +] +ignore_errors = true + [tool.pydocstyle] convention = "google" # https://github.com/PyCQA/pydocstyle/issues/363#issuecomment-625563088 @@ -258,6 +265,7 @@ exclude = [ "temporalio/api", "temporalio/bridge/proto", "temporalio/bridge/_visitor.py", + "temporalio/common/_streams/proto", "tests/worker/workflow_sandbox/testmodules/proto", ] diff --git a/scripts/_proto/Dockerfile b/scripts/_proto/Dockerfile index 0f0fb74b06..3e4aa6d879 100644 --- a/scripts/_proto/Dockerfile +++ b/scripts/_proto/Dockerfile @@ -3,7 +3,7 @@ FROM python:3.10 RUN curl -LsSf https://astral.sh/uv/install.sh | sh ENV PATH="$PATH:/root/.local/bin" RUN uv tool install poethepoet -VOLUME ["/api_new", "/bridge_new"] +VOLUME ["/api_new", "/bridge_new", "/streams_new"] COPY ./ ./ @@ -13,4 +13,4 @@ RUN uv add "protobuf<4" RUN uv sync --all-extras RUN uv run scripts/gen_protos.py -CMD ["sh", "-c", "cp -r ./temporalio/api/* /api_new && cp -r ./temporalio/bridge/proto/* /bridge_new"] +CMD ["sh", "-c", "cp -r ./temporalio/api/* /api_new && cp -r ./temporalio/bridge/proto/* /bridge_new && cp -r ./temporalio/common/_streams/proto/* /streams_new"] diff --git a/scripts/_proto/temporal/api/enums/v1/stream.proto b/scripts/_proto/temporal/api/enums/v1/stream.proto new file mode 100644 index 0000000000..683111f464 --- /dev/null +++ b/scripts/_proto/temporal/api/enums/v1/stream.proto @@ -0,0 +1,88 @@ +syntax = "proto3"; + +package temporal.api.enums.v1; + +option go_package = "go.temporal.io/api/enums/v1;enums"; +option java_package = "io.temporal.api.enums.v1"; +option java_multiple_files = true; +option java_outer_classname = "StreamProto"; +option ruby_package = "Temporalio::Api::Enums::V1"; +option csharp_namespace = "Temporalio.Api.Enums.V1"; + +// Experimental. Might change in the future. +enum StreamProducerKind { + STREAM_PRODUCER_KIND_UNSPECIFIED = 0; + STREAM_PRODUCER_KIND_WORKFLOW = 1; + STREAM_PRODUCER_KIND_LOCAL_ACTIVITY = 2; + STREAM_PRODUCER_KIND_ACTIVITY = 3; + STREAM_PRODUCER_KIND_CLIENT = 4; + STREAM_PRODUCER_KIND_STANDALONE_ACTIVITY = 5; +} + +// Experimental. Might change in the future. +// How a Stream closed, so a reader tells a terminated or timed out owner from a failed one +// without parsing the failure. +// (-- api-linter: core::0216::synonyms=disabled +// aip.dev/not-precedent: Named consistently with WorkflowExecutionStatus. --) +enum StreamCloseStatus { + STREAM_CLOSE_STATUS_UNSPECIFIED = 0; + STREAM_CLOSE_STATUS_COMPLETED = 1; + STREAM_CLOSE_STATUS_FAILED = 2; + STREAM_CLOSE_STATUS_CANCELED = 3; + // TERMINATED and TIMED_OUT come only from the owner's end, never from a caller. + STREAM_CLOSE_STATUS_TERMINATED = 4; + STREAM_CLOSE_STATUS_TIMED_OUT = 5; +} + +// Experimental. Might change in the future. +// (-- api-linter: core::0216::synonyms=disabled +// aip.dev/not-precedent: Named consistently with WorkflowExecutionStatus. --) +enum StreamStatus { + STREAM_STATUS_UNSPECIFIED = 0; + STREAM_STATUS_OPEN = 1; + STREAM_STATUS_CLOSED = 2; + STREAM_STATUS_EXPIRED = 3; +} + +// Experimental. Might change in the future. +enum StreamReadStart { + STREAM_READ_START_UNSPECIFIED = 0; + STREAM_READ_START_BEGINNING = 1; + STREAM_READ_START_END = 2; +} + +// Experimental. Might change in the future. +// Why a Stream call failed. Each reason has one gRPC status code. +enum StreamErrorReason { + STREAM_ERROR_REASON_UNSPECIFIED = 0; + // Nobody created the Stream, even after `create_wait`. NOT_FOUND. + STREAM_ERROR_REASON_NOT_FOUND = 1; + // The Stream's Retention or safety TTL passed. NOT_FOUND. + STREAM_ERROR_REASON_EXPIRED = 2; + // The Stream is closed. FAILED_PRECONDITION. + STREAM_ERROR_REASON_CLOSED = 3; + // The Producer's epoch is lower than the newest one, its Workflow Task was withdrawn, or a + // Workflow retry withdrew its run. FAILED_PRECONDITION. + STREAM_ERROR_REASON_FENCED = 4; + // The batch has a lower `sequence` than the last one of its epoch. FAILED_PRECONDITION. + STREAM_ERROR_REASON_SEQUENCE = 5; + // The batch has the `sequence` of the last one of its epoch, and other content. + // FAILED_PRECONDITION. + STREAM_ERROR_REASON_SEQ_DIVERGENT = 6; + // The Stream's owner doesn't allow this close or this Producer. FAILED_PRECONDITION. + STREAM_ERROR_REASON_NOT_OWNER = 7; + // A newer format version wrote the Stream. FAILED_PRECONDITION. + STREAM_ERROR_REASON_FORMAT_TOO_NEW = 8; + // The offset is below zero or past the newest offset + 1. INVALID_ARGUMENT. + STREAM_ERROR_REASON_OFFSET = 9; + // A Record or a batch is over the Store's limit. INVALID_ARGUMENT. + STREAM_ERROR_REASON_TOO_LARGE = 10; + // The request is malformed. INVALID_ARGUMENT. + STREAM_ERROR_REASON_INVALID = 11; + // The Store failed or did not answer. UNAVAILABLE. + STREAM_ERROR_REASON_STORAGE = 12; + // No writer has the id. The SDK's in-process writer only. NOT_FOUND. + STREAM_ERROR_REASON_UNKNOWN_WRITER = 13; + // The Store lacks a capability that the call needs. UNIMPLEMENTED. + STREAM_ERROR_REASON_UNSUPPORTED = 14; +} diff --git a/scripts/_proto/temporal/api/stream/v1/message.proto b/scripts/_proto/temporal/api/stream/v1/message.proto new file mode 100644 index 0000000000..fca6f9cc56 --- /dev/null +++ b/scripts/_proto/temporal/api/stream/v1/message.proto @@ -0,0 +1,127 @@ +syntax = "proto3"; + +package temporal.api.stream.v1; + +option go_package = "go.temporal.io/api/stream/v1;stream"; +option java_package = "io.temporal.api.stream.v1"; +option java_multiple_files = true; +option java_outer_classname = "MessageProto"; +option ruby_package = "Temporalio::Api::Stream::V1"; +option csharp_namespace = "Temporalio.Api.Stream.V1"; + +import "google/protobuf/timestamp.proto"; + +import "temporal/api/common/v1/message.proto"; +import "temporal/api/enums/v1/stream.proto"; +import "temporal/api/failure/v1/message.proto"; + +// Experimental. Might change in the future. +// Identifies one Stream. The id is opaque to callers. +message StreamRef { + string id = 1; + // Unset for a client's Stream. For a Workflow, business_id is the Workflow ID and run_id is + // the first execution run id of its run chain. For a Standalone Activity, business_id is the + // activity ID and run_id is its run id. A Store keys on id, and reads only a Workflow owner's + // business_id, to check that a close's ref names the owner's run chain. + temporal.api.common.v1.Execution owner = 2; + // The name of the backend that holds the Stream, as the Worker and client config register + // it, so every process routes a ref it receives. Empty is the default backend. The Store + // never reads it. + string backend = 3; +} + +// Experimental. Might change in the future. +// The ordered attempt of a Producer. Compared in this order: task_started_event_id, +// task_attempt, task_started_time, attempt_scheduled_time, attempt. A field that a kind does +// not use is empty and compares as zero. A timestamp compares seconds, then nanos. +message ProducerEpoch { + // The Workflow Task of a Workflow, or the one that dispatched a Local Activity. + int64 task_started_event_id = 1; + int32 task_attempt = 2; + // Tells a reused started event id apart, for example after a dropped speculative task. + google.protobuf.Timestamp task_started_time = 5; + // When the server scheduled an Activity's attempt. A reset makes it later. + google.protobuf.Timestamp attempt_scheduled_time = 3; + // Activity, Local Activity and client attempt. + int32 attempt = 4; +} + +// Experimental. Might change in the future. +message Producer { + temporal.api.enums.v1.StreamProducerKind kind = 1; + string id = 2; + ProducerEpoch epoch = 3; + // The run of the Producer, so nobody parses the id. Empty for a client. + string run_id = 4; +} + +// Experimental. Might change in the future. +message WorkflowTaskEpoch { + int64 started_event_id = 1; + int32 attempt = 2; + google.protobuf.Timestamp started_time = 3; +} + +// Experimental. Might change in the future. +message AppendRecord { + reserved 1; + reserved "topic"; + temporal.api.common.v1.Payload body = 2; +} + +// Experimental. Might change in the future. +message Supersession { + temporal.api.enums.v1.StreamProducerKind producer_kind = 1; + string producer_id = 2; + ProducerEpoch epoch = 3; + // Workflow only. + WorkflowTaskEpoch previous_completed_task = 4; + repeated int64 failed_tasks = 5; + // Workflow retry only. The previous run. Every Record whose Producer carries it is withdrawn. + string withdraw_run = 6; + // The run of the Producer, so nobody parses the id. Empty for a client. + string run_id = 7; +} + +// Experimental. Might change in the future. +// One Record that a Producer appended. +message StreamData { + reserved 1; + reserved "topic"; + Producer producer = 2; + temporal.api.common.v1.Payload body = 3; +} + +// Experimental. Might change in the future. +// The close of a Stream. A Stream has at most one, and a read returns it last. +message StreamClose { + // The Producer that closed the Stream. + Producer producer = 1; + temporal.api.common.v1.Payload result = 2; + temporal.api.failure.v1.Failure failure = 3; + // The result or the failure stays next to it. + temporal.api.enums.v1.StreamCloseStatus status = 4; +} + +// Experimental. Might change in the future. +// An offset is the Redis entry id packed as ms << 20 | seq. It is ordered and unique in one +// Stream, and it is sparse. Offsets have gaps, so they don't count Records. Zero is never an +// offset. A read names its Stream, so the read checks the offset against that Stream. +message StreamRecord { + int64 offset = 1; + // A read item is a Record, a Superseded item or the close. + oneof item { + StreamData data = 2; + // The first batch of a higher epoch carries it. It withdraws the earlier Records that its + // facts name, and a reader drops them. + Supersession superseded = 3; + StreamClose close = 4; + } +} + +// 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. +message StreamErrorDetails { + temporal.api.enums.v1.StreamErrorReason reason = 1; +} diff --git a/scripts/_proto/temporal/api/streamservice/v1/request_response.proto b/scripts/_proto/temporal/api/streamservice/v1/request_response.proto new file mode 100644 index 0000000000..e6abc2e5ed --- /dev/null +++ b/scripts/_proto/temporal/api/streamservice/v1/request_response.proto @@ -0,0 +1,175 @@ +syntax = "proto3"; + +package temporal.api.streamservice.v1; + +option go_package = "go.temporal.io/api/streamservice/v1;streamservice"; +option java_package = "io.temporal.api.streamservice.v1"; +option java_multiple_files = true; +option java_outer_classname = "RequestResponseProto"; +option ruby_package = "Temporalio::Api::StreamService::V1"; +option csharp_namespace = "Temporalio.Api.StreamService.V1"; + +// The messages of the Stream service. An SDK Store (Redis or memory) and the server answer the +// same five operations with them. A field that only the server reads says so. + +import "google/protobuf/duration.proto"; + +import "temporal/api/common/v1/message.proto"; +import "temporal/api/enums/v1/stream.proto"; +import "temporal/api/failure/v1/message.proto"; +import "temporal/api/stream/v1/message.proto"; + +// Experimental. Might change in the future. +// Creates the Stream, or gets it when it exists. Then created is false, and the Stream keeps its +// own Retention. +// (-- api-linter: core::0133=disabled +// aip.dev/not-precedent: The request names a Stream by namespace and stream_id. --) +message CreateStreamRequest { + string namespace = 1; + // A Store requires it. Outside Workflow code, the server makes an id when it is empty. + string stream_id = 2; + // The age after which a Record is dropped. Empty is the Store default. + google.protobuf.Duration retention = 3; + // Workflow only. The previous run on a retry run's first create, to withdraw its Records. + string withdraw_run = 4; + // The Workflow run that owns the Stream. Workflow kind, with its run_id and epoch. + temporal.api.stream.v1.Producer owner = 6; + // Only the server reads it. The server closes the Stream with the outcome of the owner's run + // chain, terminate and timeout included. It needs owner, and a ref with a Workflow owner. + // (-- api-linter: core::0140::prepositions=disabled + // aip.dev/not-precedent: "with" ties the close to the owner's run chain. --) + bool close_with_run = 7; + // The ref of the Stream. A Workflow-owned Stream sets the ref's owner, which names the + // owner's run chain. + temporal.api.stream.v1.StreamRef ref = 8; + // Only the server reads it. The server checks that the token is the Activity's current + // attempt, so a caller can't put a Stream into a Standalone Activity it isn't. + bytes task_token = 9; +} + +// Experimental. Might change in the future. +message CreateStreamResponse { + temporal.api.stream.v1.StreamRef stream = 1; + bool created = 2; + temporal.api.enums.v1.StreamStatus status = 3; +} + +// Experimental. Might change in the future. +// FAILED_PRECONDITION if closed, fenced, or if the Stream's meta has a higher format version v +// (detail stream_format_too_new). NOT_FOUND if absent. +message AppendRecordsRequest { + string namespace = 1; + temporal.api.stream.v1.StreamRef stream = 2; + // Required outside a Workflow. In a Workflow the transport sets it. + temporal.api.stream.v1.Producer producer = 3; + // The batch number in this epoch. With the Producer and epoch it is the dedupe key. + int64 sequence = 4; + repeated temporal.api.stream.v1.AppendRecord records = 5; + // The run's last completed Workflow Task (started_event_id 0 before the first) and its failed + // tasks, from a Local Activity's start. The batch withdraws the run's Local Activity Records + // whose task never completed. + temporal.api.stream.v1.WorkflowTaskEpoch previous_completed_task = 6; + repeated int64 failed_tasks = 7; + // Local Activity only. The previous run on a retry run's first batch, to withdraw its Records. + string withdraw_run = 8; + // Only the server reads it, on a Workflow Producer's batch. The batch withdraws no Workflow + // Task and keeps no task facts, but withdraw_run still applies. + bool defer_withdrawal = 9; + // Only the server reads it. The server checks the Producer against the Activity's current + // attempt, so nobody else writes as that Activity. + bytes task_token = 10; +} + +// Experimental. Might change in the future. +message AppendRecordsResponse { + int64 first_offset = 1; + int64 last_offset = 2; +} + +// Experimental. Might change in the future. +message CloseStreamRequest { + string namespace = 1; + temporal.api.stream.v1.StreamRef stream = 2; + temporal.api.stream.v1.Producer producer = 3; + temporal.api.common.v1.Payload result = 4; + temporal.api.failure.v1.Failure failure = 5; + // Only the server reads these. They are the run's task facts, as on the append. + temporal.api.stream.v1.WorkflowTaskEpoch previous_completed_task = 6; + repeated int64 failed_tasks = 7; + // Only the server reads it. The previous run on a retry run's close, to withdraw its Records. + string withdraw_run = 8; + // Empty is COMPLETED with no failure and FAILED with one. A caller sets CANCELED itself. + temporal.api.enums.v1.StreamCloseStatus status = 9; + // Only the server reads it. It checks the token as on the append. + bytes task_token = 10; +} + +// Experimental. Might change in the future. +message CloseStreamResponse { +} + +// Experimental. Might change in the future. +message ReadRecordsRequest { + string namespace = 1; + temporal.api.stream.v1.StreamRef stream = 2; + // Start at this offset. Zero uses start. INVALID_ARGUMENT if past the newest offset + 1. + // (-- api-linter: core::0140::prepositions=disabled + // aip.dev/not-precedent: "from" and "to" mark the ends of a range of offsets. --) + int64 from_offset = 3; + temporal.api.enums.v1.StreamReadStart start = 4; + reserved 5, 10; + reserved "topics", "topic_prefixes"; + int32 max_records = 6; + google.protobuf.Duration wait = 7; + // How long to wait for a Stream that nobody created yet. Empty fails at once with + // NOT_FOUND. At most 60 s. + google.protobuf.Duration create_wait = 8; + // The exclusive end of the read. The read scans no entry at or after it, the close Record + // included, so a replay reads the same range. Zero means no end. Only the server serves it, + // and a Store refuses it. + // (-- api-linter: core::0140::prepositions=disabled + // aip.dev/not-precedent: "from" and "to" mark the ends of a range of offsets. --) + int64 to_offset = 11; + // Park as a normal read, then answer with no Records and next_offset = from_offset, when a + // Record at or after from_offset exists, when the Stream closes (done), or when wait ends. Its + // wait may be days, since a parked read adds nothing to History while it waits. Only the + // server serves it, and a Store refuses it. + bool wake_only = 12; + // Read the newest `last` Records, oldest first. It needs no from_offset and no start. Not with + // wake_only, and not in Workflow code, where replay needs a fixed range. + int32 last = 13; + // 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. + int64 max_bytes = 14; + // The lowest offset this run read on replay. A reset run copies the earlier run's reads into + // its History, so its floor must cover them too, not only its first live page. Only the + // server reads it. + // (-- api-linter: core::0140::prepositions=disabled + // aip.dev/not-precedent: The floor is the lowest from_offset of the run's reads. --) + int64 floor_from = 15; +} + +// Experimental. Might change in the future. +message ReadRecordsResponse { + repeated temporal.api.stream.v1.StreamRecord records = 1; + // The from_offset of the next read, which is the newest offset read + 1. + int64 next_offset = 2; + bool done = 3; + // This reader's floor keeps more than half of the Stream's byte cap. The writer fails when the + // cap fills, and only the reader's continue-as-new frees it. Only the server sets it. + bool continue_as_new_suggested = 4; +} + +// Experimental. Might change in the future. +message DescribeStreamRequest { + string namespace = 1; + temporal.api.stream.v1.StreamRef stream = 2; +} + +// Experimental. Might change in the future. +message DescribeStreamResponse { + temporal.api.enums.v1.StreamStatus status = 1; + // Zero when the Stream has no Record. + int64 last_offset = 2; + google.protobuf.Duration retention = 3; +} diff --git a/scripts/_proto/temporal/sdk/streams/v1/redis.proto b/scripts/_proto/temporal/sdk/streams/v1/redis.proto new file mode 100644 index 0000000000..d00dc52d32 --- /dev/null +++ b/scripts/_proto/temporal/sdk/streams/v1/redis.proto @@ -0,0 +1,28 @@ +syntax = "proto3"; + +// The config of the Redis Stream runtime. It keeps the package of the runtime protos of +// `temporalio-streams`, so the message keeps its full name. Field numbers are part of the contract +// with every SDK, and the memory runtime's config shares the numbers of the fields both have. +package temporal.sdk.streams.v1; + +import "google/protobuf/duration.proto"; + +// The settings of the Redis Stream runtime. Lang builds it from the backend options. +message RedisStreamConfig { + // One URL, or the seed URLs of a cluster. Credentials and TLS come from the URL. + repeated string urls = 1; + bool cluster = 2; + // Empty is `temporal-streams`. + string key_prefix = 3; + // The default Retention of a new Stream. Empty is 7 days. + google.protobuf.Duration retention = 4; + // How long an open Workflow-owned Stream stays after its last write. Empty is 90 days. + google.protobuf.Duration safety_ttl = 5; + // Zero is 256. + int32 blocking_reads_per_node = 8; + // Empty is 5 s. + google.protobuf.Duration response_timeout = 9; + // The backend name that lang registered this runtime under. The crate sets it on each + // StreamRef that it builds, such as the one in a create response. + string backend = 10; +} diff --git a/scripts/_proto/temporal/sdk/streams/v1/runtime.proto b/scripts/_proto/temporal/sdk/streams/v1/runtime.proto new file mode 100644 index 0000000000..7c178cf093 --- /dev/null +++ b/scripts/_proto/temporal/sdk/streams/v1/runtime.proto @@ -0,0 +1,94 @@ +syntax = "proto3"; + +// The messages of the crate's own calls, in addition to the five Stream service operations. Lang +// sends them through the bridge's call on the runtime object, by the method name in each comment. +// Field numbers are part of the contract with every SDK. +package temporal.sdk.streams.v1; + +import "google/protobuf/duration.proto"; +import "temporal/api/stream/v1/message.proto"; + +// The settings of the memory Stream runtime, for tests. Each field has the number of the same +// field in the config of the other Stores, such as Redis, so a backend can send that config here +// too. +message MemoryStreamConfig { + // The default Retention of a new Stream. Empty is 7 days. + google.protobuf.Duration retention = 4; + // How long an open Workflow-owned Stream stays after its last write. Empty is 90 days. + google.protobuf.Duration safety_ttl = 5; + // The backend name that lang registered this runtime under. The crate sets it on each + // StreamRef that it builds, such as the one in a create response. + string backend = 10; +} + +// Method `OpenWriter`, asynchronous. A writer for an Activity or a client Producer, which keeps its +// epoch and `sequence` order. A writer of an Activity attempt above 1 first sends an empty batch, +// so that the new epoch withdraws the earlier attempt's Records. The handles of one Producer epoch +// on one Stream share one writer. Each open adds a handle, and each `CloseWriter` or `FreeWriter` +// releases one. The batches of every writer in a process take their `sequence` from one counter, so +// a writer that opens again after its close continues above its earlier batches. +message OpenWriterRequest { + string namespace = 1; + temporal.api.stream.v1.StreamRef stream = 2; + temporal.api.stream.v1.Producer producer = 3; + // A write sends at most this many Records in one batch. Zero is 1000. + int32 max_batch_size = 5; + // How long a write retries a Store failure. Empty is 10 s. + google.protobuf.Duration max_retry_duration = 6; + // Local Activity only. The run's task facts, which each batch sends as AppendRecords does, so + // the writes withdraw the run's orphaned Local Activity Records. + temporal.api.stream.v1.WorkflowTaskEpoch previous_completed_task = 8; + repeated int64 failed_tasks = 9; +} + +message OpenWriterResponse { + uint64 writer_id = 1; +} + +// Method `WriteRecords`, asynchronous. Appends Records at once for an open writer and responds when +// the Store has them. It uses the writer's Producer epoch and `sequence` order. After a Store +// failure it resends the same batch, so that the Store writes it once. Records past the Store's +// batch limit go in more batches, in order. An awaited publish of an Activity or a client uses this +// call. A failure after the retries fails the writer, since the batch may be in the Store. +message WriteRecordsRequest { + uint64 writer_id = 1; + repeated temporal.api.stream.v1.AppendRecord records = 2; +} + +message WriteRecordsResponse { + // The offsets of the first and the last Record. Zero when the request has no Record. + int64 first_offset = 1; + int64 last_offset = 2; +} + +// Method `CloseWriter`, asynchronous. Releases one handle of the writer. The last handle waits for +// a write in flight, then drops the writer. It doesn't close the Stream. It fails with the +// writer's error if a write failed. +message CloseWriterRequest { + uint64 writer_id = 1; +} + +message CloseWriterResponse { + // The offset of the newest Record the writer wrote. Zero if it wrote none. + int64 last_offset = 1; +} + +// Method `FreeWriter`. Releases one handle of the writer with no wait, for a handle that lang +// drops, such as from a finalizer. The bridge may also call it with no async runtime. A write in +// flight still completes. The writer's error, if any, is lost. +message FreeWriterRequest { + uint64 writer_id = 1; +} + +message FreeWriterResponse { +} + +// Method `CloseWriters`, asynchronous. Drops every open writer, for example when the backend's +// client or Worker shuts down. +message CloseWritersRequest { +} + +message CloseWritersResponse { + // The writers that had failed. + int32 failed = 1; +} diff --git a/scripts/gen_protos.py b/scripts/gen_protos.py index 4b2ea04569..efbe9d26f8 100644 --- a/scripts/gen_protos.py +++ b/scripts/gen_protos.py @@ -33,6 +33,7 @@ api_out_dir = base_dir / "temporalio" / "api" sdk_out_dir = base_dir / "temporalio" / "bridge" / "proto" +streams_out_dir = base_dir / "temporalio" / "common" / "_streams" / "proto" py_fixes = [ partial(re.compile(r"from temporal\.api\.").sub, r"from temporalio.api."), @@ -50,6 +51,14 @@ partial( re.compile(r"from temporal\.sdk\.core\.").sub, r"from temporalio.bridge.proto." ), + partial( + re.compile(r"from temporal\.sdk\.streams\.").sub, + r"from temporalio.common._streams.proto.", + ), + partial( + re.compile(r"'__module__' : 'temporal\.sdk\.streams\.").sub, + r"'__module__' : 'temporalio.common._streams.proto.", + ), partial( re.compile(r"'__module__' : 'temporal\.api\.").sub, r"'__module__' : 'temporalio.api.", @@ -71,6 +80,10 @@ r"temporalio.api.dependencies.nexusannotations.", ), partial(re.compile(r"temporal\.sdk\.core\.").sub, r"temporalio.bridge.proto."), + partial( + re.compile(r"temporal\.sdk\.streams\.").sub, + r"temporalio.common._streams.proto.", + ), ] find_class_re = re.compile(r"\nclass ([^_\(\:]+)") @@ -218,6 +231,11 @@ def generate_protos(output_dir: Path): for p in (output_dir / "temporal" / "sdk" / "core").iterdir(): shutil.rmtree(sdk_out_dir / p.name, ignore_errors=True) p.replace(sdk_out_dir / p.name) + # Only the Streams package imports the Stream runtime protos + streams_out_dir.mkdir(parents=True, exist_ok=True) + for p in (output_dir / "temporal" / "sdk" / "streams").iterdir(): + shutil.rmtree(streams_out_dir / p.name, ignore_errors=True) + p.replace(streams_out_dir / p.name) shutil.rmtree(sdk_out_dir / "health", ignore_errors=True) (output_dir / "health").replace(sdk_out_dir / "health") # Move test protos diff --git a/scripts/gen_protos_docker.py b/scripts/gen_protos_docker.py index 6735bdf90d..004d02de51 100644 --- a/scripts/gen_protos_docker.py +++ b/scripts/gen_protos_docker.py @@ -34,6 +34,9 @@ os.path.join(os.getcwd(), "temporalio", "api") + ":/api_new", "-v", os.path.join(os.getcwd(), "temporalio", "bridge", "proto") + ":/bridge_new", + "-v", + os.path.join(os.getcwd(), "temporalio", "common", "_streams", "proto") + + ":/streams_new", image_id, ] ) diff --git a/temporalio/api/enums/v1/__init__.py b/temporalio/api/enums/v1/__init__.py index 518a2f32bc..b57b02dc74 100644 --- a/temporalio/api/enums/v1/__init__.py +++ b/temporalio/api/enums/v1/__init__.py @@ -44,6 +44,13 @@ from .query_pb2 import QueryRejectCondition, QueryResultType from .reset_pb2 import ResetReapplyExcludeType, ResetReapplyType, ResetType from .schedule_pb2 import ScheduleOverlapPolicy +from .stream_pb2 import ( + StreamCloseStatus, + StreamErrorReason, + StreamProducerKind, + StreamReadStart, + StreamStatus, +) from .task_queue_pb2 import ( BuildIdTaskReachability, DescribeTaskQueueMode, @@ -120,6 +127,11 @@ "Severity", "SignalExternalWorkflowExecutionFailedCause", "StartChildWorkflowExecutionFailedCause", + "StreamCloseStatus", + "StreamErrorReason", + "StreamProducerKind", + "StreamReadStart", + "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..d0cf202107 --- /dev/null +++ b/temporalio/api/enums/v1/stream_pb2.py @@ -0,0 +1,81 @@ +# -*- 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 message as _message +from google.protobuf import reflection as _reflection +from google.protobuf import symbol_database as _symbol_database +from google.protobuf.internal import enum_type_wrapper + +# @@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*\xf8\x01\n\x12StreamProducerKind\x12$\n STREAM_PRODUCER_KIND_UNSPECIFIED\x10\x00\x12!\n\x1dSTREAM_PRODUCER_KIND_WORKFLOW\x10\x01\x12\'\n#STREAM_PRODUCER_KIND_LOCAL_ACTIVITY\x10\x02\x12!\n\x1dSTREAM_PRODUCER_KIND_ACTIVITY\x10\x03\x12\x1f\n\x1bSTREAM_PRODUCER_KIND_CLIENT\x10\x04\x12,\n(STREAM_PRODUCER_KIND_STANDALONE_ACTIVITY\x10\x05*\xe4\x01\n\x11StreamCloseStatus\x12#\n\x1fSTREAM_CLOSE_STATUS_UNSPECIFIED\x10\x00\x12!\n\x1dSTREAM_CLOSE_STATUS_COMPLETED\x10\x01\x12\x1e\n\x1aSTREAM_CLOSE_STATUS_FAILED\x10\x02\x12 \n\x1cSTREAM_CLOSE_STATUS_CANCELED\x10\x03\x12"\n\x1eSTREAM_CLOSE_STATUS_TERMINATED\x10\x04\x12!\n\x1dSTREAM_CLOSE_STATUS_TIMED_OUT\x10\x05*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*p\n\x0fStreamReadStart\x12!\n\x1dSTREAM_READ_START_UNSPECIFIED\x10\x00\x12\x1f\n\x1bSTREAM_READ_START_BEGINNING\x10\x01\x12\x19\n\x15STREAM_READ_START_END\x10\x02*\xa2\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\x1cSTREAM_ERROR_REASON_SEQUENCE\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\x1e\n\x1aSTREAM_ERROR_REASON_OFFSET\x10\t\x12!\n\x1dSTREAM_ERROR_REASON_TOO_LARGE\x10\n\x12\x1f\n\x1bSTREAM_ERROR_REASON_INVALID\x10\x0b\x12\x1f\n\x1bSTREAM_ERROR_REASON_STORAGE\x10\x0c\x12&\n"STREAM_ERROR_REASON_UNKNOWN_WRITER\x10\r\x12#\n\x1fSTREAM_ERROR_REASON_UNSUPPORTED\x10\x0e\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' +) + +_STREAMPRODUCERKIND = DESCRIPTOR.enum_types_by_name["StreamProducerKind"] +StreamProducerKind = enum_type_wrapper.EnumTypeWrapper(_STREAMPRODUCERKIND) +_STREAMCLOSESTATUS = DESCRIPTOR.enum_types_by_name["StreamCloseStatus"] +StreamCloseStatus = enum_type_wrapper.EnumTypeWrapper(_STREAMCLOSESTATUS) +_STREAMSTATUS = DESCRIPTOR.enum_types_by_name["StreamStatus"] +StreamStatus = enum_type_wrapper.EnumTypeWrapper(_STREAMSTATUS) +_STREAMREADSTART = DESCRIPTOR.enum_types_by_name["StreamReadStart"] +StreamReadStart = enum_type_wrapper.EnumTypeWrapper(_STREAMREADSTART) +_STREAMERRORREASON = DESCRIPTOR.enum_types_by_name["StreamErrorReason"] +StreamErrorReason = enum_type_wrapper.EnumTypeWrapper(_STREAMERRORREASON) +STREAM_PRODUCER_KIND_UNSPECIFIED = 0 +STREAM_PRODUCER_KIND_WORKFLOW = 1 +STREAM_PRODUCER_KIND_LOCAL_ACTIVITY = 2 +STREAM_PRODUCER_KIND_ACTIVITY = 3 +STREAM_PRODUCER_KIND_CLIENT = 4 +STREAM_PRODUCER_KIND_STANDALONE_ACTIVITY = 5 +STREAM_CLOSE_STATUS_UNSPECIFIED = 0 +STREAM_CLOSE_STATUS_COMPLETED = 1 +STREAM_CLOSE_STATUS_FAILED = 2 +STREAM_CLOSE_STATUS_CANCELED = 3 +STREAM_CLOSE_STATUS_TERMINATED = 4 +STREAM_CLOSE_STATUS_TIMED_OUT = 5 +STREAM_STATUS_UNSPECIFIED = 0 +STREAM_STATUS_OPEN = 1 +STREAM_STATUS_CLOSED = 2 +STREAM_STATUS_EXPIRED = 3 +STREAM_READ_START_UNSPECIFIED = 0 +STREAM_READ_START_BEGINNING = 1 +STREAM_READ_START_END = 2 +STREAM_ERROR_REASON_UNSPECIFIED = 0 +STREAM_ERROR_REASON_NOT_FOUND = 1 +STREAM_ERROR_REASON_EXPIRED = 2 +STREAM_ERROR_REASON_CLOSED = 3 +STREAM_ERROR_REASON_FENCED = 4 +STREAM_ERROR_REASON_SEQUENCE = 5 +STREAM_ERROR_REASON_SEQ_DIVERGENT = 6 +STREAM_ERROR_REASON_NOT_OWNER = 7 +STREAM_ERROR_REASON_FORMAT_TOO_NEW = 8 +STREAM_ERROR_REASON_OFFSET = 9 +STREAM_ERROR_REASON_TOO_LARGE = 10 +STREAM_ERROR_REASON_INVALID = 11 +STREAM_ERROR_REASON_STORAGE = 12 +STREAM_ERROR_REASON_UNKNOWN_WRITER = 13 +STREAM_ERROR_REASON_UNSUPPORTED = 14 + + +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" + _STREAMPRODUCERKIND._serialized_start = 62 + _STREAMPRODUCERKIND._serialized_end = 310 + _STREAMCLOSESTATUS._serialized_start = 313 + _STREAMCLOSESTATUS._serialized_end = 541 + _STREAMSTATUS._serialized_start = 543 + _STREAMSTATUS._serialized_end = 665 + _STREAMREADSTART._serialized_start = 667 + _STREAMREADSTART._serialized_end = 779 + _STREAMERRORREASON._serialized_start = 782 + _STREAMERRORREASON._serialized_end = 1328 +# @@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..0394472fc0 --- /dev/null +++ b/temporalio/api/enums/v1/stream_pb2.pyi @@ -0,0 +1,225 @@ +""" +@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 _StreamProducerKind: + ValueType = typing.NewType("ValueType", builtins.int) + V: typing_extensions.TypeAlias = ValueType + +class _StreamProducerKindEnumTypeWrapper( + google.protobuf.internal.enum_type_wrapper._EnumTypeWrapper[ + _StreamProducerKind.ValueType + ], + builtins.type, +): # noqa: F821 + DESCRIPTOR: google.protobuf.descriptor.EnumDescriptor + STREAM_PRODUCER_KIND_UNSPECIFIED: _StreamProducerKind.ValueType # 0 + STREAM_PRODUCER_KIND_WORKFLOW: _StreamProducerKind.ValueType # 1 + STREAM_PRODUCER_KIND_LOCAL_ACTIVITY: _StreamProducerKind.ValueType # 2 + STREAM_PRODUCER_KIND_ACTIVITY: _StreamProducerKind.ValueType # 3 + STREAM_PRODUCER_KIND_CLIENT: _StreamProducerKind.ValueType # 4 + STREAM_PRODUCER_KIND_STANDALONE_ACTIVITY: _StreamProducerKind.ValueType # 5 + +class StreamProducerKind( + _StreamProducerKind, metaclass=_StreamProducerKindEnumTypeWrapper +): + """Experimental. Might change in the future.""" + +STREAM_PRODUCER_KIND_UNSPECIFIED: StreamProducerKind.ValueType # 0 +STREAM_PRODUCER_KIND_WORKFLOW: StreamProducerKind.ValueType # 1 +STREAM_PRODUCER_KIND_LOCAL_ACTIVITY: StreamProducerKind.ValueType # 2 +STREAM_PRODUCER_KIND_ACTIVITY: StreamProducerKind.ValueType # 3 +STREAM_PRODUCER_KIND_CLIENT: StreamProducerKind.ValueType # 4 +STREAM_PRODUCER_KIND_STANDALONE_ACTIVITY: StreamProducerKind.ValueType # 5 +global___StreamProducerKind = StreamProducerKind + +class _StreamCloseStatus: + ValueType = typing.NewType("ValueType", builtins.int) + V: typing_extensions.TypeAlias = ValueType + +class _StreamCloseStatusEnumTypeWrapper( + google.protobuf.internal.enum_type_wrapper._EnumTypeWrapper[ + _StreamCloseStatus.ValueType + ], + builtins.type, +): # noqa: F821 + DESCRIPTOR: google.protobuf.descriptor.EnumDescriptor + STREAM_CLOSE_STATUS_UNSPECIFIED: _StreamCloseStatus.ValueType # 0 + STREAM_CLOSE_STATUS_COMPLETED: _StreamCloseStatus.ValueType # 1 + STREAM_CLOSE_STATUS_FAILED: _StreamCloseStatus.ValueType # 2 + STREAM_CLOSE_STATUS_CANCELED: _StreamCloseStatus.ValueType # 3 + STREAM_CLOSE_STATUS_TERMINATED: _StreamCloseStatus.ValueType # 4 + """TERMINATED and TIMED_OUT come only from the owner's end, never from a caller.""" + STREAM_CLOSE_STATUS_TIMED_OUT: _StreamCloseStatus.ValueType # 5 + +class StreamCloseStatus( + _StreamCloseStatus, metaclass=_StreamCloseStatusEnumTypeWrapper +): + """Experimental. Might change in the future. + How a Stream closed, so a reader tells a terminated or timed out owner from a failed one + without parsing the failure. + (-- api-linter: core::0216::synonyms=disabled + aip.dev/not-precedent: Named consistently with WorkflowExecutionStatus. --) + """ + +STREAM_CLOSE_STATUS_UNSPECIFIED: StreamCloseStatus.ValueType # 0 +STREAM_CLOSE_STATUS_COMPLETED: StreamCloseStatus.ValueType # 1 +STREAM_CLOSE_STATUS_FAILED: StreamCloseStatus.ValueType # 2 +STREAM_CLOSE_STATUS_CANCELED: StreamCloseStatus.ValueType # 3 +STREAM_CLOSE_STATUS_TERMINATED: StreamCloseStatus.ValueType # 4 +"""TERMINATED and TIMED_OUT come only from the owner's end, never from a caller.""" +STREAM_CLOSE_STATUS_TIMED_OUT: StreamCloseStatus.ValueType # 5 +global___StreamCloseStatus = StreamCloseStatus + +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, +): # noqa: F821 + 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. + (-- 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 _StreamReadStart: + ValueType = typing.NewType("ValueType", builtins.int) + V: typing_extensions.TypeAlias = ValueType + +class _StreamReadStartEnumTypeWrapper( + google.protobuf.internal.enum_type_wrapper._EnumTypeWrapper[ + _StreamReadStart.ValueType + ], + builtins.type, +): # noqa: F821 + DESCRIPTOR: google.protobuf.descriptor.EnumDescriptor + STREAM_READ_START_UNSPECIFIED: _StreamReadStart.ValueType # 0 + STREAM_READ_START_BEGINNING: _StreamReadStart.ValueType # 1 + STREAM_READ_START_END: _StreamReadStart.ValueType # 2 + +class StreamReadStart(_StreamReadStart, metaclass=_StreamReadStartEnumTypeWrapper): + """Experimental. Might change in the future.""" + +STREAM_READ_START_UNSPECIFIED: StreamReadStart.ValueType # 0 +STREAM_READ_START_BEGINNING: StreamReadStart.ValueType # 1 +STREAM_READ_START_END: StreamReadStart.ValueType # 2 +global___StreamReadStart = StreamReadStart + +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, +): # noqa: F821 + DESCRIPTOR: google.protobuf.descriptor.EnumDescriptor + STREAM_ERROR_REASON_UNSPECIFIED: _StreamErrorReason.ValueType # 0 + STREAM_ERROR_REASON_NOT_FOUND: _StreamErrorReason.ValueType # 1 + """Nobody created the Stream, even after `create_wait`. NOT_FOUND.""" + STREAM_ERROR_REASON_EXPIRED: _StreamErrorReason.ValueType # 2 + """The Stream's Retention or safety 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, its Workflow Task was withdrawn, or a + Workflow retry withdrew its run. FAILED_PRECONDITION. + """ + STREAM_ERROR_REASON_SEQUENCE: _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 Stream's owner doesn't allow this close or this Producer. FAILED_PRECONDITION.""" + STREAM_ERROR_REASON_FORMAT_TOO_NEW: _StreamErrorReason.ValueType # 8 + """A newer format version wrote the Stream. FAILED_PRECONDITION.""" + STREAM_ERROR_REASON_OFFSET: _StreamErrorReason.ValueType # 9 + """The offset is below zero or past the newest offset + 1. INVALID_ARGUMENT.""" + 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: _StreamErrorReason.ValueType # 12 + """The Store failed or did not answer. UNAVAILABLE.""" + STREAM_ERROR_REASON_UNKNOWN_WRITER: _StreamErrorReason.ValueType # 13 + """No writer has the id. The SDK's in-process writer only. NOT_FOUND.""" + STREAM_ERROR_REASON_UNSUPPORTED: _StreamErrorReason.ValueType # 14 + """The Store lacks a capability that the call needs. UNIMPLEMENTED.""" + +class StreamErrorReason( + _StreamErrorReason, metaclass=_StreamErrorReasonEnumTypeWrapper +): + """Experimental. Might change in the future. + Why a Stream call failed. Each reason has one gRPC status code. + """ + +STREAM_ERROR_REASON_UNSPECIFIED: StreamErrorReason.ValueType # 0 +STREAM_ERROR_REASON_NOT_FOUND: StreamErrorReason.ValueType # 1 +"""Nobody created the Stream, even after `create_wait`. NOT_FOUND.""" +STREAM_ERROR_REASON_EXPIRED: StreamErrorReason.ValueType # 2 +"""The Stream's Retention or safety 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, its Workflow Task was withdrawn, or a +Workflow retry withdrew its run. FAILED_PRECONDITION. +""" +STREAM_ERROR_REASON_SEQUENCE: 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 Stream's owner doesn't allow this close or this Producer. FAILED_PRECONDITION.""" +STREAM_ERROR_REASON_FORMAT_TOO_NEW: StreamErrorReason.ValueType # 8 +"""A newer format version wrote the Stream. FAILED_PRECONDITION.""" +STREAM_ERROR_REASON_OFFSET: StreamErrorReason.ValueType # 9 +"""The offset is below zero or past the newest offset + 1. INVALID_ARGUMENT.""" +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: StreamErrorReason.ValueType # 12 +"""The Store failed or did not answer. UNAVAILABLE.""" +STREAM_ERROR_REASON_UNKNOWN_WRITER: StreamErrorReason.ValueType # 13 +"""No writer has the id. The SDK's in-process writer only. NOT_FOUND.""" +STREAM_ERROR_REASON_UNSUPPORTED: StreamErrorReason.ValueType # 14 +"""The Store lacks a capability that the call needs. UNIMPLEMENTED.""" +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..a02871c3ca --- /dev/null +++ b/temporalio/api/stream/v1/__init__.py @@ -0,0 +1,25 @@ +from .message_pb2 import ( + AppendRecord, + Producer, + ProducerEpoch, + StreamClose, + StreamData, + StreamErrorDetails, + StreamRecord, + StreamRef, + Supersession, + WorkflowTaskEpoch, +) + +__all__ = [ + "AppendRecord", + "Producer", + "ProducerEpoch", + "StreamClose", + "StreamData", + "StreamErrorDetails", + "StreamRecord", + "StreamRef", + "Supersession", + "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..236c50a70b --- /dev/null +++ b/temporalio/api/stream/v1/message_pb2.py @@ -0,0 +1,177 @@ +# -*- 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 message as _message +from google.protobuf import reflection as _reflection +from google.protobuf import symbol_database as _symbol_database + +# @@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, +) +from temporalio.api.failure.v1 import ( + message_pb2 as temporal_dot_api_dot_failure_dot_v1_dot_message__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\x1a%temporal/api/failure/v1/message.proto"Z\n\tStreamRef\x12\n\n\x02id\x18\x01 \x01(\t\x12\x30\n\x05owner\x18\x02 \x01(\x0b\x32!.temporal.api.common.v1.Execution\x12\x0f\n\x07\x62\x61\x63kend\x18\x03 \x01(\t"\xc8\x01\n\rProducerEpoch\x12\x1d\n\x15task_started_event_id\x18\x01 \x01(\x03\x12\x14\n\x0ctask_attempt\x18\x02 \x01(\x05\x12\x35\n\x11task_started_time\x18\x05 \x01(\x0b\x32\x1a.google.protobuf.Timestamp\x12:\n\x16\x61ttempt_scheduled_time\x18\x03 \x01(\x0b\x32\x1a.google.protobuf.Timestamp\x12\x0f\n\x07\x61ttempt\x18\x04 \x01(\x05"\x95\x01\n\x08Producer\x12\x37\n\x04kind\x18\x01 \x01(\x0e\x32).temporal.api.enums.v1.StreamProducerKind\x12\n\n\x02id\x18\x02 \x01(\t\x12\x34\n\x05\x65poch\x18\x03 \x01(\x0b\x32%.temporal.api.stream.v1.ProducerEpoch\x12\x0e\n\x06run_id\x18\x04 \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"J\n\x0c\x41ppendRecord\x12-\n\x04\x62ody\x18\x02 \x01(\x0b\x32\x1f.temporal.api.common.v1.PayloadJ\x04\x08\x01\x10\x02R\x05topic"\xa3\x02\n\x0cSupersession\x12@\n\rproducer_kind\x18\x01 \x01(\x0e\x32).temporal.api.enums.v1.StreamProducerKind\x12\x13\n\x0bproducer_id\x18\x02 \x01(\t\x12\x34\n\x05\x65poch\x18\x03 \x01(\x0b\x32%.temporal.api.stream.v1.ProducerEpoch\x12J\n\x17previous_completed_task\x18\x04 \x01(\x0b\x32).temporal.api.stream.v1.WorkflowTaskEpoch\x12\x14\n\x0c\x66\x61iled_tasks\x18\x05 \x03(\x03\x12\x14\n\x0cwithdraw_run\x18\x06 \x01(\t\x12\x0e\n\x06run_id\x18\x07 \x01(\t"|\n\nStreamData\x12\x32\n\x08producer\x18\x02 \x01(\x0b\x32 .temporal.api.stream.v1.Producer\x12-\n\x04\x62ody\x18\x03 \x01(\x0b\x32\x1f.temporal.api.common.v1.PayloadJ\x04\x08\x01\x10\x02R\x05topic"\xdf\x01\n\x0bStreamClose\x12\x32\n\x08producer\x18\x01 \x01(\x0b\x32 .temporal.api.stream.v1.Producer\x12/\n\x06result\x18\x02 \x01(\x0b\x32\x1f.temporal.api.common.v1.Payload\x12\x31\n\x07\x66\x61ilure\x18\x03 \x01(\x0b\x32 .temporal.api.failure.v1.Failure\x12\x38\n\x06status\x18\x04 \x01(\x0e\x32(.temporal.api.enums.v1.StreamCloseStatus"\xcc\x01\n\x0cStreamRecord\x12\x0e\n\x06offset\x18\x01 \x01(\x03\x12\x32\n\x04\x64\x61ta\x18\x02 \x01(\x0b\x32".temporal.api.stream.v1.StreamDataH\x00\x12:\n\nsuperseded\x18\x03 \x01(\x0b\x32$.temporal.api.stream.v1.SupersessionH\x00\x12\x34\n\x05\x63lose\x18\x04 \x01(\x0b\x32#.temporal.api.stream.v1.StreamCloseH\x00\x42\x06\n\x04item"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' +) + + +_STREAMREF = DESCRIPTOR.message_types_by_name["StreamRef"] +_PRODUCEREPOCH = DESCRIPTOR.message_types_by_name["ProducerEpoch"] +_PRODUCER = DESCRIPTOR.message_types_by_name["Producer"] +_WORKFLOWTASKEPOCH = DESCRIPTOR.message_types_by_name["WorkflowTaskEpoch"] +_APPENDRECORD = DESCRIPTOR.message_types_by_name["AppendRecord"] +_SUPERSESSION = DESCRIPTOR.message_types_by_name["Supersession"] +_STREAMDATA = DESCRIPTOR.message_types_by_name["StreamData"] +_STREAMCLOSE = DESCRIPTOR.message_types_by_name["StreamClose"] +_STREAMRECORD = DESCRIPTOR.message_types_by_name["StreamRecord"] +_STREAMERRORDETAILS = DESCRIPTOR.message_types_by_name["StreamErrorDetails"] +StreamRef = _reflection.GeneratedProtocolMessageType( + "StreamRef", + (_message.Message,), + { + "DESCRIPTOR": _STREAMREF, + "__module__": "temporalio.api.stream.v1.message_pb2", + # @@protoc_insertion_point(class_scope:temporal.api.stream.v1.StreamRef) + }, +) +_sym_db.RegisterMessage(StreamRef) + +ProducerEpoch = _reflection.GeneratedProtocolMessageType( + "ProducerEpoch", + (_message.Message,), + { + "DESCRIPTOR": _PRODUCEREPOCH, + "__module__": "temporalio.api.stream.v1.message_pb2", + # @@protoc_insertion_point(class_scope:temporal.api.stream.v1.ProducerEpoch) + }, +) +_sym_db.RegisterMessage(ProducerEpoch) + +Producer = _reflection.GeneratedProtocolMessageType( + "Producer", + (_message.Message,), + { + "DESCRIPTOR": _PRODUCER, + "__module__": "temporalio.api.stream.v1.message_pb2", + # @@protoc_insertion_point(class_scope:temporal.api.stream.v1.Producer) + }, +) +_sym_db.RegisterMessage(Producer) + +WorkflowTaskEpoch = _reflection.GeneratedProtocolMessageType( + "WorkflowTaskEpoch", + (_message.Message,), + { + "DESCRIPTOR": _WORKFLOWTASKEPOCH, + "__module__": "temporalio.api.stream.v1.message_pb2", + # @@protoc_insertion_point(class_scope:temporal.api.stream.v1.WorkflowTaskEpoch) + }, +) +_sym_db.RegisterMessage(WorkflowTaskEpoch) + +AppendRecord = _reflection.GeneratedProtocolMessageType( + "AppendRecord", + (_message.Message,), + { + "DESCRIPTOR": _APPENDRECORD, + "__module__": "temporalio.api.stream.v1.message_pb2", + # @@protoc_insertion_point(class_scope:temporal.api.stream.v1.AppendRecord) + }, +) +_sym_db.RegisterMessage(AppendRecord) + +Supersession = _reflection.GeneratedProtocolMessageType( + "Supersession", + (_message.Message,), + { + "DESCRIPTOR": _SUPERSESSION, + "__module__": "temporalio.api.stream.v1.message_pb2", + # @@protoc_insertion_point(class_scope:temporal.api.stream.v1.Supersession) + }, +) +_sym_db.RegisterMessage(Supersession) + +StreamData = _reflection.GeneratedProtocolMessageType( + "StreamData", + (_message.Message,), + { + "DESCRIPTOR": _STREAMDATA, + "__module__": "temporalio.api.stream.v1.message_pb2", + # @@protoc_insertion_point(class_scope:temporal.api.stream.v1.StreamData) + }, +) +_sym_db.RegisterMessage(StreamData) + +StreamClose = _reflection.GeneratedProtocolMessageType( + "StreamClose", + (_message.Message,), + { + "DESCRIPTOR": _STREAMCLOSE, + "__module__": "temporalio.api.stream.v1.message_pb2", + # @@protoc_insertion_point(class_scope:temporal.api.stream.v1.StreamClose) + }, +) +_sym_db.RegisterMessage(StreamClose) + +StreamRecord = _reflection.GeneratedProtocolMessageType( + "StreamRecord", + (_message.Message,), + { + "DESCRIPTOR": _STREAMRECORD, + "__module__": "temporalio.api.stream.v1.message_pb2", + # @@protoc_insertion_point(class_scope:temporal.api.stream.v1.StreamRecord) + }, +) +_sym_db.RegisterMessage(StreamRecord) + +StreamErrorDetails = _reflection.GeneratedProtocolMessageType( + "StreamErrorDetails", + (_message.Message,), + { + "DESCRIPTOR": _STREAMERRORDETAILS, + "__module__": "temporalio.api.stream.v1.message_pb2", + # @@protoc_insertion_point(class_scope:temporal.api.stream.v1.StreamErrorDetails) + }, +) +_sym_db.RegisterMessage(StreamErrorDetails) + +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 = 210 + _STREAMREF._serialized_end = 300 + _PRODUCEREPOCH._serialized_start = 303 + _PRODUCEREPOCH._serialized_end = 503 + _PRODUCER._serialized_start = 506 + _PRODUCER._serialized_end = 655 + _WORKFLOWTASKEPOCH._serialized_start = 657 + _WORKFLOWTASKEPOCH._serialized_end = 769 + _APPENDRECORD._serialized_start = 771 + _APPENDRECORD._serialized_end = 845 + _SUPERSESSION._serialized_start = 848 + _SUPERSESSION._serialized_end = 1139 + _STREAMDATA._serialized_start = 1141 + _STREAMDATA._serialized_end = 1265 + _STREAMCLOSE._serialized_start = 1268 + _STREAMCLOSE._serialized_end = 1491 + _STREAMRECORD._serialized_start = 1494 + _STREAMRECORD._serialized_end = 1698 + _STREAMERRORDETAILS._serialized_start = 1700 + _STREAMERRORDETAILS._serialized_end = 1778 +# @@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..2a8c030366 --- /dev/null +++ b/temporalio/api/stream/v1/message_pb2.pyi @@ -0,0 +1,454 @@ +""" +@generated by mypy-protobuf. Do not edit manually! +isort:skip_file +""" + +import builtins +import collections.abc +import sys + +import google.protobuf.descriptor +import google.protobuf.internal.containers +import google.protobuf.message +import google.protobuf.timestamp_pb2 + +import temporalio.api.common.v1.message_pb2 +import temporalio.api.enums.v1.stream_pb2 +import temporalio.api.failure.v1.message_pb2 + +if sys.version_info >= (3, 8): + import typing as typing_extensions +else: + import typing_extensions + +DESCRIPTOR: google.protobuf.descriptor.FileDescriptor + +class StreamRef(google.protobuf.message.Message): + """Experimental. Might change in the future. + Identifies one Stream. The id is opaque to callers. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + ID_FIELD_NUMBER: builtins.int + OWNER_FIELD_NUMBER: builtins.int + BACKEND_FIELD_NUMBER: builtins.int + id: builtins.str + @property + def owner(self) -> temporalio.api.common.v1.message_pb2.Execution: + """Unset for a client's Stream. For a Workflow, business_id is the Workflow ID and run_id is + the first execution run id of its run chain. For a Standalone Activity, business_id is the + activity ID and run_id is its run id. A Store keys on id, and reads only a Workflow owner's + business_id, to check that a close's ref names the owner's run chain. + """ + backend: builtins.str + """The name of the backend that holds the Stream, as the Worker and client config register + it, so every process routes a ref it receives. Empty is the default backend. The Store + never reads it. + """ + def __init__( + self, + *, + id: builtins.str = ..., + owner: temporalio.api.common.v1.message_pb2.Execution | None = ..., + backend: builtins.str = ..., + ) -> None: ... + def HasField( + self, field_name: typing_extensions.Literal["owner", b"owner"] + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "backend", b"backend", "id", b"id", "owner", b"owner" + ], + ) -> None: ... + +global___StreamRef = StreamRef + +class ProducerEpoch(google.protobuf.message.Message): + """Experimental. Might change in the future. + The ordered attempt of a Producer. Compared in this order: task_started_event_id, + task_attempt, task_started_time, attempt_scheduled_time, attempt. A field that a kind does + not use is empty and compares as zero. A timestamp compares seconds, then nanos. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + TASK_STARTED_EVENT_ID_FIELD_NUMBER: builtins.int + TASK_ATTEMPT_FIELD_NUMBER: builtins.int + TASK_STARTED_TIME_FIELD_NUMBER: builtins.int + ATTEMPT_SCHEDULED_TIME_FIELD_NUMBER: builtins.int + ATTEMPT_FIELD_NUMBER: builtins.int + task_started_event_id: builtins.int + """The Workflow Task of a Workflow, or the one that dispatched a Local Activity.""" + task_attempt: builtins.int + @property + def task_started_time(self) -> google.protobuf.timestamp_pb2.Timestamp: + """Tells a reused started event id apart, for example after a dropped speculative task.""" + @property + def attempt_scheduled_time(self) -> google.protobuf.timestamp_pb2.Timestamp: + """When the server scheduled an Activity's attempt. A reset makes it later.""" + attempt: builtins.int + """Activity, Local Activity and client attempt.""" + def __init__( + self, + *, + task_started_event_id: builtins.int = ..., + task_attempt: builtins.int = ..., + task_started_time: google.protobuf.timestamp_pb2.Timestamp | None = ..., + attempt_scheduled_time: google.protobuf.timestamp_pb2.Timestamp | None = ..., + attempt: builtins.int = ..., + ) -> None: ... + def HasField( + self, + field_name: typing_extensions.Literal[ + "attempt_scheduled_time", + b"attempt_scheduled_time", + "task_started_time", + b"task_started_time", + ], + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "attempt", + b"attempt", + "attempt_scheduled_time", + b"attempt_scheduled_time", + "task_attempt", + b"task_attempt", + "task_started_event_id", + b"task_started_event_id", + "task_started_time", + b"task_started_time", + ], + ) -> None: ... + +global___ProducerEpoch = ProducerEpoch + +class Producer(google.protobuf.message.Message): + """Experimental. Might change in the future.""" + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + KIND_FIELD_NUMBER: builtins.int + ID_FIELD_NUMBER: builtins.int + EPOCH_FIELD_NUMBER: builtins.int + RUN_ID_FIELD_NUMBER: builtins.int + kind: temporalio.api.enums.v1.stream_pb2.StreamProducerKind.ValueType + id: builtins.str + @property + def epoch(self) -> global___ProducerEpoch: ... + run_id: builtins.str + """The run of the Producer, so nobody parses the id. Empty for a client.""" + def __init__( + self, + *, + kind: temporalio.api.enums.v1.stream_pb2.StreamProducerKind.ValueType = ..., + id: builtins.str = ..., + epoch: global___ProducerEpoch | None = ..., + run_id: builtins.str = ..., + ) -> None: ... + def HasField( + self, field_name: typing_extensions.Literal["epoch", b"epoch"] + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "epoch", b"epoch", "id", b"id", "kind", b"kind", "run_id", b"run_id" + ], + ) -> None: ... + +global___Producer = Producer + +class WorkflowTaskEpoch(google.protobuf.message.Message): + """Experimental. Might change in the future.""" + + 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 + attempt: builtins.int + @property + def started_time(self) -> google.protobuf.timestamp_pb2.Timestamp: ... + 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_extensions.Literal["started_time", b"started_time"] + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "attempt", + b"attempt", + "started_event_id", + b"started_event_id", + "started_time", + b"started_time", + ], + ) -> None: ... + +global___WorkflowTaskEpoch = WorkflowTaskEpoch + +class AppendRecord(google.protobuf.message.Message): + """Experimental. Might change in the future.""" + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + BODY_FIELD_NUMBER: builtins.int + @property + def body(self) -> temporalio.api.common.v1.message_pb2.Payload: ... + def __init__( + self, + *, + body: temporalio.api.common.v1.message_pb2.Payload | None = ..., + ) -> None: ... + def HasField( + self, field_name: typing_extensions.Literal["body", b"body"] + ) -> builtins.bool: ... + def ClearField( + self, field_name: typing_extensions.Literal["body", b"body"] + ) -> None: ... + +global___AppendRecord = AppendRecord + +class Supersession(google.protobuf.message.Message): + """Experimental. Might change in the future.""" + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + PRODUCER_KIND_FIELD_NUMBER: builtins.int + PRODUCER_ID_FIELD_NUMBER: builtins.int + EPOCH_FIELD_NUMBER: builtins.int + PREVIOUS_COMPLETED_TASK_FIELD_NUMBER: builtins.int + FAILED_TASKS_FIELD_NUMBER: builtins.int + WITHDRAW_RUN_FIELD_NUMBER: builtins.int + RUN_ID_FIELD_NUMBER: builtins.int + producer_kind: temporalio.api.enums.v1.stream_pb2.StreamProducerKind.ValueType + producer_id: builtins.str + @property + def epoch(self) -> global___ProducerEpoch: ... + @property + def previous_completed_task(self) -> global___WorkflowTaskEpoch: + """Workflow only.""" + @property + def failed_tasks( + self, + ) -> google.protobuf.internal.containers.RepeatedScalarFieldContainer[ + builtins.int + ]: ... + withdraw_run: builtins.str + """Workflow retry only. The previous run. Every Record whose Producer carries it is withdrawn.""" + run_id: builtins.str + """The run of the Producer, so nobody parses the id. Empty for a client.""" + def __init__( + self, + *, + producer_kind: temporalio.api.enums.v1.stream_pb2.StreamProducerKind.ValueType = ..., + producer_id: builtins.str = ..., + epoch: global___ProducerEpoch | None = ..., + previous_completed_task: global___WorkflowTaskEpoch | None = ..., + failed_tasks: collections.abc.Iterable[builtins.int] | None = ..., + withdraw_run: builtins.str = ..., + run_id: builtins.str = ..., + ) -> None: ... + def HasField( + self, + field_name: typing_extensions.Literal[ + "epoch", b"epoch", "previous_completed_task", b"previous_completed_task" + ], + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "epoch", + b"epoch", + "failed_tasks", + b"failed_tasks", + "previous_completed_task", + b"previous_completed_task", + "producer_id", + b"producer_id", + "producer_kind", + b"producer_kind", + "run_id", + b"run_id", + "withdraw_run", + b"withdraw_run", + ], + ) -> None: ... + +global___Supersession = Supersession + +class StreamData(google.protobuf.message.Message): + """Experimental. Might change in the future. + One Record that a Producer appended. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + PRODUCER_FIELD_NUMBER: builtins.int + BODY_FIELD_NUMBER: builtins.int + @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 = ..., + ) -> None: ... + def HasField( + self, + field_name: typing_extensions.Literal["body", b"body", "producer", b"producer"], + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal["body", b"body", "producer", b"producer"], + ) -> None: ... + +global___StreamData = StreamData + +class StreamClose(google.protobuf.message.Message): + """Experimental. Might change in the future. + The close of a Stream. A Stream has at most one, and a read returns it last. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + PRODUCER_FIELD_NUMBER: builtins.int + RESULT_FIELD_NUMBER: builtins.int + FAILURE_FIELD_NUMBER: builtins.int + STATUS_FIELD_NUMBER: builtins.int + @property + def producer(self) -> global___Producer: + """The Producer that closed the Stream.""" + @property + def result(self) -> temporalio.api.common.v1.message_pb2.Payload: ... + @property + def failure(self) -> temporalio.api.failure.v1.message_pb2.Failure: ... + status: temporalio.api.enums.v1.stream_pb2.StreamCloseStatus.ValueType + """The result or the failure stays next to it.""" + def __init__( + self, + *, + producer: global___Producer | None = ..., + result: temporalio.api.common.v1.message_pb2.Payload | None = ..., + failure: temporalio.api.failure.v1.message_pb2.Failure | None = ..., + status: temporalio.api.enums.v1.stream_pb2.StreamCloseStatus.ValueType = ..., + ) -> None: ... + def HasField( + self, + field_name: typing_extensions.Literal[ + "failure", b"failure", "producer", b"producer", "result", b"result" + ], + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "failure", + b"failure", + "producer", + b"producer", + "result", + b"result", + "status", + b"status", + ], + ) -> None: ... + +global___StreamClose = StreamClose + +class StreamRecord(google.protobuf.message.Message): + """Experimental. Might change in the future. + An offset is the Redis entry id packed as ms << 20 | seq. It is ordered and unique in one + Stream, and it is sparse. Offsets have gaps, so they don't count Records. Zero is never an + offset. A read names its Stream, so the read checks the offset against that Stream. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + OFFSET_FIELD_NUMBER: builtins.int + DATA_FIELD_NUMBER: builtins.int + SUPERSEDED_FIELD_NUMBER: builtins.int + CLOSE_FIELD_NUMBER: builtins.int + offset: builtins.int + @property + def data(self) -> global___StreamData: ... + @property + def superseded(self) -> global___Supersession: + """The first batch of a higher epoch carries it. It withdraws the earlier Records that its + facts name, and a reader drops them. + """ + @property + def close(self) -> global___StreamClose: ... + def __init__( + self, + *, + offset: builtins.int = ..., + data: global___StreamData | None = ..., + superseded: global___Supersession | None = ..., + close: global___StreamClose | None = ..., + ) -> None: ... + def HasField( + self, + field_name: typing_extensions.Literal[ + "close", + b"close", + "data", + b"data", + "item", + b"item", + "superseded", + b"superseded", + ], + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "close", + b"close", + "data", + b"data", + "item", + b"item", + "offset", + b"offset", + "superseded", + b"superseded", + ], + ) -> None: ... + def WhichOneof( + self, oneof_group: typing_extensions.Literal["item", b"item"] + ) -> typing_extensions.Literal["data", "superseded", "close"] | None: ... + +global___StreamRecord = StreamRecord + +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_extensions.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..d53285014f --- /dev/null +++ b/temporalio/api/streamservice/v1/request_response_pb2.py @@ -0,0 +1,180 @@ +# -*- 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 message as _message +from google.protobuf import reflection as _reflection +from google.protobuf import symbol_database as _symbol_database + +# @@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.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, +) +from temporalio.api.failure.v1 import ( + message_pb2 as temporal_dot_api_dot_failure_dot_v1_dot_message__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/common/v1/message.proto\x1a"temporal/api/enums/v1/stream.proto\x1a%temporal/api/failure/v1/message.proto\x1a$temporal/api/stream/v1/message.proto"\x8c\x02\n\x13\x43reateStreamRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12\x11\n\tstream_id\x18\x02 \x01(\t\x12,\n\tretention\x18\x03 \x01(\x0b\x32\x19.google.protobuf.Duration\x12\x14\n\x0cwithdraw_run\x18\x04 \x01(\t\x12/\n\x05owner\x18\x06 \x01(\x0b\x32 .temporal.api.stream.v1.Producer\x12\x16\n\x0e\x63lose_with_run\x18\x07 \x01(\x08\x12.\n\x03ref\x18\x08 \x01(\x0b\x32!.temporal.api.stream.v1.StreamRef\x12\x12\n\ntask_token\x18\t \x01(\x0c"\x8f\x01\n\x14\x43reateStreamResponse\x12\x31\n\x06stream\x18\x01 \x01(\x0b\x32!.temporal.api.stream.v1.StreamRef\x12\x0f\n\x07\x63reated\x18\x02 \x01(\x08\x12\x33\n\x06status\x18\x03 \x01(\x0e\x32#.temporal.api.enums.v1.StreamStatus"\xff\x02\n\x14\x41ppendRecordsRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12\x31\n\x06stream\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\x12J\n\x17previous_completed_task\x18\x06 \x01(\x0b\x32).temporal.api.stream.v1.WorkflowTaskEpoch\x12\x14\n\x0c\x66\x61iled_tasks\x18\x07 \x03(\x03\x12\x14\n\x0cwithdraw_run\x18\x08 \x01(\t\x12\x18\n\x10\x64\x65\x66\x65r_withdrawal\x18\t \x01(\x08\x12\x12\n\ntask_token\x18\n \x01(\x0c"B\n\x15\x41ppendRecordsResponse\x12\x14\n\x0c\x66irst_offset\x18\x01 \x01(\x03\x12\x13\n\x0blast_offset\x18\x02 \x01(\x03"\xb8\x03\n\x12\x43loseStreamRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12\x31\n\x06stream\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/\n\x06result\x18\x04 \x01(\x0b\x32\x1f.temporal.api.common.v1.Payload\x12\x31\n\x07\x66\x61ilure\x18\x05 \x01(\x0b\x32 .temporal.api.failure.v1.Failure\x12J\n\x17previous_completed_task\x18\x06 \x01(\x0b\x32).temporal.api.stream.v1.WorkflowTaskEpoch\x12\x14\n\x0c\x66\x61iled_tasks\x18\x07 \x03(\x03\x12\x14\n\x0cwithdraw_run\x18\x08 \x01(\t\x12\x38\n\x06status\x18\t \x01(\x0e\x32(.temporal.api.enums.v1.StreamCloseStatus\x12\x12\n\ntask_token\x18\n \x01(\x0c"\x15\n\x13\x43loseStreamResponse"\x93\x03\n\x12ReadRecordsRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12\x31\n\x06stream\x18\x02 \x01(\x0b\x32!.temporal.api.stream.v1.StreamRef\x12\x13\n\x0b\x66rom_offset\x18\x03 \x01(\x03\x12\x35\n\x05start\x18\x04 \x01(\x0e\x32&.temporal.api.enums.v1.StreamReadStart\x12\x13\n\x0bmax_records\x18\x06 \x01(\x05\x12\'\n\x04wait\x18\x07 \x01(\x0b\x32\x19.google.protobuf.Duration\x12.\n\x0b\x63reate_wait\x18\x08 \x01(\x0b\x32\x19.google.protobuf.Duration\x12\x11\n\tto_offset\x18\x0b \x01(\x03\x12\x11\n\twake_only\x18\x0c \x01(\x08\x12\x0c\n\x04last\x18\r \x01(\x05\x12\x11\n\tmax_bytes\x18\x0e \x01(\x03\x12\x12\n\nfloor_from\x18\x0f \x01(\x03J\x04\x08\x05\x10\x06J\x04\x08\n\x10\x0bR\x06topicsR\x0etopic_prefixes"\x92\x01\n\x13ReadRecordsResponse\x12\x35\n\x07records\x18\x01 \x03(\x0b\x32$.temporal.api.stream.v1.StreamRecord\x12\x13\n\x0bnext_offset\x18\x02 \x01(\x03\x12\x0c\n\x04\x64one\x18\x03 \x01(\x08\x12!\n\x19\x63ontinue_as_new_suggested\x18\x04 \x01(\x08"]\n\x15\x44\x65scribeStreamRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12\x31\n\x06stream\x18\x02 \x01(\x0b\x32!.temporal.api.stream.v1.StreamRef"\x90\x01\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(\x03\x12,\n\tretention\x18\x03 \x01(\x0b\x32\x19.google.protobuf.DurationB\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' +) + + +_CREATESTREAMREQUEST = DESCRIPTOR.message_types_by_name["CreateStreamRequest"] +_CREATESTREAMRESPONSE = DESCRIPTOR.message_types_by_name["CreateStreamResponse"] +_APPENDRECORDSREQUEST = DESCRIPTOR.message_types_by_name["AppendRecordsRequest"] +_APPENDRECORDSRESPONSE = DESCRIPTOR.message_types_by_name["AppendRecordsResponse"] +_CLOSESTREAMREQUEST = DESCRIPTOR.message_types_by_name["CloseStreamRequest"] +_CLOSESTREAMRESPONSE = DESCRIPTOR.message_types_by_name["CloseStreamResponse"] +_READRECORDSREQUEST = DESCRIPTOR.message_types_by_name["ReadRecordsRequest"] +_READRECORDSRESPONSE = DESCRIPTOR.message_types_by_name["ReadRecordsResponse"] +_DESCRIBESTREAMREQUEST = DESCRIPTOR.message_types_by_name["DescribeStreamRequest"] +_DESCRIBESTREAMRESPONSE = DESCRIPTOR.message_types_by_name["DescribeStreamResponse"] +CreateStreamRequest = _reflection.GeneratedProtocolMessageType( + "CreateStreamRequest", + (_message.Message,), + { + "DESCRIPTOR": _CREATESTREAMREQUEST, + "__module__": "temporalio.api.streamservice.v1.request_response_pb2", + # @@protoc_insertion_point(class_scope:temporal.api.streamservice.v1.CreateStreamRequest) + }, +) +_sym_db.RegisterMessage(CreateStreamRequest) + +CreateStreamResponse = _reflection.GeneratedProtocolMessageType( + "CreateStreamResponse", + (_message.Message,), + { + "DESCRIPTOR": _CREATESTREAMRESPONSE, + "__module__": "temporalio.api.streamservice.v1.request_response_pb2", + # @@protoc_insertion_point(class_scope:temporal.api.streamservice.v1.CreateStreamResponse) + }, +) +_sym_db.RegisterMessage(CreateStreamResponse) + +AppendRecordsRequest = _reflection.GeneratedProtocolMessageType( + "AppendRecordsRequest", + (_message.Message,), + { + "DESCRIPTOR": _APPENDRECORDSREQUEST, + "__module__": "temporalio.api.streamservice.v1.request_response_pb2", + # @@protoc_insertion_point(class_scope:temporal.api.streamservice.v1.AppendRecordsRequest) + }, +) +_sym_db.RegisterMessage(AppendRecordsRequest) + +AppendRecordsResponse = _reflection.GeneratedProtocolMessageType( + "AppendRecordsResponse", + (_message.Message,), + { + "DESCRIPTOR": _APPENDRECORDSRESPONSE, + "__module__": "temporalio.api.streamservice.v1.request_response_pb2", + # @@protoc_insertion_point(class_scope:temporal.api.streamservice.v1.AppendRecordsResponse) + }, +) +_sym_db.RegisterMessage(AppendRecordsResponse) + +CloseStreamRequest = _reflection.GeneratedProtocolMessageType( + "CloseStreamRequest", + (_message.Message,), + { + "DESCRIPTOR": _CLOSESTREAMREQUEST, + "__module__": "temporalio.api.streamservice.v1.request_response_pb2", + # @@protoc_insertion_point(class_scope:temporal.api.streamservice.v1.CloseStreamRequest) + }, +) +_sym_db.RegisterMessage(CloseStreamRequest) + +CloseStreamResponse = _reflection.GeneratedProtocolMessageType( + "CloseStreamResponse", + (_message.Message,), + { + "DESCRIPTOR": _CLOSESTREAMRESPONSE, + "__module__": "temporalio.api.streamservice.v1.request_response_pb2", + # @@protoc_insertion_point(class_scope:temporal.api.streamservice.v1.CloseStreamResponse) + }, +) +_sym_db.RegisterMessage(CloseStreamResponse) + +ReadRecordsRequest = _reflection.GeneratedProtocolMessageType( + "ReadRecordsRequest", + (_message.Message,), + { + "DESCRIPTOR": _READRECORDSREQUEST, + "__module__": "temporalio.api.streamservice.v1.request_response_pb2", + # @@protoc_insertion_point(class_scope:temporal.api.streamservice.v1.ReadRecordsRequest) + }, +) +_sym_db.RegisterMessage(ReadRecordsRequest) + +ReadRecordsResponse = _reflection.GeneratedProtocolMessageType( + "ReadRecordsResponse", + (_message.Message,), + { + "DESCRIPTOR": _READRECORDSRESPONSE, + "__module__": "temporalio.api.streamservice.v1.request_response_pb2", + # @@protoc_insertion_point(class_scope:temporal.api.streamservice.v1.ReadRecordsResponse) + }, +) +_sym_db.RegisterMessage(ReadRecordsResponse) + +DescribeStreamRequest = _reflection.GeneratedProtocolMessageType( + "DescribeStreamRequest", + (_message.Message,), + { + "DESCRIPTOR": _DESCRIBESTREAMREQUEST, + "__module__": "temporalio.api.streamservice.v1.request_response_pb2", + # @@protoc_insertion_point(class_scope:temporal.api.streamservice.v1.DescribeStreamRequest) + }, +) +_sym_db.RegisterMessage(DescribeStreamRequest) + +DescribeStreamResponse = _reflection.GeneratedProtocolMessageType( + "DescribeStreamResponse", + (_message.Message,), + { + "DESCRIPTOR": _DESCRIBESTREAMRESPONSE, + "__module__": "temporalio.api.streamservice.v1.request_response_pb2", + # @@protoc_insertion_point(class_scope:temporal.api.streamservice.v1.DescribeStreamResponse) + }, +) +_sym_db.RegisterMessage(DescribeStreamResponse) + +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 = 271 + _CREATESTREAMREQUEST._serialized_end = 539 + _CREATESTREAMRESPONSE._serialized_start = 542 + _CREATESTREAMRESPONSE._serialized_end = 685 + _APPENDRECORDSREQUEST._serialized_start = 688 + _APPENDRECORDSREQUEST._serialized_end = 1071 + _APPENDRECORDSRESPONSE._serialized_start = 1073 + _APPENDRECORDSRESPONSE._serialized_end = 1139 + _CLOSESTREAMREQUEST._serialized_start = 1142 + _CLOSESTREAMREQUEST._serialized_end = 1582 + _CLOSESTREAMRESPONSE._serialized_start = 1584 + _CLOSESTREAMRESPONSE._serialized_end = 1605 + _READRECORDSREQUEST._serialized_start = 1608 + _READRECORDSREQUEST._serialized_end = 2011 + _READRECORDSRESPONSE._serialized_start = 2014 + _READRECORDSRESPONSE._serialized_end = 2160 + _DESCRIBESTREAMREQUEST._serialized_start = 2162 + _DESCRIBESTREAMREQUEST._serialized_end = 2255 + _DESCRIBESTREAMRESPONSE._serialized_start = 2258 + _DESCRIBESTREAMRESPONSE._serialized_end = 2402 +# @@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..f7bd4e2fab --- /dev/null +++ b/temporalio/api/streamservice/v1/request_response_pb2.pyi @@ -0,0 +1,619 @@ +""" +@generated by mypy-protobuf. Do not edit manually! +isort:skip_file +""" + +import builtins +import collections.abc +import sys + +import google.protobuf.descriptor +import google.protobuf.duration_pb2 +import google.protobuf.internal.containers +import google.protobuf.message + +import temporalio.api.common.v1.message_pb2 +import temporalio.api.enums.v1.stream_pb2 +import temporalio.api.failure.v1.message_pb2 +import temporalio.api.stream.v1.message_pb2 + +if sys.version_info >= (3, 8): + import typing as typing_extensions +else: + import typing_extensions + +DESCRIPTOR: google.protobuf.descriptor.FileDescriptor + +class CreateStreamRequest(google.protobuf.message.Message): + """Experimental. Might change in the future. + Creates the Stream, or gets it when it exists. Then created is false, and the Stream keeps its + own Retention. + (-- api-linter: core::0133=disabled + aip.dev/not-precedent: The request names a Stream by namespace and stream_id. --) + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + NAMESPACE_FIELD_NUMBER: builtins.int + STREAM_ID_FIELD_NUMBER: builtins.int + RETENTION_FIELD_NUMBER: builtins.int + WITHDRAW_RUN_FIELD_NUMBER: builtins.int + OWNER_FIELD_NUMBER: builtins.int + CLOSE_WITH_RUN_FIELD_NUMBER: builtins.int + REF_FIELD_NUMBER: builtins.int + TASK_TOKEN_FIELD_NUMBER: builtins.int + namespace: builtins.str + stream_id: builtins.str + """A Store requires it. Outside Workflow code, the server makes an id when it is empty.""" + @property + def retention(self) -> google.protobuf.duration_pb2.Duration: + """The age after which a Record is dropped. Empty is the Store default.""" + withdraw_run: builtins.str + """Workflow only. The previous run on a retry run's first create, to withdraw its Records.""" + @property + def owner(self) -> temporalio.api.stream.v1.message_pb2.Producer: + """The Workflow run that owns the Stream. Workflow kind, with its run_id and epoch.""" + close_with_run: builtins.bool + """Only the server reads it. The server closes the Stream with the outcome of the owner's run + chain, terminate and timeout included. It needs owner, and a ref with a Workflow owner. + (-- api-linter: core::0140::prepositions=disabled + aip.dev/not-precedent: "with" ties the close to the owner's run chain. --) + """ + @property + def ref(self) -> temporalio.api.stream.v1.message_pb2.StreamRef: + """The ref of the Stream. A Workflow-owned Stream sets the ref's owner, which names the + owner's run chain. + """ + task_token: builtins.bytes + """Only the server reads it. The server checks that the token is the Activity's current + attempt, so a caller can't put a Stream into a Standalone Activity it isn't. + """ + def __init__( + self, + *, + namespace: builtins.str = ..., + stream_id: builtins.str = ..., + retention: google.protobuf.duration_pb2.Duration | None = ..., + withdraw_run: builtins.str = ..., + owner: temporalio.api.stream.v1.message_pb2.Producer | None = ..., + close_with_run: builtins.bool = ..., + ref: temporalio.api.stream.v1.message_pb2.StreamRef | None = ..., + task_token: builtins.bytes = ..., + ) -> None: ... + def HasField( + self, + field_name: typing_extensions.Literal[ + "owner", b"owner", "ref", b"ref", "retention", b"retention" + ], + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "close_with_run", + b"close_with_run", + "namespace", + b"namespace", + "owner", + b"owner", + "ref", + b"ref", + "retention", + b"retention", + "stream_id", + b"stream_id", + "task_token", + b"task_token", + "withdraw_run", + b"withdraw_run", + ], + ) -> None: ... + +global___CreateStreamRequest = CreateStreamRequest + +class CreateStreamResponse(google.protobuf.message.Message): + """Experimental. Might change in the future.""" + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + STREAM_FIELD_NUMBER: builtins.int + CREATED_FIELD_NUMBER: builtins.int + STATUS_FIELD_NUMBER: builtins.int + @property + def stream(self) -> temporalio.api.stream.v1.message_pb2.StreamRef: ... + created: builtins.bool + status: temporalio.api.enums.v1.stream_pb2.StreamStatus.ValueType + def __init__( + self, + *, + stream: temporalio.api.stream.v1.message_pb2.StreamRef | None = ..., + created: builtins.bool = ..., + status: temporalio.api.enums.v1.stream_pb2.StreamStatus.ValueType = ..., + ) -> None: ... + def HasField( + self, field_name: typing_extensions.Literal["stream", b"stream"] + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "created", b"created", "status", b"status", "stream", b"stream" + ], + ) -> None: ... + +global___CreateStreamResponse = CreateStreamResponse + +class AppendRecordsRequest(google.protobuf.message.Message): + """Experimental. Might change in the future. + FAILED_PRECONDITION if closed, fenced, or if the Stream's meta has a higher format version v + (detail stream_format_too_new). NOT_FOUND if absent. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + NAMESPACE_FIELD_NUMBER: builtins.int + STREAM_FIELD_NUMBER: builtins.int + PRODUCER_FIELD_NUMBER: builtins.int + SEQUENCE_FIELD_NUMBER: builtins.int + RECORDS_FIELD_NUMBER: builtins.int + PREVIOUS_COMPLETED_TASK_FIELD_NUMBER: builtins.int + FAILED_TASKS_FIELD_NUMBER: builtins.int + WITHDRAW_RUN_FIELD_NUMBER: builtins.int + DEFER_WITHDRAWAL_FIELD_NUMBER: builtins.int + TASK_TOKEN_FIELD_NUMBER: builtins.int + namespace: builtins.str + @property + def stream(self) -> temporalio.api.stream.v1.message_pb2.StreamRef: ... + @property + def producer(self) -> temporalio.api.stream.v1.message_pb2.Producer: + """Required outside a Workflow. In a Workflow the transport sets it.""" + sequence: builtins.int + """The batch number in this epoch. With the Producer and epoch it is the dedupe key.""" + @property + def records( + self, + ) -> google.protobuf.internal.containers.RepeatedCompositeFieldContainer[ + temporalio.api.stream.v1.message_pb2.AppendRecord + ]: ... + @property + def previous_completed_task( + self, + ) -> temporalio.api.stream.v1.message_pb2.WorkflowTaskEpoch: + """The run's last completed Workflow Task (started_event_id 0 before the first) and its failed + tasks, from a Local Activity's start. The batch withdraws the run's Local Activity Records + whose task never completed. + """ + @property + def failed_tasks( + self, + ) -> google.protobuf.internal.containers.RepeatedScalarFieldContainer[ + builtins.int + ]: ... + withdraw_run: builtins.str + """Local Activity only. The previous run on a retry run's first batch, to withdraw its Records.""" + defer_withdrawal: builtins.bool + """Only the server reads it, on a Workflow Producer's batch. The batch withdraws no Workflow + Task and keeps no task facts, but withdraw_run still applies. + """ + task_token: builtins.bytes + """Only the server reads it. The server checks the Producer against the Activity's current + attempt, so nobody else writes as that Activity. + """ + def __init__( + self, + *, + namespace: builtins.str = ..., + stream: 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 = ..., + previous_completed_task: temporalio.api.stream.v1.message_pb2.WorkflowTaskEpoch + | None = ..., + failed_tasks: collections.abc.Iterable[builtins.int] | None = ..., + withdraw_run: builtins.str = ..., + defer_withdrawal: builtins.bool = ..., + task_token: builtins.bytes = ..., + ) -> None: ... + def HasField( + self, + field_name: typing_extensions.Literal[ + "previous_completed_task", + b"previous_completed_task", + "producer", + b"producer", + "stream", + b"stream", + ], + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "defer_withdrawal", + b"defer_withdrawal", + "failed_tasks", + b"failed_tasks", + "namespace", + b"namespace", + "previous_completed_task", + b"previous_completed_task", + "producer", + b"producer", + "records", + b"records", + "sequence", + b"sequence", + "stream", + b"stream", + "task_token", + b"task_token", + "withdraw_run", + b"withdraw_run", + ], + ) -> None: ... + +global___AppendRecordsRequest = AppendRecordsRequest + +class AppendRecordsResponse(google.protobuf.message.Message): + """Experimental. Might change in the future.""" + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + FIRST_OFFSET_FIELD_NUMBER: builtins.int + LAST_OFFSET_FIELD_NUMBER: builtins.int + first_offset: builtins.int + last_offset: builtins.int + def __init__( + self, + *, + first_offset: builtins.int = ..., + last_offset: builtins.int = ..., + ) -> None: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "first_offset", b"first_offset", "last_offset", b"last_offset" + ], + ) -> None: ... + +global___AppendRecordsResponse = AppendRecordsResponse + +class CloseStreamRequest(google.protobuf.message.Message): + """Experimental. Might change in the future.""" + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + NAMESPACE_FIELD_NUMBER: builtins.int + STREAM_FIELD_NUMBER: builtins.int + PRODUCER_FIELD_NUMBER: builtins.int + RESULT_FIELD_NUMBER: builtins.int + FAILURE_FIELD_NUMBER: builtins.int + PREVIOUS_COMPLETED_TASK_FIELD_NUMBER: builtins.int + FAILED_TASKS_FIELD_NUMBER: builtins.int + WITHDRAW_RUN_FIELD_NUMBER: builtins.int + STATUS_FIELD_NUMBER: builtins.int + TASK_TOKEN_FIELD_NUMBER: builtins.int + namespace: builtins.str + @property + def stream(self) -> temporalio.api.stream.v1.message_pb2.StreamRef: ... + @property + def producer(self) -> temporalio.api.stream.v1.message_pb2.Producer: ... + @property + def result(self) -> temporalio.api.common.v1.message_pb2.Payload: ... + @property + def failure(self) -> temporalio.api.failure.v1.message_pb2.Failure: ... + @property + def previous_completed_task( + self, + ) -> temporalio.api.stream.v1.message_pb2.WorkflowTaskEpoch: + """Only the server reads these. They are the run's task facts, as on the append.""" + @property + def failed_tasks( + self, + ) -> google.protobuf.internal.containers.RepeatedScalarFieldContainer[ + builtins.int + ]: ... + withdraw_run: builtins.str + """Only the server reads it. The previous run on a retry run's close, to withdraw its Records.""" + status: temporalio.api.enums.v1.stream_pb2.StreamCloseStatus.ValueType + """Empty is COMPLETED with no failure and FAILED with one. A caller sets CANCELED itself.""" + task_token: builtins.bytes + """Only the server reads it. It checks the token as on the append.""" + def __init__( + self, + *, + namespace: builtins.str = ..., + stream: temporalio.api.stream.v1.message_pb2.StreamRef | None = ..., + producer: temporalio.api.stream.v1.message_pb2.Producer | None = ..., + result: temporalio.api.common.v1.message_pb2.Payload | None = ..., + failure: temporalio.api.failure.v1.message_pb2.Failure | None = ..., + previous_completed_task: temporalio.api.stream.v1.message_pb2.WorkflowTaskEpoch + | None = ..., + failed_tasks: collections.abc.Iterable[builtins.int] | None = ..., + withdraw_run: builtins.str = ..., + status: temporalio.api.enums.v1.stream_pb2.StreamCloseStatus.ValueType = ..., + task_token: builtins.bytes = ..., + ) -> None: ... + def HasField( + self, + field_name: typing_extensions.Literal[ + "failure", + b"failure", + "previous_completed_task", + b"previous_completed_task", + "producer", + b"producer", + "result", + b"result", + "stream", + b"stream", + ], + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "failed_tasks", + b"failed_tasks", + "failure", + b"failure", + "namespace", + b"namespace", + "previous_completed_task", + b"previous_completed_task", + "producer", + b"producer", + "result", + b"result", + "status", + b"status", + "stream", + b"stream", + "task_token", + b"task_token", + "withdraw_run", + b"withdraw_run", + ], + ) -> None: ... + +global___CloseStreamRequest = CloseStreamRequest + +class CloseStreamResponse(google.protobuf.message.Message): + """Experimental. Might change in the future.""" + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + def __init__( + self, + ) -> None: ... + +global___CloseStreamResponse = CloseStreamResponse + +class ReadRecordsRequest(google.protobuf.message.Message): + """Experimental. Might change in the future.""" + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + NAMESPACE_FIELD_NUMBER: builtins.int + STREAM_FIELD_NUMBER: builtins.int + FROM_OFFSET_FIELD_NUMBER: builtins.int + START_FIELD_NUMBER: builtins.int + MAX_RECORDS_FIELD_NUMBER: builtins.int + WAIT_FIELD_NUMBER: builtins.int + CREATE_WAIT_FIELD_NUMBER: builtins.int + TO_OFFSET_FIELD_NUMBER: builtins.int + WAKE_ONLY_FIELD_NUMBER: builtins.int + LAST_FIELD_NUMBER: builtins.int + MAX_BYTES_FIELD_NUMBER: builtins.int + FLOOR_FROM_FIELD_NUMBER: builtins.int + namespace: builtins.str + @property + def stream(self) -> temporalio.api.stream.v1.message_pb2.StreamRef: ... + from_offset: builtins.int + """Start at this offset. Zero uses start. INVALID_ARGUMENT if past the newest offset + 1. + (-- api-linter: core::0140::prepositions=disabled + aip.dev/not-precedent: "from" and "to" mark the ends of a range of offsets. --) + """ + start: temporalio.api.enums.v1.stream_pb2.StreamReadStart.ValueType + max_records: builtins.int + @property + def wait(self) -> google.protobuf.duration_pb2.Duration: ... + @property + def create_wait(self) -> google.protobuf.duration_pb2.Duration: + """How long to wait for a Stream that nobody created yet. Empty fails at once with + NOT_FOUND. At most 60 s. + """ + to_offset: builtins.int + """The exclusive end of the read. The read scans no entry at or after it, the close Record + included, so a replay reads the same range. Zero means no end. Only the server serves it, + and a Store refuses it. + (-- api-linter: core::0140::prepositions=disabled + aip.dev/not-precedent: "from" and "to" mark the ends of a range of offsets. --) + """ + wake_only: builtins.bool + """Park as a normal read, then answer with no Records and next_offset = from_offset, when a + Record at or after from_offset exists, when the Stream closes (done), or when wait ends. Its + wait may be days, since a parked read adds nothing to History while it waits. Only the + server serves it, and a Store refuses it. + """ + last: builtins.int + """Read the newest `last` Records, oldest first. It needs no from_offset and no start. Not with + wake_only, and not in Workflow code, where replay needs a fixed range. + """ + 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. + """ + floor_from: builtins.int + """The lowest offset this run read on replay. A reset run copies the earlier run's reads into + its History, so its floor must cover them too, not only its first live page. Only the + server reads it. + (-- api-linter: core::0140::prepositions=disabled + aip.dev/not-precedent: The floor is the lowest from_offset of the run's reads. --) + """ + def __init__( + self, + *, + namespace: builtins.str = ..., + stream: temporalio.api.stream.v1.message_pb2.StreamRef | None = ..., + from_offset: builtins.int = ..., + start: temporalio.api.enums.v1.stream_pb2.StreamReadStart.ValueType = ..., + max_records: builtins.int = ..., + wait: google.protobuf.duration_pb2.Duration | None = ..., + create_wait: google.protobuf.duration_pb2.Duration | None = ..., + to_offset: builtins.int = ..., + wake_only: builtins.bool = ..., + last: builtins.int = ..., + max_bytes: builtins.int = ..., + floor_from: builtins.int = ..., + ) -> None: ... + def HasField( + self, + field_name: typing_extensions.Literal[ + "create_wait", b"create_wait", "stream", b"stream", "wait", b"wait" + ], + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "create_wait", + b"create_wait", + "floor_from", + b"floor_from", + "from_offset", + b"from_offset", + "last", + b"last", + "max_bytes", + b"max_bytes", + "max_records", + b"max_records", + "namespace", + b"namespace", + "start", + b"start", + "stream", + b"stream", + "to_offset", + b"to_offset", + "wait", + b"wait", + "wake_only", + b"wake_only", + ], + ) -> None: ... + +global___ReadRecordsRequest = ReadRecordsRequest + +class ReadRecordsResponse(google.protobuf.message.Message): + """Experimental. Might change in the future.""" + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + RECORDS_FIELD_NUMBER: builtins.int + NEXT_OFFSET_FIELD_NUMBER: builtins.int + DONE_FIELD_NUMBER: builtins.int + CONTINUE_AS_NEW_SUGGESTED_FIELD_NUMBER: builtins.int + @property + def records( + self, + ) -> google.protobuf.internal.containers.RepeatedCompositeFieldContainer[ + temporalio.api.stream.v1.message_pb2.StreamRecord + ]: ... + next_offset: builtins.int + """The from_offset of the next read, which is the newest offset read + 1.""" + done: builtins.bool + continue_as_new_suggested: builtins.bool + """This reader's floor keeps more than half of the Stream's byte cap. The writer fails when the + cap fills, and only the reader's continue-as-new frees it. Only the server sets it. + """ + def __init__( + self, + *, + records: collections.abc.Iterable[ + temporalio.api.stream.v1.message_pb2.StreamRecord + ] + | None = ..., + next_offset: builtins.int = ..., + done: builtins.bool = ..., + continue_as_new_suggested: builtins.bool = ..., + ) -> None: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "continue_as_new_suggested", + b"continue_as_new_suggested", + "done", + b"done", + "next_offset", + b"next_offset", + "records", + b"records", + ], + ) -> None: ... + +global___ReadRecordsResponse = ReadRecordsResponse + +class DescribeStreamRequest(google.protobuf.message.Message): + """Experimental. Might change in the future.""" + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + NAMESPACE_FIELD_NUMBER: builtins.int + STREAM_FIELD_NUMBER: builtins.int + namespace: builtins.str + @property + def stream(self) -> temporalio.api.stream.v1.message_pb2.StreamRef: ... + def __init__( + self, + *, + namespace: builtins.str = ..., + stream: temporalio.api.stream.v1.message_pb2.StreamRef | None = ..., + ) -> None: ... + def HasField( + self, field_name: typing_extensions.Literal["stream", b"stream"] + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "namespace", b"namespace", "stream", b"stream" + ], + ) -> None: ... + +global___DescribeStreamRequest = DescribeStreamRequest + +class DescribeStreamResponse(google.protobuf.message.Message): + """Experimental. Might change in the future.""" + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + STATUS_FIELD_NUMBER: builtins.int + LAST_OFFSET_FIELD_NUMBER: builtins.int + RETENTION_FIELD_NUMBER: builtins.int + status: temporalio.api.enums.v1.stream_pb2.StreamStatus.ValueType + last_offset: builtins.int + """Zero when the Stream has no Record.""" + @property + def retention(self) -> google.protobuf.duration_pb2.Duration: ... + def __init__( + self, + *, + status: temporalio.api.enums.v1.stream_pb2.StreamStatus.ValueType = ..., + last_offset: builtins.int = ..., + retention: google.protobuf.duration_pb2.Duration | None = ..., + ) -> None: ... + def HasField( + self, field_name: typing_extensions.Literal["retention", b"retention"] + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "last_offset", + b"last_offset", + "retention", + b"retention", + "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..f3a5a087ef --- /dev/null +++ b/temporalio/api/streamservice/v1/request_response_pb2_grpc.pyi @@ -0,0 +1,4 @@ +""" +@generated by mypy-protobuf. Do not edit manually! +isort:skip_file +""" diff --git a/temporalio/common/_streams/__init__.py b/temporalio/common/_streams/__init__.py new file mode 100644 index 0000000000..3888aaa785 --- /dev/null +++ b/temporalio/common/_streams/__init__.py @@ -0,0 +1 @@ +"""The Streams API, which :py:mod:`temporalio.common` exports.""" diff --git a/temporalio/common/_streams/proto/__init__.py b/temporalio/common/_streams/proto/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/temporalio/common/_streams/proto/v1/__init__.py b/temporalio/common/_streams/proto/v1/__init__.py new file mode 100644 index 0000000000..f94fb9f8a4 --- /dev/null +++ b/temporalio/common/_streams/proto/v1/__init__.py @@ -0,0 +1,29 @@ +from .redis_pb2 import RedisStreamConfig +from .runtime_pb2 import ( + CloseWriterRequest, + CloseWriterResponse, + CloseWritersRequest, + CloseWritersResponse, + FreeWriterRequest, + FreeWriterResponse, + MemoryStreamConfig, + OpenWriterRequest, + OpenWriterResponse, + WriteRecordsRequest, + WriteRecordsResponse, +) + +__all__ = [ + "CloseWriterRequest", + "CloseWriterResponse", + "CloseWritersRequest", + "CloseWritersResponse", + "FreeWriterRequest", + "FreeWriterResponse", + "MemoryStreamConfig", + "OpenWriterRequest", + "OpenWriterResponse", + "RedisStreamConfig", + "WriteRecordsRequest", + "WriteRecordsResponse", +] diff --git a/temporalio/common/_streams/proto/v1/redis_pb2.py b/temporalio/common/_streams/proto/v1/redis_pb2.py new file mode 100644 index 0000000000..046c2b46e0 --- /dev/null +++ b/temporalio/common/_streams/proto/v1/redis_pb2.py @@ -0,0 +1,40 @@ +# -*- coding: utf-8 -*- +# Generated by the protocol buffer compiler. DO NOT EDIT! +# source: temporal/sdk/streams/v1/redis.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 message as _message +from google.protobuf import reflection as _reflection +from google.protobuf import symbol_database as _symbol_database + +# @@protoc_insertion_point(imports) + +_sym_db = _symbol_database.Default() + + +from google.protobuf import duration_pb2 as google_dot_protobuf_dot_duration__pb2 + +DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile( + b'\n#temporal/sdk/streams/v1/redis.proto\x12\x17temporal.sdk.streams.v1\x1a\x1egoogle/protobuf/duration.proto"\x8a\x02\n\x11RedisStreamConfig\x12\x0c\n\x04urls\x18\x01 \x03(\t\x12\x0f\n\x07\x63luster\x18\x02 \x01(\x08\x12\x12\n\nkey_prefix\x18\x03 \x01(\t\x12,\n\tretention\x18\x04 \x01(\x0b\x32\x19.google.protobuf.Duration\x12-\n\nsafety_ttl\x18\x05 \x01(\x0b\x32\x19.google.protobuf.Duration\x12\x1f\n\x17\x62locking_reads_per_node\x18\x08 \x01(\x05\x12\x33\n\x10response_timeout\x18\t \x01(\x0b\x32\x19.google.protobuf.Duration\x12\x0f\n\x07\x62\x61\x63kend\x18\n \x01(\tb\x06proto3' +) + + +_REDISSTREAMCONFIG = DESCRIPTOR.message_types_by_name["RedisStreamConfig"] +RedisStreamConfig = _reflection.GeneratedProtocolMessageType( + "RedisStreamConfig", + (_message.Message,), + { + "DESCRIPTOR": _REDISSTREAMCONFIG, + "__module__": "temporalio.common._streams.proto.v1.redis_pb2", + # @@protoc_insertion_point(class_scope:temporal.sdk.streams.v1.RedisStreamConfig) + }, +) +_sym_db.RegisterMessage(RedisStreamConfig) + +if _descriptor._USE_C_DESCRIPTORS == False: + DESCRIPTOR._options = None + _REDISSTREAMCONFIG._serialized_start = 97 + _REDISSTREAMCONFIG._serialized_end = 363 +# @@protoc_insertion_point(module_scope) diff --git a/temporalio/common/_streams/proto/v1/redis_pb2.pyi b/temporalio/common/_streams/proto/v1/redis_pb2.pyi new file mode 100644 index 0000000000..50d56b184d --- /dev/null +++ b/temporalio/common/_streams/proto/v1/redis_pb2.pyi @@ -0,0 +1,106 @@ +""" +@generated by mypy-protobuf. Do not edit manually! +isort:skip_file +The config of the Redis Stream runtime. It keeps the package of the runtime protos of +`temporalio-streams`, so the message keeps its full name. Field numbers are part of the contract +with every SDK, and the memory runtime's config shares the numbers of the fields both have. +""" + +import builtins +import collections.abc +import sys + +import google.protobuf.descriptor +import google.protobuf.duration_pb2 +import google.protobuf.internal.containers +import google.protobuf.message + +if sys.version_info >= (3, 8): + import typing as typing_extensions +else: + import typing_extensions + +DESCRIPTOR: google.protobuf.descriptor.FileDescriptor + +class RedisStreamConfig(google.protobuf.message.Message): + """The settings of the Redis Stream runtime. Lang builds it from the backend options.""" + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + URLS_FIELD_NUMBER: builtins.int + CLUSTER_FIELD_NUMBER: builtins.int + KEY_PREFIX_FIELD_NUMBER: builtins.int + RETENTION_FIELD_NUMBER: builtins.int + SAFETY_TTL_FIELD_NUMBER: builtins.int + BLOCKING_READS_PER_NODE_FIELD_NUMBER: builtins.int + RESPONSE_TIMEOUT_FIELD_NUMBER: builtins.int + BACKEND_FIELD_NUMBER: builtins.int + @property + def urls( + self, + ) -> google.protobuf.internal.containers.RepeatedScalarFieldContainer[builtins.str]: + """One URL, or the seed URLs of a cluster. Credentials and TLS come from the URL.""" + cluster: builtins.bool + key_prefix: builtins.str + """Empty is `temporal-streams`.""" + @property + def retention(self) -> google.protobuf.duration_pb2.Duration: + """The default Retention of a new Stream. Empty is 7 days.""" + @property + def safety_ttl(self) -> google.protobuf.duration_pb2.Duration: + """How long an open Workflow-owned Stream stays after its last write. Empty is 90 days.""" + blocking_reads_per_node: builtins.int + """Zero is 256.""" + @property + def response_timeout(self) -> google.protobuf.duration_pb2.Duration: + """Empty is 5 s.""" + backend: builtins.str + """The backend name that lang registered this runtime under. The crate sets it on each + StreamRef that it builds, such as the one in a create response. + """ + def __init__( + self, + *, + urls: collections.abc.Iterable[builtins.str] | None = ..., + cluster: builtins.bool = ..., + key_prefix: builtins.str = ..., + retention: google.protobuf.duration_pb2.Duration | None = ..., + safety_ttl: google.protobuf.duration_pb2.Duration | None = ..., + blocking_reads_per_node: builtins.int = ..., + response_timeout: google.protobuf.duration_pb2.Duration | None = ..., + backend: builtins.str = ..., + ) -> None: ... + def HasField( + self, + field_name: typing_extensions.Literal[ + "response_timeout", + b"response_timeout", + "retention", + b"retention", + "safety_ttl", + b"safety_ttl", + ], + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "backend", + b"backend", + "blocking_reads_per_node", + b"blocking_reads_per_node", + "cluster", + b"cluster", + "key_prefix", + b"key_prefix", + "response_timeout", + b"response_timeout", + "retention", + b"retention", + "safety_ttl", + b"safety_ttl", + "urls", + b"urls", + ], + ) -> None: ... + +global___RedisStreamConfig = RedisStreamConfig diff --git a/temporalio/common/_streams/proto/v1/runtime_pb2.py b/temporalio/common/_streams/proto/v1/runtime_pb2.py new file mode 100644 index 0000000000..2901be6f73 --- /dev/null +++ b/temporalio/common/_streams/proto/v1/runtime_pb2.py @@ -0,0 +1,184 @@ +# -*- coding: utf-8 -*- +# Generated by the protocol buffer compiler. DO NOT EDIT! +# source: temporal/sdk/streams/v1/runtime.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 message as _message +from google.protobuf import reflection as _reflection +from google.protobuf import symbol_database as _symbol_database + +# @@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.stream.v1 import ( + message_pb2 as temporal_dot_api_dot_stream_dot_v1_dot_message__pb2, +) + +DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile( + b'\n%temporal/sdk/streams/v1/runtime.proto\x12\x17temporal.sdk.streams.v1\x1a\x1egoogle/protobuf/duration.proto\x1a$temporal/api/stream/v1/message.proto"\x82\x01\n\x12MemoryStreamConfig\x12,\n\tretention\x18\x04 \x01(\x0b\x32\x19.google.protobuf.Duration\x12-\n\nsafety_ttl\x18\x05 \x01(\x0b\x32\x19.google.protobuf.Duration\x12\x0f\n\x07\x62\x61\x63kend\x18\n \x01(\t"\xbe\x02\n\x11OpenWriterRequest\x12\x11\n\tnamespace\x18\x01 \x01(\t\x12\x31\n\x06stream\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\x16\n\x0emax_batch_size\x18\x05 \x01(\x05\x12\x35\n\x12max_retry_duration\x18\x06 \x01(\x0b\x32\x19.google.protobuf.Duration\x12J\n\x17previous_completed_task\x18\x08 \x01(\x0b\x32).temporal.api.stream.v1.WorkflowTaskEpoch\x12\x14\n\x0c\x66\x61iled_tasks\x18\t \x03(\x03"\'\n\x12OpenWriterResponse\x12\x11\n\twriter_id\x18\x01 \x01(\x04"_\n\x13WriteRecordsRequest\x12\x11\n\twriter_id\x18\x01 \x01(\x04\x12\x35\n\x07records\x18\x02 \x03(\x0b\x32$.temporal.api.stream.v1.AppendRecord"A\n\x14WriteRecordsResponse\x12\x14\n\x0c\x66irst_offset\x18\x01 \x01(\x03\x12\x13\n\x0blast_offset\x18\x02 \x01(\x03"\'\n\x12\x43loseWriterRequest\x12\x11\n\twriter_id\x18\x01 \x01(\x04"*\n\x13\x43loseWriterResponse\x12\x13\n\x0blast_offset\x18\x01 \x01(\x03"&\n\x11\x46reeWriterRequest\x12\x11\n\twriter_id\x18\x01 \x01(\x04"\x14\n\x12\x46reeWriterResponse"\x15\n\x13\x43loseWritersRequest"&\n\x14\x43loseWritersResponse\x12\x0e\n\x06\x66\x61iled\x18\x01 \x01(\x05\x62\x06proto3' +) + + +_MEMORYSTREAMCONFIG = DESCRIPTOR.message_types_by_name["MemoryStreamConfig"] +_OPENWRITERREQUEST = DESCRIPTOR.message_types_by_name["OpenWriterRequest"] +_OPENWRITERRESPONSE = DESCRIPTOR.message_types_by_name["OpenWriterResponse"] +_WRITERECORDSREQUEST = DESCRIPTOR.message_types_by_name["WriteRecordsRequest"] +_WRITERECORDSRESPONSE = DESCRIPTOR.message_types_by_name["WriteRecordsResponse"] +_CLOSEWRITERREQUEST = DESCRIPTOR.message_types_by_name["CloseWriterRequest"] +_CLOSEWRITERRESPONSE = DESCRIPTOR.message_types_by_name["CloseWriterResponse"] +_FREEWRITERREQUEST = DESCRIPTOR.message_types_by_name["FreeWriterRequest"] +_FREEWRITERRESPONSE = DESCRIPTOR.message_types_by_name["FreeWriterResponse"] +_CLOSEWRITERSREQUEST = DESCRIPTOR.message_types_by_name["CloseWritersRequest"] +_CLOSEWRITERSRESPONSE = DESCRIPTOR.message_types_by_name["CloseWritersResponse"] +MemoryStreamConfig = _reflection.GeneratedProtocolMessageType( + "MemoryStreamConfig", + (_message.Message,), + { + "DESCRIPTOR": _MEMORYSTREAMCONFIG, + "__module__": "temporalio.common._streams.proto.v1.runtime_pb2", + # @@protoc_insertion_point(class_scope:temporal.sdk.streams.v1.MemoryStreamConfig) + }, +) +_sym_db.RegisterMessage(MemoryStreamConfig) + +OpenWriterRequest = _reflection.GeneratedProtocolMessageType( + "OpenWriterRequest", + (_message.Message,), + { + "DESCRIPTOR": _OPENWRITERREQUEST, + "__module__": "temporalio.common._streams.proto.v1.runtime_pb2", + # @@protoc_insertion_point(class_scope:temporal.sdk.streams.v1.OpenWriterRequest) + }, +) +_sym_db.RegisterMessage(OpenWriterRequest) + +OpenWriterResponse = _reflection.GeneratedProtocolMessageType( + "OpenWriterResponse", + (_message.Message,), + { + "DESCRIPTOR": _OPENWRITERRESPONSE, + "__module__": "temporalio.common._streams.proto.v1.runtime_pb2", + # @@protoc_insertion_point(class_scope:temporal.sdk.streams.v1.OpenWriterResponse) + }, +) +_sym_db.RegisterMessage(OpenWriterResponse) + +WriteRecordsRequest = _reflection.GeneratedProtocolMessageType( + "WriteRecordsRequest", + (_message.Message,), + { + "DESCRIPTOR": _WRITERECORDSREQUEST, + "__module__": "temporalio.common._streams.proto.v1.runtime_pb2", + # @@protoc_insertion_point(class_scope:temporal.sdk.streams.v1.WriteRecordsRequest) + }, +) +_sym_db.RegisterMessage(WriteRecordsRequest) + +WriteRecordsResponse = _reflection.GeneratedProtocolMessageType( + "WriteRecordsResponse", + (_message.Message,), + { + "DESCRIPTOR": _WRITERECORDSRESPONSE, + "__module__": "temporalio.common._streams.proto.v1.runtime_pb2", + # @@protoc_insertion_point(class_scope:temporal.sdk.streams.v1.WriteRecordsResponse) + }, +) +_sym_db.RegisterMessage(WriteRecordsResponse) + +CloseWriterRequest = _reflection.GeneratedProtocolMessageType( + "CloseWriterRequest", + (_message.Message,), + { + "DESCRIPTOR": _CLOSEWRITERREQUEST, + "__module__": "temporalio.common._streams.proto.v1.runtime_pb2", + # @@protoc_insertion_point(class_scope:temporal.sdk.streams.v1.CloseWriterRequest) + }, +) +_sym_db.RegisterMessage(CloseWriterRequest) + +CloseWriterResponse = _reflection.GeneratedProtocolMessageType( + "CloseWriterResponse", + (_message.Message,), + { + "DESCRIPTOR": _CLOSEWRITERRESPONSE, + "__module__": "temporalio.common._streams.proto.v1.runtime_pb2", + # @@protoc_insertion_point(class_scope:temporal.sdk.streams.v1.CloseWriterResponse) + }, +) +_sym_db.RegisterMessage(CloseWriterResponse) + +FreeWriterRequest = _reflection.GeneratedProtocolMessageType( + "FreeWriterRequest", + (_message.Message,), + { + "DESCRIPTOR": _FREEWRITERREQUEST, + "__module__": "temporalio.common._streams.proto.v1.runtime_pb2", + # @@protoc_insertion_point(class_scope:temporal.sdk.streams.v1.FreeWriterRequest) + }, +) +_sym_db.RegisterMessage(FreeWriterRequest) + +FreeWriterResponse = _reflection.GeneratedProtocolMessageType( + "FreeWriterResponse", + (_message.Message,), + { + "DESCRIPTOR": _FREEWRITERRESPONSE, + "__module__": "temporalio.common._streams.proto.v1.runtime_pb2", + # @@protoc_insertion_point(class_scope:temporal.sdk.streams.v1.FreeWriterResponse) + }, +) +_sym_db.RegisterMessage(FreeWriterResponse) + +CloseWritersRequest = _reflection.GeneratedProtocolMessageType( + "CloseWritersRequest", + (_message.Message,), + { + "DESCRIPTOR": _CLOSEWRITERSREQUEST, + "__module__": "temporalio.common._streams.proto.v1.runtime_pb2", + # @@protoc_insertion_point(class_scope:temporal.sdk.streams.v1.CloseWritersRequest) + }, +) +_sym_db.RegisterMessage(CloseWritersRequest) + +CloseWritersResponse = _reflection.GeneratedProtocolMessageType( + "CloseWritersResponse", + (_message.Message,), + { + "DESCRIPTOR": _CLOSEWRITERSRESPONSE, + "__module__": "temporalio.common._streams.proto.v1.runtime_pb2", + # @@protoc_insertion_point(class_scope:temporal.sdk.streams.v1.CloseWritersResponse) + }, +) +_sym_db.RegisterMessage(CloseWritersResponse) + +if _descriptor._USE_C_DESCRIPTORS == False: + DESCRIPTOR._options = None + _MEMORYSTREAMCONFIG._serialized_start = 137 + _MEMORYSTREAMCONFIG._serialized_end = 267 + _OPENWRITERREQUEST._serialized_start = 270 + _OPENWRITERREQUEST._serialized_end = 588 + _OPENWRITERRESPONSE._serialized_start = 590 + _OPENWRITERRESPONSE._serialized_end = 629 + _WRITERECORDSREQUEST._serialized_start = 631 + _WRITERECORDSREQUEST._serialized_end = 726 + _WRITERECORDSRESPONSE._serialized_start = 728 + _WRITERECORDSRESPONSE._serialized_end = 793 + _CLOSEWRITERREQUEST._serialized_start = 795 + _CLOSEWRITERREQUEST._serialized_end = 834 + _CLOSEWRITERRESPONSE._serialized_start = 836 + _CLOSEWRITERRESPONSE._serialized_end = 878 + _FREEWRITERREQUEST._serialized_start = 880 + _FREEWRITERREQUEST._serialized_end = 918 + _FREEWRITERRESPONSE._serialized_start = 920 + _FREEWRITERRESPONSE._serialized_end = 940 + _CLOSEWRITERSREQUEST._serialized_start = 942 + _CLOSEWRITERSREQUEST._serialized_end = 963 + _CLOSEWRITERSRESPONSE._serialized_start = 965 + _CLOSEWRITERSRESPONSE._serialized_end = 1003 +# @@protoc_insertion_point(module_scope) diff --git a/temporalio/common/_streams/proto/v1/runtime_pb2.pyi b/temporalio/common/_streams/proto/v1/runtime_pb2.pyi new file mode 100644 index 0000000000..b36671c579 --- /dev/null +++ b/temporalio/common/_streams/proto/v1/runtime_pb2.pyi @@ -0,0 +1,335 @@ +""" +@generated by mypy-protobuf. Do not edit manually! +isort:skip_file +The messages of the crate's own calls, in addition to the five Stream service operations. Lang +sends them through the bridge's call on the runtime object, by the method name in each comment. +Field numbers are part of the contract with every SDK. +""" + +import builtins +import collections.abc +import sys + +import google.protobuf.descriptor +import google.protobuf.duration_pb2 +import google.protobuf.internal.containers +import google.protobuf.message + +import temporalio.api.stream.v1.message_pb2 + +if sys.version_info >= (3, 8): + import typing as typing_extensions +else: + import typing_extensions + +DESCRIPTOR: google.protobuf.descriptor.FileDescriptor + +class MemoryStreamConfig(google.protobuf.message.Message): + """The settings of the memory Stream runtime, for tests. Each field has the number of the same + field in the config of the other Stores, such as Redis, so a backend can send that config here + too. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + RETENTION_FIELD_NUMBER: builtins.int + SAFETY_TTL_FIELD_NUMBER: builtins.int + BACKEND_FIELD_NUMBER: builtins.int + @property + def retention(self) -> google.protobuf.duration_pb2.Duration: + """The default Retention of a new Stream. Empty is 7 days.""" + @property + def safety_ttl(self) -> google.protobuf.duration_pb2.Duration: + """How long an open Workflow-owned Stream stays after its last write. Empty is 90 days.""" + backend: builtins.str + """The backend name that lang registered this runtime under. The crate sets it on each + StreamRef that it builds, such as the one in a create response. + """ + def __init__( + self, + *, + retention: google.protobuf.duration_pb2.Duration | None = ..., + safety_ttl: google.protobuf.duration_pb2.Duration | None = ..., + backend: builtins.str = ..., + ) -> None: ... + def HasField( + self, + field_name: typing_extensions.Literal[ + "retention", b"retention", "safety_ttl", b"safety_ttl" + ], + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "backend", + b"backend", + "retention", + b"retention", + "safety_ttl", + b"safety_ttl", + ], + ) -> None: ... + +global___MemoryStreamConfig = MemoryStreamConfig + +class OpenWriterRequest(google.protobuf.message.Message): + """Method `OpenWriter`, asynchronous. A writer for an Activity or a client Producer, which keeps its + epoch and `sequence` order. A writer of an Activity attempt above 1 first sends an empty batch, + so that the new epoch withdraws the earlier attempt's Records. The handles of one Producer epoch + on one Stream share one writer. Each open adds a handle, and each `CloseWriter` or `FreeWriter` + releases one. The batches of every writer in a process take their `sequence` from one counter, so + a writer that opens again after its close continues above its earlier batches. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + NAMESPACE_FIELD_NUMBER: builtins.int + STREAM_FIELD_NUMBER: builtins.int + PRODUCER_FIELD_NUMBER: builtins.int + MAX_BATCH_SIZE_FIELD_NUMBER: builtins.int + MAX_RETRY_DURATION_FIELD_NUMBER: builtins.int + PREVIOUS_COMPLETED_TASK_FIELD_NUMBER: builtins.int + FAILED_TASKS_FIELD_NUMBER: builtins.int + namespace: builtins.str + @property + def stream(self) -> temporalio.api.stream.v1.message_pb2.StreamRef: ... + @property + def producer(self) -> temporalio.api.stream.v1.message_pb2.Producer: ... + max_batch_size: builtins.int + """A write sends at most this many Records in one batch. Zero is 1000.""" + @property + def max_retry_duration(self) -> google.protobuf.duration_pb2.Duration: + """How long a write retries a Store failure. Empty is 10 s.""" + @property + def previous_completed_task( + self, + ) -> temporalio.api.stream.v1.message_pb2.WorkflowTaskEpoch: + """Local Activity only. The run's task facts, which each batch sends as AppendRecords does, so + the writes withdraw the run's orphaned Local Activity Records. + """ + @property + def failed_tasks( + self, + ) -> google.protobuf.internal.containers.RepeatedScalarFieldContainer[ + builtins.int + ]: ... + def __init__( + self, + *, + namespace: builtins.str = ..., + stream: temporalio.api.stream.v1.message_pb2.StreamRef | None = ..., + producer: temporalio.api.stream.v1.message_pb2.Producer | None = ..., + max_batch_size: builtins.int = ..., + max_retry_duration: google.protobuf.duration_pb2.Duration | None = ..., + previous_completed_task: temporalio.api.stream.v1.message_pb2.WorkflowTaskEpoch + | None = ..., + failed_tasks: collections.abc.Iterable[builtins.int] | None = ..., + ) -> None: ... + def HasField( + self, + field_name: typing_extensions.Literal[ + "max_retry_duration", + b"max_retry_duration", + "previous_completed_task", + b"previous_completed_task", + "producer", + b"producer", + "stream", + b"stream", + ], + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "failed_tasks", + b"failed_tasks", + "max_batch_size", + b"max_batch_size", + "max_retry_duration", + b"max_retry_duration", + "namespace", + b"namespace", + "previous_completed_task", + b"previous_completed_task", + "producer", + b"producer", + "stream", + b"stream", + ], + ) -> None: ... + +global___OpenWriterRequest = OpenWriterRequest + +class OpenWriterResponse(google.protobuf.message.Message): + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + WRITER_ID_FIELD_NUMBER: builtins.int + writer_id: builtins.int + def __init__( + self, + *, + writer_id: builtins.int = ..., + ) -> None: ... + def ClearField( + self, field_name: typing_extensions.Literal["writer_id", b"writer_id"] + ) -> None: ... + +global___OpenWriterResponse = OpenWriterResponse + +class WriteRecordsRequest(google.protobuf.message.Message): + """Method `WriteRecords`, asynchronous. Appends Records at once for an open writer and responds when + the Store has them. It uses the writer's Producer epoch and `sequence` order. After a Store + failure it resends the same batch, so that the Store writes it once. Records past the Store's + batch limit go in more batches, in order. An awaited publish of an Activity or a client uses this + call. A failure after the retries fails the writer, since the batch may be in the Store. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + WRITER_ID_FIELD_NUMBER: builtins.int + RECORDS_FIELD_NUMBER: builtins.int + writer_id: builtins.int + @property + def records( + self, + ) -> google.protobuf.internal.containers.RepeatedCompositeFieldContainer[ + temporalio.api.stream.v1.message_pb2.AppendRecord + ]: ... + def __init__( + self, + *, + writer_id: builtins.int = ..., + records: collections.abc.Iterable[ + temporalio.api.stream.v1.message_pb2.AppendRecord + ] + | None = ..., + ) -> None: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "records", b"records", "writer_id", b"writer_id" + ], + ) -> None: ... + +global___WriteRecordsRequest = WriteRecordsRequest + +class WriteRecordsResponse(google.protobuf.message.Message): + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + FIRST_OFFSET_FIELD_NUMBER: builtins.int + LAST_OFFSET_FIELD_NUMBER: builtins.int + first_offset: builtins.int + """The offsets of the first and the last Record. Zero when the request has no Record.""" + last_offset: builtins.int + def __init__( + self, + *, + first_offset: builtins.int = ..., + last_offset: builtins.int = ..., + ) -> None: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "first_offset", b"first_offset", "last_offset", b"last_offset" + ], + ) -> None: ... + +global___WriteRecordsResponse = WriteRecordsResponse + +class CloseWriterRequest(google.protobuf.message.Message): + """Method `CloseWriter`, asynchronous. Releases one handle of the writer. The last handle waits for + a write in flight, then drops the writer. It doesn't close the Stream. It fails with the + writer's error if a write failed. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + WRITER_ID_FIELD_NUMBER: builtins.int + writer_id: builtins.int + def __init__( + self, + *, + writer_id: builtins.int = ..., + ) -> None: ... + def ClearField( + self, field_name: typing_extensions.Literal["writer_id", b"writer_id"] + ) -> None: ... + +global___CloseWriterRequest = CloseWriterRequest + +class CloseWriterResponse(google.protobuf.message.Message): + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + LAST_OFFSET_FIELD_NUMBER: builtins.int + last_offset: builtins.int + """The offset of the newest Record the writer wrote. Zero if it wrote none.""" + def __init__( + self, + *, + last_offset: builtins.int = ..., + ) -> None: ... + def ClearField( + self, field_name: typing_extensions.Literal["last_offset", b"last_offset"] + ) -> None: ... + +global___CloseWriterResponse = CloseWriterResponse + +class FreeWriterRequest(google.protobuf.message.Message): + """Method `FreeWriter`. Releases one handle of the writer with no wait, for a handle that lang + drops, such as from a finalizer. The bridge may also call it with no async runtime. A write in + flight still completes. The writer's error, if any, is lost. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + WRITER_ID_FIELD_NUMBER: builtins.int + writer_id: builtins.int + def __init__( + self, + *, + writer_id: builtins.int = ..., + ) -> None: ... + def ClearField( + self, field_name: typing_extensions.Literal["writer_id", b"writer_id"] + ) -> None: ... + +global___FreeWriterRequest = FreeWriterRequest + +class FreeWriterResponse(google.protobuf.message.Message): + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + def __init__( + self, + ) -> None: ... + +global___FreeWriterResponse = FreeWriterResponse + +class CloseWritersRequest(google.protobuf.message.Message): + """Method `CloseWriters`, asynchronous. Drops every open writer, for example when the backend's + client or Worker shuts down. + """ + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + def __init__( + self, + ) -> None: ... + +global___CloseWritersRequest = CloseWritersRequest + +class CloseWritersResponse(google.protobuf.message.Message): + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + FAILED_FIELD_NUMBER: builtins.int + failed: builtins.int + """The writers that had failed.""" + def __init__( + self, + *, + failed: builtins.int = ..., + ) -> None: ... + def ClearField( + self, field_name: typing_extensions.Literal["failed", b"failed"] + ) -> None: ... + +global___CloseWritersResponse = CloseWritersResponse