From bb2675db9d09029f74f061376215308b266e5f18 Mon Sep 17 00:00:00 2001 From: Moe Dashti Date: Tue, 27 Oct 2026 11:20:02 -0700 Subject: [PATCH 1/2] fix: plan a oneof case that names a proto-bound record as that record Variant planning kept such a case as the raw proto message. Python and TypeScript then panicked on a message conversion that wasn't model-shaped, and .NET typed the case as the proto class. The case plans as the record, as a record field does. A .NET case record names its payload's model in full, since it hides a model of its own name. Brook's stream service generates for TypeScript, and its .NET read item holds the records. --- GUIDE.md | 3 + advanced/samples/README.md | 9 +- .../dotnet/tests/StreamServiceChecks.cs | 53 + .../dotnet/wit/stream-service/Models.cs | 303 ++++- advanced/samples/inputs/examples.toml | 2 +- .../typescript/tests/stream-service.test.ts | 89 ++ advanced/samples/typescript/tsconfig.json | 3 +- .../typescript/wit/stream-service/index.ts | 19 + .../wit/stream-service/interceptors.ts | 98 ++ .../typescript/wit/stream-service/models.ts | 1071 +++++++++++++++++ .../operations/append-records.ts | 20 + .../stream-service/operations/close-stream.ts | 18 + .../operations/create-stream.ts | 36 + .../operations/describe-stream.ts | 18 + .../stream-service/operations/read-records.ts | 23 + .../typescript/wit/stream-service/registry.ts | 24 + .../wit/stream-service/resources.ts | 61 + .../typescript/wit/stream-service/services.ts | 71 ++ changelog/fixed/lantern-otter.md | 1 + src/generator/dotnet.rs | 64 +- src/planning/type_planning.rs | 40 + 21 files changed, 2007 insertions(+), 19 deletions(-) create mode 100644 advanced/samples/typescript/tests/stream-service.test.ts create mode 100644 advanced/samples/typescript/wit/stream-service/index.ts create mode 100644 advanced/samples/typescript/wit/stream-service/interceptors.ts create mode 100644 advanced/samples/typescript/wit/stream-service/models.ts create mode 100644 advanced/samples/typescript/wit/stream-service/operations/append-records.ts create mode 100644 advanced/samples/typescript/wit/stream-service/operations/close-stream.ts create mode 100644 advanced/samples/typescript/wit/stream-service/operations/create-stream.ts create mode 100644 advanced/samples/typescript/wit/stream-service/operations/describe-stream.ts create mode 100644 advanced/samples/typescript/wit/stream-service/operations/read-records.ts create mode 100644 advanced/samples/typescript/wit/stream-service/registry.ts create mode 100644 advanced/samples/typescript/wit/stream-service/resources.ts create mode 100644 advanced/samples/typescript/wit/stream-service/services.ts create mode 100644 changelog/fixed/lantern-otter.md diff --git a/GUIDE.md b/GUIDE.md index 089080ee0..653238de8 100644 --- a/GUIDE.md +++ b/GUIDE.md @@ -1509,6 +1509,9 @@ record response { } ``` +A case may name a proto-bound record. Its payload is that record's model, which +converts with the record's own converter. + Python represents a protobuf oneof-backed variant as a union of slotted case dataclasses. Each case contains its payload in a `value` field. For example, the variant above generates `OutcomeSuccess(value)` and `OutcomeFailure(value)`. The diff --git a/advanced/samples/README.md b/advanced/samples/README.md index 184ed79d7..7ff2e9f08 100644 --- a/advanced/samples/README.md +++ b/advanced/samples/README.md @@ -163,13 +163,16 @@ crate omits from the API of a resource method's request is left out of the model. nexgen ignores the directives that it doesn't know yet, such as `@nexus.varargs` and `@nexus.close`. -Python and TypeScript don't generate the service. They panic on the read item, a -`oneof` of proto-bound records, and Python can't import the protos. .NET types -each case of the read item as its proto class. +Python doesn't generate the service, since it can't import the protos. +TypeScript passes a `number` where a resource method's proto request wants a +`Long`. So `tsc` skips the service, but the test still checks its models. - WIT: [`inputs/stream-service.wit`](inputs/stream-service.wit) - .NET: [`dotnet/wit/stream-service/`](dotnet/wit/stream-service/), [`dotnet/tests/StreamServiceChecks.cs`](dotnet/tests/StreamServiceChecks.cs) +- TypeScript: + [`typescript/wit/stream-service/`](typescript/wit/stream-service/), + [`typescript/tests/stream-service.test.ts`](typescript/tests/stream-service.test.ts) ## Supporting files diff --git a/advanced/samples/dotnet/tests/StreamServiceChecks.cs b/advanced/samples/dotnet/tests/StreamServiceChecks.cs index d3253d856..a7f3ef5c7 100644 --- a/advanced/samples/dotnet/tests/StreamServiceChecks.cs +++ b/advanced/samples/dotnet/tests/StreamServiceChecks.cs @@ -1,8 +1,12 @@ extern alias Streams; +using System.Linq; using Temporalio.Converters; using Xunit; +using static Nexgen.DotNetExamples.Tests.TransferTypeTestSupport; using StreamServiceExample = Streams::Temporalio.Workflows; +using WireReadRecordsResponse = Streams::Temporalio.Api.StreamService.V1.ReadRecordsResponse; +using WireStreamRecord = Streams::Temporalio.Api.Stream.V1.StreamRecord; namespace Nexgen.DotNetExamples.Tests { @@ -18,5 +22,54 @@ public void StreamWithItsOwnFieldNameRoundTrips() Assert.Equal("stream-1", decoded.StreamValue.Id); } + + // A data item's body goes through the payload converter of the sample assembly, which a + // test can't set. The TypeScript test covers data items. + [Fact] + public void ReadItemsRoundTripAsTheirRecords() + { + var model = new StreamServiceExample.ReadRecordsResponse(new[] + { + new StreamServiceExample.StreamRecord + { + Item = new StreamServiceExample.StreamItem.Superseded( + new StreamServiceExample.Supersession("workflow-1", new long[] { 7 })), + Offset = 2, + }, + new StreamServiceExample.StreamRecord + { + Item = new StreamServiceExample.StreamItem.Close( + new StreamServiceExample.StreamClose + { + Status = StreamServiceExample.StreamCloseStatus + .StreamCloseStatusCompleted, + }), + Offset = 3, + }, + }); + + var wire = Assert.IsType( + CreateSdkConverter(typeof(StreamServiceExample.ReadRecordsResponse)) + .ToTransferType(model)); + Assert.Equal( + new[] + { + WireStreamRecord.ItemOneofCase.Superseded, + WireStreamRecord.ItemOneofCase.Close, + }, + wire.Records.Select(record => record.ItemCase)); + + var decoded = FromPayload(ToPayload(model)); + var superseded = Assert.IsType( + decoded.Records[0].Item); + Assert.Equal("workflow-1", superseded.Value.ProducerId); + Assert.Equal(new long[] { 7 }, superseded.Value.FailedTasks); + var close = Assert.IsType( + decoded.Records[1].Item); + Assert.Equal( + StreamServiceExample.StreamCloseStatus.StreamCloseStatusCompleted, + close.Value.Status); + Assert.Equal(3, decoded.Records[1].Offset); + } } } diff --git a/advanced/samples/dotnet/wit/stream-service/Models.cs b/advanced/samples/dotnet/wit/stream-service/Models.cs index 28c9a743d..f07253c58 100644 --- a/advanced/samples/dotnet/wit/stream-service/Models.cs +++ b/advanced/samples/dotnet/wit/stream-service/Models.cs @@ -19,6 +19,17 @@ public enum ExecutionType ExecutionTypeNexusOperation = 3, } + [GeneratedCode("nexgen", null)] + public enum StreamCloseStatus + { + StreamCloseStatusUnspecified = 0, + StreamCloseStatusCompleted = 1, + StreamCloseStatusFailed = 2, + StreamCloseStatusCanceled = 3, + StreamCloseStatusTerminated = 4, + StreamCloseStatusTimedOut = 5, + } + [GeneratedCode("nexgen", null)] public enum StreamProducerKind { @@ -56,13 +67,13 @@ public abstract record StreamItem private StreamItem() { } [GeneratedCode("nexgen", null)] - public sealed record Data(Temporalio.Api.Stream.V1.StreamData Value) : StreamItem; + public sealed record Data(StreamData Value) : StreamItem; [GeneratedCode("nexgen", null)] - public sealed record Superseded(Temporalio.Api.Stream.V1.Supersession Value) : StreamItem; + public sealed record Superseded(Supersession Value) : StreamItem; [GeneratedCode("nexgen", null)] - public sealed record Close(Temporalio.Api.Stream.V1.StreamClose Value) : StreamItem; + public sealed record Close(StreamClose Value) : StreamItem; } /// @@ -943,6 +954,131 @@ public sealed class TransferTypeConverter : Temporalio.Converters.ITemporalTrans } + /// + /// The close of a Stream. A read gives it last. + /// + [Temporalio.Converters.TemporalTransferTypeConverter(typeof(StreamClose.TransferTypeConverter))] + [GeneratedCode("nexgen", null)] + public record StreamClose + { + /// + /// The Producer that closed the Stream. + /// + public Producer? Producer { get; init; } + public object? Result { get; init; } + public System.Exception? Failure { get; init; } + /// + /// Set together with the result or the failure. + /// + public StreamCloseStatus? Status { get; init; } + + internal static StreamClose FromTransferType(Temporalio.Api.Stream.V1.StreamClose wire) + { + return new StreamClose() + { + Producer = wire.Producer == null ? null : Producer.FromTransferType(wire.Producer), + Result = wire.Result == null ? null : Nexgen.Support.ProtoExtensions.FromPayload(wire.Result), + Failure = wire.Failure == null ? null : Nexgen.Support.ProtoExtensions.FromFailureProto(wire.Failure), + Status = (int)wire.Status == 0 ? null : (StreamCloseStatus)wire.Status, + }; + } + + internal Temporalio.Api.Stream.V1.StreamClose ToTransferType() + { + var proto = new Temporalio.Api.Stream.V1.StreamClose(); + if (Producer is { } producer) + { + proto.Producer = (Temporalio.Api.Stream.V1.Producer)producer.ToTransferType(); + } + if (Result is { } result) + { + proto.Result = Nexgen.Support.ProtoExtensions.ToPayload(result); + } + if (Failure is { } failure) + { + proto.Failure = Nexgen.Support.ProtoExtensions.ToFailureProto(failure); + } + if (Status is { } status) + { + proto.Status = (Temporalio.Api.Enums.V1.StreamCloseStatus)status; + } + return proto; + } + + /// + /// Converts this model to and from its generated transfer type. + /// + public sealed class TransferTypeConverter : Temporalio.Converters.ITemporalTransferTypeConverter + { + /// Gets the generated transfer type. + public System.Type TransferType => typeof(Temporalio.Api.Stream.V1.StreamClose); + + /// Converts a model value to its transfer type. + public object? ToTransferType(object? value) => value is null ? null : ((StreamClose)value).ToTransferType(); + + /// Converts a transfer-type value to this model. + public object? FromTransferType(object? transferType) => transferType is null ? null : StreamClose.FromTransferType((Temporalio.Api.Stream.V1.StreamClose)transferType); + } + + } + + /// + /// One Record that a Producer appended. + /// + [Temporalio.Converters.TemporalTransferTypeConverter(typeof(StreamData.TransferTypeConverter))] + [GeneratedCode("nexgen", null)] + public record StreamData + { + /// + /// Initializes a new instance of the class. + /// + public StreamData(object? body) + { + Body = body; + } + + /// + /// The value. + /// + public object? Body { get; init; } + public Producer? Producer { get; init; } + + internal static StreamData FromTransferType(Temporalio.Api.Stream.V1.StreamData wire) + { + return new StreamData(Nexgen.Support.ProtoExtensions.FromPayload(wire.Body)) + { + Producer = wire.Producer == null ? null : Producer.FromTransferType(wire.Producer), + }; + } + + internal Temporalio.Api.Stream.V1.StreamData ToTransferType() + { + var proto = new Temporalio.Api.Stream.V1.StreamData(); + proto.Body = Nexgen.Support.ProtoExtensions.ToPayload(Body); + if (Producer is { } producer) + { + proto.Producer = (Temporalio.Api.Stream.V1.Producer)producer.ToTransferType(); + } + return proto; + } + + /// + /// Converts this model to and from its generated transfer type. + /// + public sealed class TransferTypeConverter : Temporalio.Converters.ITemporalTransferTypeConverter + { + /// Gets the generated transfer type. + public System.Type TransferType => typeof(Temporalio.Api.Stream.V1.StreamData); + + /// Converts a model value to its transfer type. + public object? ToTransferType(object? value) => value is null ? null : ((StreamData)value).ToTransferType(); + + /// Converts a transfer-type value to this model. + public object? FromTransferType(object? transferType) => transferType is null ? null : StreamData.FromTransferType((Temporalio.Api.Stream.V1.StreamData)transferType); + } + + } + /// /// One item that a read gives. /// @@ -962,13 +1098,13 @@ internal static StreamRecord FromTransferType(Temporalio.Api.Stream.V1.StreamRec switch (wire.ItemCase) { case Temporalio.Api.Stream.V1.StreamRecord.ItemOneofCase.Data: - itemOneof = new StreamItem.Data(wire.Data); + itemOneof = new StreamItem.Data(StreamData.FromTransferType(wire.Data)); break; case Temporalio.Api.Stream.V1.StreamRecord.ItemOneofCase.Superseded: - itemOneof = new StreamItem.Superseded(wire.Superseded); + itemOneof = new StreamItem.Superseded(Supersession.FromTransferType(wire.Superseded)); break; case Temporalio.Api.Stream.V1.StreamRecord.ItemOneofCase.Close: - itemOneof = new StreamItem.Close(wire.Close); + itemOneof = new StreamItem.Close(StreamClose.FromTransferType(wire.Close)); break; default: itemOneof = null; @@ -988,13 +1124,13 @@ internal Temporalio.Api.Stream.V1.StreamRecord ToTransferType() switch (Item) { case StreamItem.Data dataCase: - proto.Data = dataCase.Value; + proto.Data = (Temporalio.Api.Stream.V1.StreamData)dataCase.Value.ToTransferType(); break; case StreamItem.Superseded supersededCase: - proto.Superseded = supersededCase.Value; + proto.Superseded = (Temporalio.Api.Stream.V1.Supersession)supersededCase.Value.ToTransferType(); break; case StreamItem.Close closeCase: - proto.Close = closeCase.Value; + proto.Close = (Temporalio.Api.Stream.V1.StreamClose)closeCase.Value.ToTransferType(); break; } if (Offset is { } offset) @@ -1090,6 +1226,99 @@ public sealed class TransferTypeConverter : Temporalio.Converters.ITemporalTrans } + /// + /// Which earlier Records a Superseded item withdraws. + /// + [Temporalio.Converters.TemporalTransferTypeConverter(typeof(Supersession.TransferTypeConverter))] + [GeneratedCode("nexgen", null)] + public record Supersession + { + /// + /// Initializes a new instance of the class. + /// + public Supersession(string producerId, IReadOnlyList failedTasks) + { + ProducerId = producerId; + FailedTasks = failedTasks; + } + + public string ProducerId { get; init; } + /// + /// Workflow only. Started event ids of failed or timed out Workflow Tasks. + /// + public IReadOnlyList FailedTasks { get; init; } + public StreamProducerKind? ProducerKind { get; init; } + public ProducerEpoch? Epoch { get; init; } + /// + /// Workflow only. The previous completed Workflow Task of the run. + /// + public WorkflowTaskEpoch? PreviousCompletedTask { get; init; } + /// + /// Workflow retry only. The previous run, all of whose Records are withdrawn. + /// + public string? WithdrawRun { get; init; } + /// + /// The withdrawing Producer's run, so no Reader parses ids. Empty for a client. + /// + public string? RunId { get; init; } + + internal static Supersession FromTransferType(Temporalio.Api.Stream.V1.Supersession wire) + { + return new Supersession(wire.ProducerId, System.Linq.Enumerable.ToArray(System.Linq.Enumerable.Select(wire.FailedTasks, item => item))) + { + ProducerKind = (int)wire.ProducerKind == 0 ? null : (StreamProducerKind)wire.ProducerKind, + Epoch = wire.Epoch == null ? null : ProducerEpoch.FromTransferType(wire.Epoch), + PreviousCompletedTask = wire.PreviousCompletedTask == null ? null : WorkflowTaskEpoch.FromTransferType(wire.PreviousCompletedTask), + WithdrawRun = string.IsNullOrEmpty(wire.WithdrawRun) ? null : wire.WithdrawRun, + RunId = string.IsNullOrEmpty(wire.RunId) ? null : wire.RunId, + }; + } + + internal Temporalio.Api.Stream.V1.Supersession ToTransferType() + { + var proto = new Temporalio.Api.Stream.V1.Supersession(); + proto.ProducerId = ProducerId; + proto.FailedTasks.Add(System.Linq.Enumerable.Select(FailedTasks, item => item)); + if (ProducerKind is { } producerKind) + { + proto.ProducerKind = (Temporalio.Api.Enums.V1.StreamProducerKind)producerKind; + } + if (Epoch is { } epoch) + { + proto.Epoch = (Temporalio.Api.Stream.V1.ProducerEpoch)epoch.ToTransferType(); + } + if (PreviousCompletedTask is { } previousCompletedTask) + { + proto.PreviousCompletedTask = (Temporalio.Api.Stream.V1.WorkflowTaskEpoch)previousCompletedTask.ToTransferType(); + } + if (WithdrawRun is { } withdrawRun) + { + proto.WithdrawRun = withdrawRun; + } + if (RunId is { } runId) + { + proto.RunId = runId; + } + return proto; + } + + /// + /// Converts this model to and from its generated transfer type. + /// + public sealed class TransferTypeConverter : Temporalio.Converters.ITemporalTransferTypeConverter + { + /// Gets the generated transfer type. + public System.Type TransferType => typeof(Temporalio.Api.Stream.V1.Supersession); + + /// Converts a model value to its transfer type. + public object? ToTransferType(object? value) => value is null ? null : ((Supersession)value).ToTransferType(); + + /// Converts a transfer-type value to this model. + public object? FromTransferType(object? transferType) => transferType is null ? null : Supersession.FromTransferType((Temporalio.Api.Stream.V1.Supersession)transferType); + } + + } + [Temporalio.Converters.TemporalTransferTypeConverter(typeof(Timestamp.TransferTypeConverter))] [GeneratedCode("nexgen", null)] public record Timestamp @@ -1137,4 +1366,60 @@ public sealed class TransferTypeConverter : Temporalio.Converters.ITemporalTrans } + /// + /// A Workflow Task attempt of the run. + /// + [Temporalio.Converters.TemporalTransferTypeConverter(typeof(WorkflowTaskEpoch.TransferTypeConverter))] + [GeneratedCode("nexgen", null)] + public record WorkflowTaskEpoch + { + public long? StartedEventId { get; init; } + public int? Attempt { get; init; } + public Timestamp? StartedTime { get; init; } + + internal static WorkflowTaskEpoch FromTransferType(Temporalio.Api.Stream.V1.WorkflowTaskEpoch wire) + { + return new WorkflowTaskEpoch() + { + StartedEventId = wire.StartedEventId == 0 ? null : wire.StartedEventId, + Attempt = wire.Attempt == 0 ? null : wire.Attempt, + StartedTime = wire.StartedTime == null ? null : Timestamp.FromTransferType(wire.StartedTime), + }; + } + + internal Temporalio.Api.Stream.V1.WorkflowTaskEpoch ToTransferType() + { + var proto = new Temporalio.Api.Stream.V1.WorkflowTaskEpoch(); + if (StartedEventId is { } startedEventId) + { + proto.StartedEventId = startedEventId; + } + if (Attempt is { } attempt) + { + proto.Attempt = attempt; + } + if (StartedTime is { } startedTime) + { + proto.StartedTime = (Google.Protobuf.WellKnownTypes.Timestamp)startedTime.ToTransferType(); + } + return proto; + } + + /// + /// Converts this model to and from its generated transfer type. + /// + public sealed class TransferTypeConverter : Temporalio.Converters.ITemporalTransferTypeConverter + { + /// Gets the generated transfer type. + public System.Type TransferType => typeof(Temporalio.Api.Stream.V1.WorkflowTaskEpoch); + + /// Converts a model value to its transfer type. + public object? ToTransferType(object? value) => value is null ? null : ((WorkflowTaskEpoch)value).ToTransferType(); + + /// Converts a transfer-type value to this model. + public object? FromTransferType(object? transferType) => transferType is null ? null : WorkflowTaskEpoch.FromTransferType((Temporalio.Api.Stream.V1.WorkflowTaskEpoch)transferType); + } + + } + } diff --git a/advanced/samples/inputs/examples.toml b/advanced/samples/inputs/examples.toml index c2c8b6709..e80804baf 100644 --- a/advanced/samples/inputs/examples.toml +++ b/advanced/samples/inputs/examples.toml @@ -29,7 +29,7 @@ descriptors = "temporal" # The Stream service WIT of temporalio-streams, on the Stream protos of temporalio/api#895 at # 44519d8a. Its .NET output builds in Nexgen.DotNetStreamService.csproj. [stream-service] -languages = ["dotnet"] +languages = ["typescript", "dotnet"] system_nexus = true descriptors = ["temporal_api.bin", "streams.bin"] diff --git a/advanced/samples/typescript/tests/stream-service.test.ts b/advanced/samples/typescript/tests/stream-service.test.ts new file mode 100644 index 000000000..680ca8050 --- /dev/null +++ b/advanced/samples/typescript/tests/stream-service.test.ts @@ -0,0 +1,89 @@ +import * as common from "@temporalio/common"; +import type { temporal } from "@temporalio/proto"; +import Long from "long"; +import { describe, expect, test } from "vitest"; + +import { + type ReadRecordsResponse, + StreamCloseStatus, + readRecordsRequestTransferTypeConverter, + readRecordsResponseTransferTypeConverter, +} from "../wit/stream-service/models.ts"; + +describe("stream-service generated output", () => { + // No SDK has these protos yet, so proto/streams.d.ts declares them. The converters fail to + // compile when those declarations drift from what they read and write. + test("maps each read item kind through the proto oneof", () => { + const model: ReadRecordsResponse = { + records: [ + { + item: { + tag: "data", + value: { + body: common.defaultPayloadConverter.toPayload("hi"), + }, + }, + offset: Long.fromNumber(1), + }, + { + item: { + tag: "superseded", + value: { producerId: "workflow-1", failedTasks: [Long.fromNumber(7)] }, + }, + offset: Long.fromNumber(2), + }, + { + item: { + tag: "close", + value: { status: StreamCloseStatus.StreamCloseStatusCompleted }, + }, + offset: Long.fromNumber(3), + }, + ], + nextOffset: Long.fromNumber(4), + done: true, + }; + + const proto: temporal.api.streamservice.v1.IReadRecordsResponse = + readRecordsResponseTransferTypeConverter.toTransferType(model); + + expect(proto.records?.map((record) => record.item)).toEqual([ + "data", + "superseded", + "close", + ]); + expect(proto.records?.[1].superseded?.failedTasks?.[0].toNumber()).toBe(7); + expect(readRecordsResponseTransferTypeConverter.fromTransferType(proto)).toEqual( + model, + ); + }); + + test("gives no item for a read record whose item kind only a newer server knows", () => { + const proto: temporal.api.streamservice.v1.IReadRecordsResponse = { + records: [{ offset: Long.fromNumber(1) }], + }; + + const model = readRecordsResponseTransferTypeConverter.fromTransferType(proto); + + expect(model.records?.[0].item).toBeUndefined(); + expect(model.records?.[0].offset?.toNumber()).toBe(1); + }); + + test("maps every field of a read request through its proto", () => { + const proto: temporal.api.streamservice.v1.IReadRecordsRequest = { + stream: { id: "stream-1", owner: { businessId: "workflow-1" } }, + fromOffset: Long.fromNumber(7), + maxRecords: 10, + }; + + const model = readRecordsRequestTransferTypeConverter.fromTransferType(proto); + + expect(model.stream.id).toBe("stream-1"); + expect(model.stream.owner?.businessId).toBe("workflow-1"); + expect(model.fromOffset?.toNumber()).toBe(7); + expect(model.maxRecords).toBe(10); + expect( + readRecordsRequestTransferTypeConverter.toTransferType(model).stream?.id, + ).toBe("stream-1"); + }); +}); diff --git a/advanced/samples/typescript/tsconfig.json b/advanced/samples/typescript/tsconfig.json index 44b6d192e..fe20217fc 100644 --- a/advanced/samples/typescript/tsconfig.json +++ b/advanced/samples/typescript/tsconfig.json @@ -16,5 +16,6 @@ "support/**/*.ts", "tests/**/*.ts", "proto/**/*.d.ts" - ] + ], + "exclude": ["wit/stream-service/**"] } diff --git a/advanced/samples/typescript/wit/stream-service/index.ts b/advanced/samples/typescript/wit/stream-service/index.ts new file mode 100644 index 000000000..7c543d6a0 --- /dev/null +++ b/advanced/samples/typescript/wit/stream-service/index.ts @@ -0,0 +1,19 @@ +// Generated by nexgen latest. DO NOT EDIT! + +export { createStream } from "./operations/create-stream"; +export { appendRecords } from "./operations/append-records"; +export { closeStream } from "./operations/close-stream"; +export { readRecords } from "./operations/read-records"; +export { describeStream } from "./operations/describe-stream"; +export type { + AppendRecordsRequest, + AppendRecordsResponse, + CloseStreamRequest, + CloseStreamResponse, + CreateStreamRequest, + CreateStreamResult, + DescribeStreamRequest, + DescribeStreamResponse, + ReadRecordsRequest, + ReadRecordsResponse, +} from "./models"; diff --git a/advanced/samples/typescript/wit/stream-service/interceptors.ts b/advanced/samples/typescript/wit/stream-service/interceptors.ts new file mode 100644 index 000000000..e72c557ac --- /dev/null +++ b/advanced/samples/typescript/wit/stream-service/interceptors.ts @@ -0,0 +1,98 @@ +// Generated by nexgen latest. DO NOT EDIT! + +import { composeInterceptors } from "../../../interceptor-composition"; +import type * as nexus from "../../../nexus"; +import type { + CreateStreamResult, + CreateStreamRequest, + AppendRecordsResponse, + AppendRecordsRequest, + CloseStreamResponse, + CloseStreamRequest, + ReadRecordsResponse, + ReadRecordsRequest, + DescribeStreamResponse, + DescribeStreamRequest, +} from "./models"; + +export interface SystemNexusWorkflowOutboundCallsInterceptor { + createStream?: ( + input: CreateStreamRequest, + next: ( + input: CreateStreamRequest, + ) => Promise>, + ) => Promise>; + appendRecords?: ( + input: AppendRecordsRequest, + next: ( + input: AppendRecordsRequest, + ) => Promise>, + ) => Promise>; + closeStream?: ( + input: CloseStreamRequest, + next: ( + input: CloseStreamRequest, + ) => Promise>, + ) => Promise>; + readRecords?: ( + input: ReadRecordsRequest, + next: ( + input: ReadRecordsRequest, + ) => Promise>, + ) => Promise>; + describeStream?: ( + input: DescribeStreamRequest, + next: ( + input: DescribeStreamRequest, + ) => Promise>, + ) => Promise>; +} + +/** Dispatches the operation-specific interceptor chain selected by service and operation. */ +export function dispatchSystemNexusSpecificInterceptors( + service: string, + operation: string, + interceptors: SystemNexusWorkflowOutboundCallsInterceptor[], + input: unknown, + next: (input: Input) => Promise>, +): Promise> { + switch (service) { + case "temporal.api.streamservice.v1.StreamService": + switch (operation) { + case "CreateStream": + return composeInterceptors(interceptors, "createStream", (request) => + next(request), + )(input as CreateStreamRequest) as Promise< + nexus.NexusOperationHandle + >; + case "AppendRecords": + return composeInterceptors(interceptors, "appendRecords", (request) => + next(request), + )(input as AppendRecordsRequest) as Promise< + nexus.NexusOperationHandle + >; + case "CloseStream": + return composeInterceptors(interceptors, "closeStream", (request) => + next(request), + )(input as CloseStreamRequest) as Promise< + nexus.NexusOperationHandle + >; + case "ReadRecords": + return composeInterceptors(interceptors, "readRecords", (request) => + next(request), + )(input as ReadRecordsRequest) as Promise< + nexus.NexusOperationHandle + >; + case "DescribeStream": + return composeInterceptors(interceptors, "describeStream", (request) => + next(request), + )(input as DescribeStreamRequest) as Promise< + nexus.NexusOperationHandle + >; + default: + return next(input); + } + default: + return next(input); + } +} diff --git a/advanced/samples/typescript/wit/stream-service/models.ts b/advanced/samples/typescript/wit/stream-service/models.ts new file mode 100644 index 000000000..826a3fd50 --- /dev/null +++ b/advanced/samples/typescript/wit/stream-service/models.ts @@ -0,0 +1,1071 @@ +// Generated by nexgen latest. DO NOT EDIT! + +import type * as common from "@temporalio/common"; +import type { google, temporal } from "@temporalio/proto"; +import type Long from "long"; +import * as support from "../../support/temporal_model_converters"; + +function requiredField( + value: T | null | undefined, + owner: string, + field: string, +): T { + if (value == null) { + throw new Error(`missing required field ${owner}.${field}`); + } + return value; +} + +export enum ExecutionType { + ExecutionTypeUnspecified = 0, + ExecutionTypeWorkflow = 1, + ExecutionTypeActivity = 2, + ExecutionTypeNexusOperation = 3, +} + +export enum StreamProducerKind { + StreamProducerKindUnspecified = 0, + StreamProducerKindWorkflow = 1, + StreamProducerKindLocalActivity = 2, + StreamProducerKindActivity = 3, + StreamProducerKindClient = 4, + StreamProducerKindStandaloneActivity = 5, +} + +export enum StreamCloseStatus { + StreamCloseStatusUnspecified = 0, + StreamCloseStatusCompleted = 1, + StreamCloseStatusFailed = 2, + StreamCloseStatusCanceled = 3, + StreamCloseStatusTerminated = 4, + StreamCloseStatusTimedOut = 5, +} + +export enum StreamReadStart { + StreamReadStartUnspecified = 0, + StreamReadStartBeginning = 1, + StreamReadStartEnd = 2, +} + +export enum StreamStatus { + StreamStatusUnspecified = 0, + StreamStatusOpen = 1, + StreamStatusClosed = 2, + StreamStatusExpired = 3, +} + +/** + * A read item: a Record, a Superseded item or the close. + */ +export type StreamItem = + | { tag: "data"; value: StreamData } + | { tag: "superseded"; value: Supersession } + | { tag: "close"; value: StreamClose }; + +export interface CreateStreamResult { + stream?: StreamRef; + created?: boolean; +} + +function createStreamResultFromProto( + proto: temporal.api.streamservice.v1.ICreateStreamResponse, +): CreateStreamResult { + return { + stream: proto.stream == null ? undefined : streamRefFromProto(proto.stream), + created: proto.created ?? undefined, + }; +} + +function createStreamResultToProto( + model: CreateStreamResult, +): temporal.api.streamservice.v1.ICreateStreamResponse { + return { + stream: model.stream == null ? undefined : streamRefToProto(model.stream), + created: model.created, + }; +} + +export const createStreamResultTransferTypeConverter = { + fromTransferType( + value: temporal.api.streamservice.v1.ICreateStreamResponse, + ): CreateStreamResult { + return createStreamResultFromProto(value); + }, + + toTransferType( + value: CreateStreamResult, + ): temporal.api.streamservice.v1.ICreateStreamResponse { + return createStreamResultToProto(value); + }, +}; +export interface StreamRef { + /** + * The Stream id, 1 to 4096 bytes. + */ + id: string; + /** + * The owning Workflow run chain or Standalone Activity. Empty on a client ref. + */ + owner?: Execution; + /** + * Registered backend name, empty for the default. The Store never reads it. + */ + backend?: string; +} + +function streamRefFromProto(proto: temporal.api.stream.v1.IStreamRef): StreamRef { + return { + id: requiredField(proto.id === "" ? undefined : proto.id, "StreamRef", "id"), + owner: proto.owner == null ? undefined : executionFromProto(proto.owner), + backend: proto.backend ?? undefined, + }; +} + +function streamRefToProto(model: StreamRef): temporal.api.stream.v1.IStreamRef { + return { + id: requiredField(model.id, "StreamRef", "id"), + owner: model.owner == null ? undefined : executionToProto(model.owner), + backend: model.backend, + }; +} + +export const streamRefTransferTypeConverter = { + fromTransferType(value: temporal.api.stream.v1.IStreamRef): StreamRef { + return streamRefFromProto(value); + }, + + toTransferType(value: StreamRef): temporal.api.stream.v1.IStreamRef { + return streamRefToProto(value); + }, +}; +export interface Execution { + type?: ExecutionType; + /** + * The Workflow id, or the Standalone Activity's id. + */ + businessId?: string; + /** + * The run chain's first run id, or the Standalone Activity's run id. + */ + runId?: string; +} + +function executionFromProto(proto: temporal.api.common.v1.IExecution): Execution { + return { + type: proto.type == null ? undefined : (proto.type as number as ExecutionType), + businessId: proto.businessId ?? undefined, + runId: proto.runId ?? undefined, + }; +} + +function executionToProto(model: Execution): temporal.api.common.v1.IExecution { + return { + type: + model.type == null + ? undefined + : (model.type as number as temporal.api.enums.v1.ExecutionType), + businessId: model.businessId, + runId: model.runId, + }; +} + +export const executionTransferTypeConverter = { + fromTransferType(value: temporal.api.common.v1.IExecution): Execution { + return executionFromProto(value); + }, + + toTransferType(value: Execution): temporal.api.common.v1.IExecution { + return executionToProto(value); + }, +}; +export interface CreateStreamRequest { + /** + * The Stream id. Required, and never empty. + */ + id: string; + /** + * The age after which a Record is dropped. Empty is the Store default. + */ + retention?: common.Duration; + /** + * Retry run only. The previous run, whose Records this withdraws. + */ + withdrawRun?: string; + /** + * The Workflow run that owns the Stream. The Workflow transport sets it. + */ + owner?: Producer; + /** + * The ref, whose id is stream-id. A create with an owner needs the ref's owner. + */ + ref?: StreamRef; +} + +function createStreamRequestFromProto( + proto: temporal.api.streamservice.v1.ICreateStreamRequest, +): CreateStreamRequest { + return { + id: requiredField( + proto.streamId === "" ? undefined : proto.streamId, + "CreateStreamRequest", + "id", + ), + retention: + proto.retention == null ? undefined : support.durationFromProto(proto.retention), + withdrawRun: proto.withdrawRun ?? undefined, + owner: proto.owner == null ? undefined : producerFromProto(proto.owner), + ref: proto.ref == null ? undefined : streamRefFromProto(proto.ref), + }; +} + +function createStreamRequestToProto( + model: CreateStreamRequest, +): temporal.api.streamservice.v1.ICreateStreamRequest { + return { + streamId: requiredField(model.id, "CreateStreamRequest", "id"), + retention: + model.retention == null ? undefined : support.durationToProto(model.retention), + withdrawRun: model.withdrawRun, + owner: model.owner == null ? undefined : producerToProto(model.owner), + ref: model.ref == null ? undefined : streamRefToProto(model.ref), + }; +} + +export const createStreamRequestTransferTypeConverter = { + fromTransferType( + value: temporal.api.streamservice.v1.ICreateStreamRequest, + ): CreateStreamRequest { + return createStreamRequestFromProto(value); + }, + + toTransferType( + value: CreateStreamRequest, + ): temporal.api.streamservice.v1.ICreateStreamRequest { + return createStreamRequestToProto(value); + }, +}; +export interface Producer { + /** + * Stable across attempts. The proposal gives the form per kind. + */ + id: string; + kind?: StreamProducerKind; + epoch?: ProducerEpoch; + /** + * The run of the Producer, so nobody parses the id. Empty for a client. + */ + runId?: string; +} + +function producerFromProto(proto: temporal.api.stream.v1.IProducer): Producer { + return { + id: requiredField(proto.id === "" ? undefined : proto.id, "Producer", "id"), + kind: proto.kind == null ? undefined : (proto.kind as number as StreamProducerKind), + epoch: proto.epoch == null ? undefined : producerEpochFromProto(proto.epoch), + runId: proto.runId ?? undefined, + }; +} + +function producerToProto(model: Producer): temporal.api.stream.v1.IProducer { + return { + id: requiredField(model.id, "Producer", "id"), + kind: + model.kind == null + ? undefined + : (model.kind as number as temporal.api.enums.v1.StreamProducerKind), + epoch: model.epoch == null ? undefined : producerEpochToProto(model.epoch), + runId: model.runId, + }; +} + +export const producerTransferTypeConverter = { + fromTransferType(value: temporal.api.stream.v1.IProducer): Producer { + return producerFromProto(value); + }, + + toTransferType(value: Producer): temporal.api.stream.v1.IProducer { + return producerToProto(value); + }, +}; +export interface ProducerEpoch { + /** + * The started event id of the Workflow Task. + */ + taskStartedEventId?: Long; + /** + * The attempt of the Workflow Task. + */ + taskAttempt?: number; + /** + * When the Workflow Task attempt started. It tells a reused started id apart. + */ + taskStartedTime?: Timestamp; + /** + * When the server scheduled this Activity attempt. A reset makes it later. + */ + attemptScheduledTime?: Timestamp; + /** + * The Activity, Local Activity or client attempt. + */ + attempt?: number; +} + +function producerEpochFromProto( + proto: temporal.api.stream.v1.IProducerEpoch, +): ProducerEpoch { + return { + taskStartedEventId: proto.taskStartedEventId ?? undefined, + taskAttempt: proto.taskAttempt ?? undefined, + taskStartedTime: + proto.taskStartedTime == null + ? undefined + : timestampFromProto(proto.taskStartedTime), + attemptScheduledTime: + proto.attemptScheduledTime == null + ? undefined + : timestampFromProto(proto.attemptScheduledTime), + attempt: proto.attempt ?? undefined, + }; +} + +function producerEpochToProto( + model: ProducerEpoch, +): temporal.api.stream.v1.IProducerEpoch { + return { + taskStartedEventId: model.taskStartedEventId, + taskAttempt: model.taskAttempt, + taskStartedTime: + model.taskStartedTime == null + ? undefined + : timestampToProto(model.taskStartedTime), + attemptScheduledTime: + model.attemptScheduledTime == null + ? undefined + : timestampToProto(model.attemptScheduledTime), + attempt: model.attempt, + }; +} + +export const producerEpochTransferTypeConverter = { + fromTransferType(value: temporal.api.stream.v1.IProducerEpoch): ProducerEpoch { + return producerEpochFromProto(value); + }, + + toTransferType(value: ProducerEpoch): temporal.api.stream.v1.IProducerEpoch { + return producerEpochToProto(value); + }, +}; +export interface Timestamp { + seconds?: Long; + nanos?: number; +} + +function timestampFromProto(proto: google.protobuf.ITimestamp): Timestamp { + return { + seconds: proto.seconds ?? undefined, + nanos: proto.nanos ?? undefined, + }; +} + +function timestampToProto(model: Timestamp): google.protobuf.ITimestamp { + return { + seconds: model.seconds, + nanos: model.nanos, + }; +} + +export const timestampTransferTypeConverter = { + fromTransferType(value: google.protobuf.ITimestamp): Timestamp { + return timestampFromProto(value); + }, + + toTransferType(value: Timestamp): google.protobuf.ITimestamp { + return timestampToProto(value); + }, +}; +export interface AppendRecordsResponse { + firstOffset?: Long; + lastOffset?: Long; +} + +function appendRecordsResponseFromProto( + proto: temporal.api.streamservice.v1.IAppendRecordsResponse, +): AppendRecordsResponse { + return { + firstOffset: proto.firstOffset ?? undefined, + lastOffset: proto.lastOffset ?? undefined, + }; +} + +function appendRecordsResponseToProto( + model: AppendRecordsResponse, +): temporal.api.streamservice.v1.IAppendRecordsResponse { + return { + firstOffset: model.firstOffset, + lastOffset: model.lastOffset, + }; +} + +export const appendRecordsResponseTransferTypeConverter = { + fromTransferType( + value: temporal.api.streamservice.v1.IAppendRecordsResponse, + ): AppendRecordsResponse { + return appendRecordsResponseFromProto(value); + }, + + toTransferType( + value: AppendRecordsResponse, + ): temporal.api.streamservice.v1.IAppendRecordsResponse { + return appendRecordsResponseToProto(value); + }, +}; +export interface AppendRecordsRequest { + stream: StreamRef; + records?: AppendRecord[]; + /** + * Required outside a Workflow. In a Workflow the transport sets it. + */ + producer?: Producer; +} + +function appendRecordsRequestFromProto( + proto: temporal.api.streamservice.v1.IAppendRecordsRequest, +): AppendRecordsRequest { + return { + stream: streamRefFromProto( + requiredField(proto.stream, "AppendRecordsRequest", "stream"), + ), + records: proto.records?.map((value: any) => appendRecordFromProto(value)), + producer: proto.producer == null ? undefined : producerFromProto(proto.producer), + }; +} + +function appendRecordsRequestToProto( + model: AppendRecordsRequest, +): temporal.api.streamservice.v1.IAppendRecordsRequest { + return { + stream: streamRefToProto( + requiredField(model.stream, "AppendRecordsRequest", "stream"), + ), + records: model.records?.map((value) => appendRecordToProto(value)), + producer: model.producer == null ? undefined : producerToProto(model.producer), + }; +} + +export const appendRecordsRequestTransferTypeConverter = { + fromTransferType( + value: temporal.api.streamservice.v1.IAppendRecordsRequest, + ): AppendRecordsRequest { + return appendRecordsRequestFromProto(value); + }, + + toTransferType( + value: AppendRecordsRequest, + ): temporal.api.streamservice.v1.IAppendRecordsRequest { + return appendRecordsRequestToProto(value); + }, +}; +export interface AppendRecord { + /** + * The value. + */ + body: common.Payload; +} + +function appendRecordFromProto( + proto: temporal.api.stream.v1.IAppendRecord, +): AppendRecord { + return { + body: requiredField( + support.payloadFromProto(requiredField(proto.body, "AppendRecord", "body")), + "AppendRecord", + "body", + ), + }; +} + +function appendRecordToProto( + model: AppendRecord, +): temporal.api.stream.v1.IAppendRecord { + return { + body: support.payloadToProto(requiredField(model.body, "AppendRecord", "body")), + }; +} + +export const appendRecordTransferTypeConverter = { + fromTransferType(value: temporal.api.stream.v1.IAppendRecord): AppendRecord { + return appendRecordFromProto(value); + }, + + toTransferType(value: AppendRecord): temporal.api.stream.v1.IAppendRecord { + return appendRecordToProto(value); + }, +}; +export type CloseStreamResponse = Record; + +function closeStreamResponseFromProto( + proto: temporal.api.streamservice.v1.ICloseStreamResponse, +): CloseStreamResponse { + return {}; +} + +function closeStreamResponseToProto( + model: CloseStreamResponse, +): temporal.api.streamservice.v1.ICloseStreamResponse { + return {}; +} + +export const closeStreamResponseTransferTypeConverter = { + fromTransferType( + value: temporal.api.streamservice.v1.ICloseStreamResponse, + ): CloseStreamResponse { + return closeStreamResponseFromProto(value); + }, + + toTransferType( + value: CloseStreamResponse, + ): temporal.api.streamservice.v1.ICloseStreamResponse { + return closeStreamResponseToProto(value); + }, +}; +export interface CloseStreamRequest { + stream: StreamRef; + producer?: Producer; + result?: common.Payload; + failure?: Error; +} + +function closeStreamRequestFromProto( + proto: temporal.api.streamservice.v1.ICloseStreamRequest, +): CloseStreamRequest { + return { + stream: streamRefFromProto( + requiredField(proto.stream, "CloseStreamRequest", "stream"), + ), + producer: proto.producer == null ? undefined : producerFromProto(proto.producer), + result: proto.result == null ? undefined : support.payloadFromProto(proto.result), + failure: + proto.failure == null ? undefined : support.errorFromFailure(proto.failure), + }; +} + +function closeStreamRequestToProto( + model: CloseStreamRequest, +): temporal.api.streamservice.v1.ICloseStreamRequest { + return { + stream: streamRefToProto( + requiredField(model.stream, "CloseStreamRequest", "stream"), + ), + producer: model.producer == null ? undefined : producerToProto(model.producer), + result: model.result == null ? undefined : support.payloadToProto(model.result), + failure: model.failure == null ? undefined : support.errorToFailure(model.failure), + }; +} + +export const closeStreamRequestTransferTypeConverter = { + fromTransferType( + value: temporal.api.streamservice.v1.ICloseStreamRequest, + ): CloseStreamRequest { + return closeStreamRequestFromProto(value); + }, + + toTransferType( + value: CloseStreamRequest, + ): temporal.api.streamservice.v1.ICloseStreamRequest { + return closeStreamRequestToProto(value); + }, +}; +export interface ReadRecordsResponse { + records?: StreamRecord[]; + /** + * The from-offset of the next read, which is the newest offset read + 1. + */ + nextOffset?: Long; + /** + * True after the close Record. No later read gives a Record. + */ + done?: boolean; +} + +function readRecordsResponseFromProto( + proto: temporal.api.streamservice.v1.IReadRecordsResponse, +): ReadRecordsResponse { + return { + records: proto.records?.map((value: any) => streamRecordFromProto(value)), + nextOffset: proto.nextOffset ?? undefined, + done: proto.done ?? undefined, + }; +} + +function readRecordsResponseToProto( + model: ReadRecordsResponse, +): temporal.api.streamservice.v1.IReadRecordsResponse { + return { + records: model.records?.map((value) => streamRecordToProto(value)), + nextOffset: model.nextOffset, + done: model.done, + }; +} + +export const readRecordsResponseTransferTypeConverter = { + fromTransferType( + value: temporal.api.streamservice.v1.IReadRecordsResponse, + ): ReadRecordsResponse { + return readRecordsResponseFromProto(value); + }, + + toTransferType( + value: ReadRecordsResponse, + ): temporal.api.streamservice.v1.IReadRecordsResponse { + return readRecordsResponseToProto(value); + }, +}; +export interface StreamRecord { + item?: StreamItem; + /** + * The place of this item in the Stream. Ordered and unique, with gaps. + */ + offset?: Long; +} + +function streamRecordFromProto( + proto: temporal.api.stream.v1.IStreamRecord, +): StreamRecord { + let result: StreamItem | undefined; + switch (proto.item) { + case "data": + result = { + tag: "data", + value: streamDataFromProto(requiredField(proto.data, "StreamRecord", "data")), + }; + break; + case "superseded": + result = { + tag: "superseded", + value: supersessionFromProto( + requiredField(proto.superseded, "StreamRecord", "superseded"), + ), + }; + break; + case "close": + result = { + tag: "close", + value: streamCloseFromProto( + requiredField(proto.close, "StreamRecord", "close"), + ), + }; + break; + case undefined: + result = undefined; + break; + default: + proto.item satisfies never; + throw new Error("invalid oneof discriminator for StreamRecord.item"); + } + return { + item: result, + offset: proto.offset ?? undefined, + }; +} + +function streamRecordToProto( + model: StreamRecord, +): temporal.api.stream.v1.IStreamRecord { + const proto: temporal.api.stream.v1.IStreamRecord = {}; + const result = model.item; + if (result != null) { + switch (result.tag) { + case "data": + proto.item = "data"; + proto.data = streamDataToProto(result.value); + break; + case "superseded": + proto.item = "superseded"; + proto.superseded = supersessionToProto(result.value); + break; + case "close": + proto.item = "close"; + proto.close = streamCloseToProto(result.value); + break; + default: + result satisfies never; + throw new Error("invalid oneof tag for StreamRecord.item"); + } + } + proto.offset = model.offset; + return proto; +} + +export const streamRecordTransferTypeConverter = { + fromTransferType(value: temporal.api.stream.v1.IStreamRecord): StreamRecord { + return streamRecordFromProto(value); + }, + + toTransferType(value: StreamRecord): temporal.api.stream.v1.IStreamRecord { + return streamRecordToProto(value); + }, +}; +export interface StreamData { + /** + * The value. + */ + body: common.Payload; + producer?: Producer; +} + +function streamDataFromProto(proto: temporal.api.stream.v1.IStreamData): StreamData { + return { + body: requiredField( + support.payloadFromProto(requiredField(proto.body, "StreamData", "body")), + "StreamData", + "body", + ), + producer: proto.producer == null ? undefined : producerFromProto(proto.producer), + }; +} + +function streamDataToProto(model: StreamData): temporal.api.stream.v1.IStreamData { + return { + body: support.payloadToProto(requiredField(model.body, "StreamData", "body")), + producer: model.producer == null ? undefined : producerToProto(model.producer), + }; +} + +export const streamDataTransferTypeConverter = { + fromTransferType(value: temporal.api.stream.v1.IStreamData): StreamData { + return streamDataFromProto(value); + }, + + toTransferType(value: StreamData): temporal.api.stream.v1.IStreamData { + return streamDataToProto(value); + }, +}; +export interface Supersession { + producerId: string; + /** + * Workflow only. Started event ids of failed or timed out Workflow Tasks. + */ + failedTasks?: Long[]; + producerKind?: StreamProducerKind; + epoch?: ProducerEpoch; + /** + * Workflow only. The previous completed Workflow Task of the run. + */ + previousCompletedTask?: WorkflowTaskEpoch; + /** + * Workflow retry only. The previous run, all of whose Records are withdrawn. + */ + withdrawRun?: string; + /** + * The withdrawing Producer's run, so no Reader parses ids. Empty for a client. + */ + runId?: string; +} + +function supersessionFromProto( + proto: temporal.api.stream.v1.ISupersession, +): Supersession { + return { + producerId: requiredField( + proto.producerId === "" ? undefined : proto.producerId, + "Supersession", + "producerId", + ), + failedTasks: proto.failedTasks?.slice(), + producerKind: + proto.producerKind == null + ? undefined + : (proto.producerKind as number as StreamProducerKind), + epoch: proto.epoch == null ? undefined : producerEpochFromProto(proto.epoch), + previousCompletedTask: + proto.previousCompletedTask == null + ? undefined + : workflowTaskEpochFromProto(proto.previousCompletedTask), + withdrawRun: proto.withdrawRun ?? undefined, + runId: proto.runId ?? undefined, + }; +} + +function supersessionToProto( + model: Supersession, +): temporal.api.stream.v1.ISupersession { + return { + producerId: requiredField(model.producerId, "Supersession", "producerId"), + failedTasks: model.failedTasks?.slice(), + producerKind: + model.producerKind == null + ? undefined + : (model.producerKind as number as temporal.api.enums.v1.StreamProducerKind), + epoch: model.epoch == null ? undefined : producerEpochToProto(model.epoch), + previousCompletedTask: + model.previousCompletedTask == null + ? undefined + : workflowTaskEpochToProto(model.previousCompletedTask), + withdrawRun: model.withdrawRun, + runId: model.runId, + }; +} + +export const supersessionTransferTypeConverter = { + fromTransferType(value: temporal.api.stream.v1.ISupersession): Supersession { + return supersessionFromProto(value); + }, + + toTransferType(value: Supersession): temporal.api.stream.v1.ISupersession { + return supersessionToProto(value); + }, +}; +export interface WorkflowTaskEpoch { + startedEventId?: Long; + attempt?: number; + startedTime?: Timestamp; +} + +function workflowTaskEpochFromProto( + proto: temporal.api.stream.v1.IWorkflowTaskEpoch, +): WorkflowTaskEpoch { + return { + startedEventId: proto.startedEventId ?? undefined, + attempt: proto.attempt ?? undefined, + startedTime: + proto.startedTime == null ? undefined : timestampFromProto(proto.startedTime), + }; +} + +function workflowTaskEpochToProto( + model: WorkflowTaskEpoch, +): temporal.api.stream.v1.IWorkflowTaskEpoch { + return { + startedEventId: model.startedEventId, + attempt: model.attempt, + startedTime: + model.startedTime == null ? undefined : timestampToProto(model.startedTime), + }; +} + +export const workflowTaskEpochTransferTypeConverter = { + fromTransferType( + value: temporal.api.stream.v1.IWorkflowTaskEpoch, + ): WorkflowTaskEpoch { + return workflowTaskEpochFromProto(value); + }, + + toTransferType(value: WorkflowTaskEpoch): temporal.api.stream.v1.IWorkflowTaskEpoch { + return workflowTaskEpochToProto(value); + }, +}; +export interface StreamClose { + /** + * The Producer that closed the Stream. + */ + producer?: Producer; + result?: common.Payload; + failure?: Error; + /** + * Set together with the result or the failure. + */ + status?: StreamCloseStatus; +} + +function streamCloseFromProto(proto: temporal.api.stream.v1.IStreamClose): StreamClose { + return { + producer: proto.producer == null ? undefined : producerFromProto(proto.producer), + result: proto.result == null ? undefined : support.payloadFromProto(proto.result), + failure: + proto.failure == null ? undefined : support.errorFromFailure(proto.failure), + status: + proto.status == null ? undefined : (proto.status as number as StreamCloseStatus), + }; +} + +function streamCloseToProto(model: StreamClose): temporal.api.stream.v1.IStreamClose { + return { + producer: model.producer == null ? undefined : producerToProto(model.producer), + result: model.result == null ? undefined : support.payloadToProto(model.result), + failure: model.failure == null ? undefined : support.errorToFailure(model.failure), + status: + model.status == null + ? undefined + : (model.status as number as temporal.api.enums.v1.StreamCloseStatus), + }; +} + +export const streamCloseTransferTypeConverter = { + fromTransferType(value: temporal.api.stream.v1.IStreamClose): StreamClose { + return streamCloseFromProto(value); + }, + + toTransferType(value: StreamClose): temporal.api.stream.v1.IStreamClose { + return streamCloseToProto(value); + }, +}; +export interface ReadRecordsRequest { + stream: StreamRef; + /** + * Start offset. Empty uses start. OUT_OF_RANGE past the newest offset + 1. + */ + fromOffset?: Long; + /** + * Where to start when from-offset is empty. Unspecified is the beginning. + */ + start?: StreamReadStart; + /** + * About this many items per page, 1 to 1000. A Superseded item stays with its Record. + */ + maxRecords?: number; + /** + * How long to wait for a new Record before an empty answer. At most 60 s. + */ + wait?: common.Duration; + /** + * How long to wait for an uncreated Stream. Empty fails at once. At most 60 s. + */ + createWait?: common.Duration; + /** + * Read the newest this many Records, oldest first. No from-offset. No wait. + */ + last?: number; + /** + * The maximum Record bytes in a page. A larger Record is still returned alone. + */ + maxBytes?: Long; +} + +function readRecordsRequestFromProto( + proto: temporal.api.streamservice.v1.IReadRecordsRequest, +): ReadRecordsRequest { + return { + stream: streamRefFromProto( + requiredField(proto.stream, "ReadRecordsRequest", "stream"), + ), + fromOffset: proto.fromOffset ?? undefined, + start: proto.start == null ? undefined : (proto.start as number as StreamReadStart), + maxRecords: proto.maxRecords ?? undefined, + wait: proto.wait == null ? undefined : support.durationFromProto(proto.wait), + createWait: + proto.createWait == null + ? undefined + : support.durationFromProto(proto.createWait), + last: proto.last ?? undefined, + maxBytes: proto.maxBytes ?? undefined, + }; +} + +function readRecordsRequestToProto( + model: ReadRecordsRequest, +): temporal.api.streamservice.v1.IReadRecordsRequest { + return { + stream: streamRefToProto( + requiredField(model.stream, "ReadRecordsRequest", "stream"), + ), + fromOffset: model.fromOffset, + start: + model.start == null + ? undefined + : (model.start as number as temporal.api.enums.v1.StreamReadStart), + maxRecords: model.maxRecords, + wait: model.wait == null ? undefined : support.durationToProto(model.wait), + createWait: + model.createWait == null ? undefined : support.durationToProto(model.createWait), + last: model.last, + maxBytes: model.maxBytes, + }; +} + +export const readRecordsRequestTransferTypeConverter = { + fromTransferType( + value: temporal.api.streamservice.v1.IReadRecordsRequest, + ): ReadRecordsRequest { + return readRecordsRequestFromProto(value); + }, + + toTransferType( + value: ReadRecordsRequest, + ): temporal.api.streamservice.v1.IReadRecordsRequest { + return readRecordsRequestToProto(value); + }, +}; +export interface DescribeStreamResponse { + status?: StreamStatus; + /** + * The newest offset, of a Record, a Superseded item or the close. Empty for none. + */ + lastOffset?: Long; + retention?: common.Duration; +} + +function describeStreamResponseFromProto( + proto: temporal.api.streamservice.v1.IDescribeStreamResponse, +): DescribeStreamResponse { + return { + status: proto.status == null ? undefined : (proto.status as number as StreamStatus), + lastOffset: proto.lastOffset ?? undefined, + retention: + proto.retention == null ? undefined : support.durationFromProto(proto.retention), + }; +} + +function describeStreamResponseToProto( + model: DescribeStreamResponse, +): temporal.api.streamservice.v1.IDescribeStreamResponse { + return { + status: + model.status == null + ? undefined + : (model.status as number as temporal.api.enums.v1.StreamStatus), + lastOffset: model.lastOffset, + retention: + model.retention == null ? undefined : support.durationToProto(model.retention), + }; +} + +export const describeStreamResponseTransferTypeConverter = { + fromTransferType( + value: temporal.api.streamservice.v1.IDescribeStreamResponse, + ): DescribeStreamResponse { + return describeStreamResponseFromProto(value); + }, + + toTransferType( + value: DescribeStreamResponse, + ): temporal.api.streamservice.v1.IDescribeStreamResponse { + return describeStreamResponseToProto(value); + }, +}; +export interface DescribeStreamRequest { + stream: StreamRef; +} + +function describeStreamRequestFromProto( + proto: temporal.api.streamservice.v1.IDescribeStreamRequest, +): DescribeStreamRequest { + return { + stream: streamRefFromProto( + requiredField(proto.stream, "DescribeStreamRequest", "stream"), + ), + }; +} + +function describeStreamRequestToProto( + model: DescribeStreamRequest, +): temporal.api.streamservice.v1.IDescribeStreamRequest { + return { + stream: streamRefToProto( + requiredField(model.stream, "DescribeStreamRequest", "stream"), + ), + }; +} + +export const describeStreamRequestTransferTypeConverter = { + fromTransferType( + value: temporal.api.streamservice.v1.IDescribeStreamRequest, + ): DescribeStreamRequest { + return describeStreamRequestFromProto(value); + }, + + toTransferType( + value: DescribeStreamRequest, + ): temporal.api.streamservice.v1.IDescribeStreamRequest { + return describeStreamRequestToProto(value); + }, +}; diff --git a/advanced/samples/typescript/wit/stream-service/operations/append-records.ts b/advanced/samples/typescript/wit/stream-service/operations/append-records.ts new file mode 100644 index 000000000..48e5d1fff --- /dev/null +++ b/advanced/samples/typescript/wit/stream-service/operations/append-records.ts @@ -0,0 +1,20 @@ +// Generated by nexgen latest. DO NOT EDIT! + +import * as nexus from "../../../../nexus"; +import { streamService } from "../services"; +import type { AppendRecordsResponse, AppendRecordsRequest } from "../models"; + +/** + * Append one batch. Workflow code makes the call from a Local Activity. + * + * @param request - Request for the operation. + */ +export async function appendRecords( + request: AppendRecordsRequest, +): Promise> { + const client = nexus.createNexusServiceClient({ + service: streamService, + endpoint: "__temporal_system", + }); + return await client.startOperation(streamService.operations.appendRecords, request); +} diff --git a/advanced/samples/typescript/wit/stream-service/operations/close-stream.ts b/advanced/samples/typescript/wit/stream-service/operations/close-stream.ts new file mode 100644 index 000000000..3fc8c5656 --- /dev/null +++ b/advanced/samples/typescript/wit/stream-service/operations/close-stream.ts @@ -0,0 +1,18 @@ +// Generated by nexgen latest. DO NOT EDIT! + +import * as nexus from "../../../../nexus"; +import { streamService } from "../services"; +import type { CloseStreamResponse, CloseStreamRequest } from "../models"; + +/** + * @param request - Request for the operation. + */ +export async function closeStream( + request: CloseStreamRequest, +): Promise> { + const client = nexus.createNexusServiceClient({ + service: streamService, + endpoint: "__temporal_system", + }); + return await client.startOperation(streamService.operations.closeStream, request); +} diff --git a/advanced/samples/typescript/wit/stream-service/operations/create-stream.ts b/advanced/samples/typescript/wit/stream-service/operations/create-stream.ts new file mode 100644 index 000000000..cf3a94619 --- /dev/null +++ b/advanced/samples/typescript/wit/stream-service/operations/create-stream.ts @@ -0,0 +1,36 @@ +// Generated by nexgen latest. DO NOT EDIT! + +import * as nexus from "../../../../nexus"; +import { streamService } from "../services"; +import type { CreateStreamRequest } from "../models"; +import { Stream } from "../resources"; + +function requiredField( + value: T | null | undefined, + owner: string, + field: string, +): T { + if (value == null) { + throw new Error(`missing required field ${owner}.${field}`); + } + return value; +} + +/** + * @param request - Request for the operation. + */ +export async function createStream(request: CreateStreamRequest): Promise { + const client = nexus.createNexusServiceClient({ + service: streamService, + endpoint: "__temporal_system", + }); + const handle = await client.startOperation( + streamService.operations.createStream, + request, + ); + const result = await handle.result(); + return new Stream( + requiredField(result.stream, "resource", "stream"), + result.created ?? undefined, + ); +} diff --git a/advanced/samples/typescript/wit/stream-service/operations/describe-stream.ts b/advanced/samples/typescript/wit/stream-service/operations/describe-stream.ts new file mode 100644 index 000000000..3348a6cd1 --- /dev/null +++ b/advanced/samples/typescript/wit/stream-service/operations/describe-stream.ts @@ -0,0 +1,18 @@ +// Generated by nexgen latest. DO NOT EDIT! + +import * as nexus from "../../../../nexus"; +import { streamService } from "../services"; +import type { DescribeStreamResponse, DescribeStreamRequest } from "../models"; + +/** + * @param request - Request for the operation. + */ +export async function describeStream( + request: DescribeStreamRequest, +): Promise> { + const client = nexus.createNexusServiceClient({ + service: streamService, + endpoint: "__temporal_system", + }); + return await client.startOperation(streamService.operations.describeStream, request); +} diff --git a/advanced/samples/typescript/wit/stream-service/operations/read-records.ts b/advanced/samples/typescript/wit/stream-service/operations/read-records.ts new file mode 100644 index 000000000..bdb16735f --- /dev/null +++ b/advanced/samples/typescript/wit/stream-service/operations/read-records.ts @@ -0,0 +1,23 @@ +// Generated by nexgen latest. DO NOT EDIT! + +import * as nexus from "../../../../nexus"; +import { streamService } from "../services"; +import type { ReadRecordsResponse, ReadRecordsRequest } from "../models"; +import "../resources"; +import type { Stream } from "../resources"; + +/** + * A long poll. NOT_FOUND for an uncreated Stream. Workflow code reads from a Local + * Activity. + * + * @param request - Request for the operation. + */ +export async function readRecords( + request: ReadRecordsRequest, +): Promise> { + const client = nexus.createNexusServiceClient({ + service: streamService, + endpoint: "__temporal_system", + }); + return await client.startOperation(streamService.operations.readRecords, request); +} diff --git a/advanced/samples/typescript/wit/stream-service/registry.ts b/advanced/samples/typescript/wit/stream-service/registry.ts new file mode 100644 index 000000000..c0d7421f0 --- /dev/null +++ b/advanced/samples/typescript/wit/stream-service/registry.ts @@ -0,0 +1,24 @@ +// Generated by nexgen latest. DO NOT EDIT! + +export const operationRegistry = [ + { + service: "temporal.api.streamservice.v1.StreamService", + operation: "CreateStream", + }, + { + service: "temporal.api.streamservice.v1.StreamService", + operation: "AppendRecords", + }, + { + service: "temporal.api.streamservice.v1.StreamService", + operation: "CloseStream", + }, + { + service: "temporal.api.streamservice.v1.StreamService", + operation: "ReadRecords", + }, + { + service: "temporal.api.streamservice.v1.StreamService", + operation: "DescribeStream", + }, +] as const; diff --git a/advanced/samples/typescript/wit/stream-service/resources.ts b/advanced/samples/typescript/wit/stream-service/resources.ts new file mode 100644 index 000000000..5dcb834e0 --- /dev/null +++ b/advanced/samples/typescript/wit/stream-service/resources.ts @@ -0,0 +1,61 @@ +// Generated by nexgen latest. DO NOT EDIT! + +import type * as common from "@temporalio/common"; +import type { + StreamReadStart, + StreamRef, + AppendRecordsResponse, + AppendRecord, + ReadRecordsResponse, + DescribeStreamResponse, +} from "./models"; +import { appendRecords } from "./operations/append-records"; +import { closeStream } from "./operations/close-stream"; +import { readRecords } from "./operations/read-records"; +import { describeStream } from "./operations/describe-stream"; + +export class Stream { + public constructor( + public readonly stream: StreamRef, + public readonly created: boolean | undefined, + ) {} + + public async publish(records: AppendRecord[]): Promise { + const request = { stream: this.stream, records: records }; + const handle = await appendRecords(request); + return await handle.result(); + } + + public async close(result?: common.Payload, failure?: Error): Promise { + const request = { stream: this.stream, result: result, failure: failure }; + const handle = await closeStream(request); + await handle.result(); + } + + public async read( + fromOffset?: number, + start?: StreamReadStart, + maxRecords?: number, + wait?: common.Duration, + createWait?: common.Duration, + maxBytes?: number, + ): Promise { + const request = { + stream: this.stream, + fromOffset: fromOffset, + start: start, + maxRecords: maxRecords, + wait: wait, + createWait: createWait, + maxBytes: maxBytes, + }; + const handle = await readRecords(request); + return await handle.result(); + } + + public async describe(): Promise { + const request = { stream: this.stream }; + const handle = await describeStream(request); + return await handle.result(); + } +} diff --git a/advanced/samples/typescript/wit/stream-service/services.ts b/advanced/samples/typescript/wit/stream-service/services.ts new file mode 100644 index 000000000..f90f7dd73 --- /dev/null +++ b/advanced/samples/typescript/wit/stream-service/services.ts @@ -0,0 +1,71 @@ +// Generated by nexgen latest. DO NOT EDIT! + +import * as nexus from "nexus-rpc"; +import { + createStreamResultTransferTypeConverter, + createStreamRequestTransferTypeConverter, + appendRecordsResponseTransferTypeConverter, + appendRecordsRequestTransferTypeConverter, + closeStreamResponseTransferTypeConverter, + closeStreamRequestTransferTypeConverter, + readRecordsResponseTransferTypeConverter, + readRecordsRequestTransferTypeConverter, + describeStreamResponseTransferTypeConverter, + describeStreamRequestTransferTypeConverter, +} from "./models"; +import type { + CreateStreamResult, + CreateStreamRequest, + AppendRecordsResponse, + AppendRecordsRequest, + CloseStreamResponse, + CloseStreamRequest, + ReadRecordsResponse, + ReadRecordsRequest, + DescribeStreamResponse, + DescribeStreamRequest, +} from "./models"; +import type { Stream } from "./resources"; + +/** + * @experimental This API is experimental and subject to change. + */ +export const streamService = nexus.service( + "temporal.api.streamservice.v1.StreamService", + { + createStream: nexus.operation({ + name: "CreateStream", + inputType: { transferTypeConverter: createStreamRequestTransferTypeConverter }, + outputType: { transferTypeConverter: createStreamResultTransferTypeConverter }, + }), + /** + * Append one batch. Workflow code makes the call from a Local Activity. + */ + appendRecords: nexus.operation({ + name: "AppendRecords", + inputType: { transferTypeConverter: appendRecordsRequestTransferTypeConverter }, + outputType: { transferTypeConverter: appendRecordsResponseTransferTypeConverter }, + }), + closeStream: nexus.operation({ + name: "CloseStream", + inputType: { transferTypeConverter: closeStreamRequestTransferTypeConverter }, + outputType: { transferTypeConverter: closeStreamResponseTransferTypeConverter }, + }), + /** + * A long poll. NOT_FOUND for an uncreated Stream. Workflow code reads from a Local + * Activity. + */ + readRecords: nexus.operation({ + name: "ReadRecords", + inputType: { transferTypeConverter: readRecordsRequestTransferTypeConverter }, + outputType: { transferTypeConverter: readRecordsResponseTransferTypeConverter }, + }), + describeStream: nexus.operation({ + name: "DescribeStream", + inputType: { transferTypeConverter: describeStreamRequestTransferTypeConverter }, + outputType: { + transferTypeConverter: describeStreamResponseTransferTypeConverter, + }, + }), + }, +); diff --git a/changelog/fixed/lantern-otter.md b/changelog/fixed/lantern-otter.md new file mode 100644 index 000000000..2cf80ac67 --- /dev/null +++ b/changelog/fixed/lantern-otter.md @@ -0,0 +1 @@ +A protobuf `oneof` case that names a proto-bound WIT record plans as that record, so Python and TypeScript no longer panic on it, and .NET types the case as the record. diff --git a/src/generator/dotnet.rs b/src/generator/dotnet.rs index 0ccb5be2a..6ee85e190 100644 --- a/src/generator/dotnet.rs +++ b/src/generator/dotnet.rs @@ -1,4 +1,4 @@ -use std::collections::BTreeMap; +use std::collections::{BTreeMap, BTreeSet}; use std::rc::Rc; use heck::{ToLowerCamelCase, ToUpperCamelCase}; @@ -143,7 +143,7 @@ impl<'a> ApiPlanner<'a> { render_flags(&mut output, flag_set); } for variant in self.api_plan.variants().map(|(_, variant)| variant) { - self.render_variant(&mut output, variant); + self.render_variant(&mut output, variant, namespace); } for model in models { self.render_model(&mut output, model); @@ -628,7 +628,7 @@ impl<'a> ApiPlanner<'a> { } } - fn render_variant(&self, output: &mut String, variant: &PlannedVariant) { + fn render_variant(&self, output: &mut String, variant: &PlannedVariant, namespace: &str) { let type_parameters = self.api_plan.variant_type_parameters(&variant.full_name); let type_arguments = type_parameters .iter() @@ -649,6 +649,11 @@ impl<'a> ApiPlanner<'a> { output.push_str(" private "); output.push_str(&csharp_type_name(&variant.name)); output.push_str("() { }\n\n"); + let case_names = variant + .cases + .iter() + .map(|case| csharp_type_name(&case.name)) + .collect::>(); for (index, case) in variant.cases.iter().enumerate() { let case_name = csharp_type_name(&case.name); render_xml_summary(output, " ", case.doc.value(), variant.experimental); @@ -658,7 +663,12 @@ impl<'a> ApiPlanner<'a> { output.push_str(" public sealed record "); output.push_str(&case_name); if let Some(payload) = &case.payload { - let payload_type = self.dotnet_value_type(payload); + // Inside the variant, a case record hides the model of the same name. + let payload_type = qualify_csharp_type_names( + &self.dotnet_value_type(payload), + &case_names, + namespace, + ); output.push('('); output.push_str(&payload_type); output.push_str(" Value) : "); @@ -2706,6 +2716,30 @@ fn dotnet_source_expr(source_expr: &str, support_namespace: Option<&str>) -> Str ) } +/// Writes each bare type name of `type_expr` that is in `names` as `global::.`. +fn qualify_csharp_type_names(type_expr: &str, names: &BTreeSet, namespace: &str) -> String { + let mut output = String::with_capacity(type_expr.len()); + let mut rest = type_expr; + while let Some(start) = rest.find(|c: char| c.is_ascii_alphabetic() || c == '_' || c == '@') { + let (before, from_start) = rest.split_at(start); + output.push_str(before); + let end = from_start + .find(|c: char| !(c.is_ascii_alphanumeric() || c == '_' || c == '@')) + .unwrap_or(from_start.len()); + let (ident, after) = from_start.split_at(end); + let qualified = output.ends_with('.') || output.ends_with("::"); + if !qualified && names.contains(ident) { + output.push_str("global::"); + output.push_str(namespace); + output.push('.'); + } + output.push_str(ident); + rest = after; + } + output.push_str(rest); + output +} + fn generated_file_prelude(namespace: &str, imports: &[&str]) -> String { let mut output = String::new(); output.push_str(GENERATED_HEADER); @@ -4453,7 +4487,27 @@ const CSHARP_KEYWORDS: &[&str] = &[ #[cfg(test)] mod tests { - use super::{csharp_type_parameter_name, dotnet_source_expr}; + use std::collections::BTreeSet; + + use super::{csharp_type_parameter_name, dotnet_source_expr, qualify_csharp_type_names}; + + #[test] + fn a_case_payload_names_a_model_of_a_case_name_in_full() { + let cases = BTreeSet::from(["Superseded".to_string()]); + let qualify = + |type_expr| qualify_csharp_type_names(type_expr, &cases, "Temporalio.Workflows"); + + assert_eq!( + qualify("Superseded"), + "global::Temporalio.Workflows.Superseded" + ); + assert_eq!( + qualify("IReadOnlyList?"), + "IReadOnlyList?" + ); + assert_eq!(qualify("Proto.Superseded"), "Proto.Superseded"); + assert_eq!(qualify("StreamClose"), "StreamClose"); + } #[test] fn sourced_helper_is_qualified_without_changing_its_arguments() { diff --git a/src/planning/type_planning.rs b/src/planning/type_planning.rs index 43a2593e4..c82fbad17 100644 --- a/src/planning/type_planning.rs +++ b/src/planning/type_planning.rs @@ -520,9 +520,49 @@ impl<'a> TypePlanningContext<'a> { planner: self, }); self.resolve_planned_record_field_types(&source_spec, &mut planned_spec); + self.resolve_planned_variant_case_types(&source_spec, &mut planned_spec); planned_spec } + // `map_names` keeps a oneof case payload as its proto message. Re-plan it as the bound record. + fn resolve_planned_variant_case_types( + &mut self, + source_spec: &ApiSpec, + planned_spec: &mut PlannedSpec, + ) { + for (variant_name, planned_entry) in &mut planned_spec.types { + let TypeDeclSpec::Variant(planned_variant) = &mut planned_entry.declaration else { + continue; + }; + let Some(source_variant) = source_spec.variant(variant_name).filter(|variant| { + variant + .source + .as_ref() + .is_some_and(|source| source.proto_oneof().is_some()) + }) else { + continue; + }; + for (case, source_case) in planned_variant.cases.iter_mut().zip(&source_variant.cases) { + let Some(payload) = source_case.payload.as_ref() else { + continue; + }; + let TypeSpec::External(ExternalTypeSpec::Proto(proto_name)) = + payload.without_option() + else { + continue; + }; + if self.spec.record_for_proto(proto_name.as_str()).is_none() { + continue; + } + let record = self.planned_type_from_authored(payload); + case.payload = Some(match payload { + TypeSpec::Option(_) => TypeSpec::Option(Box::new(record)), + _ => record, + }); + } + } + } + fn resolve_planned_record_field_types( &mut self, source_spec: &ApiSpec, From 54da5a5c7621cc42fdfcb8f8a21e257d88dc8b43 Mon Sep 17 00:00:00 2001 From: Moe Dashti Date: Wed, 28 Oct 2026 16:57:07 -0700 Subject: [PATCH 2/2] Generated the stream service for Python, with today's output. nexgen maps every temporal/... proto to a temporalio module. The Stream protos aren't in temporalio yet, so the service's imports don't resolve, and basedpyright skips it. --- advanced/samples/README.md | 4 +- advanced/samples/inputs/examples.toml | 2 +- advanced/samples/python/pyproject.toml | 3 +- .../python/wit/stream_service/__init__.py | 81 + .../wit/stream_service/_resources/__init__.py | 9 + .../wit/stream_service/_resources/stream.py | 230 +++ .../_system_nexus_interceptor.py | 266 ++++ .../python/wit/stream_service/models.py | 1395 +++++++++++++++++ .../wit/stream_service/operations/__init__.py | 3 + .../operations/create_stream.py | 62 + .../python/wit/stream_service/services.py | 50 + 11 files changed, 2102 insertions(+), 3 deletions(-) create mode 100644 advanced/samples/python/wit/stream_service/__init__.py create mode 100644 advanced/samples/python/wit/stream_service/_resources/__init__.py create mode 100644 advanced/samples/python/wit/stream_service/_resources/stream.py create mode 100644 advanced/samples/python/wit/stream_service/_system_nexus_interceptor.py create mode 100644 advanced/samples/python/wit/stream_service/models.py create mode 100644 advanced/samples/python/wit/stream_service/operations/__init__.py create mode 100644 advanced/samples/python/wit/stream_service/operations/create_stream.py create mode 100644 advanced/samples/python/wit/stream_service/services.py diff --git a/advanced/samples/README.md b/advanced/samples/README.md index 7ff2e9f08..8785e064b 100644 --- a/advanced/samples/README.md +++ b/advanced/samples/README.md @@ -163,13 +163,15 @@ crate omits from the API of a resource method's request is left out of the model. nexgen ignores the directives that it doesn't know yet, such as `@nexus.varargs` and `@nexus.close`. -Python doesn't generate the service, since it can't import the protos. +Some of the output doesn't type-check. Python imports the Stream protos from +`temporalio`, which doesn't have them, so basedpyright skips the service. TypeScript passes a `number` where a resource method's proto request wants a `Long`. So `tsc` skips the service, but the test still checks its models. - WIT: [`inputs/stream-service.wit`](inputs/stream-service.wit) - .NET: [`dotnet/wit/stream-service/`](dotnet/wit/stream-service/), [`dotnet/tests/StreamServiceChecks.cs`](dotnet/tests/StreamServiceChecks.cs) +- Python: [`python/wit/stream_service/`](python/wit/stream_service/) - TypeScript: [`typescript/wit/stream-service/`](typescript/wit/stream-service/), [`typescript/tests/stream-service.test.ts`](typescript/tests/stream-service.test.ts) diff --git a/advanced/samples/inputs/examples.toml b/advanced/samples/inputs/examples.toml index e80804baf..db39644df 100644 --- a/advanced/samples/inputs/examples.toml +++ b/advanced/samples/inputs/examples.toml @@ -29,7 +29,7 @@ descriptors = "temporal" # The Stream service WIT of temporalio-streams, on the Stream protos of temporalio/api#895 at # 44519d8a. Its .NET output builds in Nexgen.DotNetStreamService.csproj. [stream-service] -languages = ["typescript", "dotnet"] +languages = ["python", "typescript", "dotnet"] system_nexus = true descriptors = ["temporal_api.bin", "streams.bin"] diff --git a/advanced/samples/python/pyproject.toml b/advanced/samples/python/pyproject.toml index 2089cc161..d3bd88365 100644 --- a/advanced/samples/python/pyproject.toml +++ b/advanced/samples/python/pyproject.toml @@ -26,7 +26,8 @@ target-version = "py310" extend-exclude = ["streamprotos"] [tool.basedpyright] -exclude = [".venv", "streamprotos"] +# The stream service imports Stream protos that temporalio doesn't have yet. +exclude = [".venv", "streamprotos", "wit/stream_service"] reportAny = false reportExplicitAny = false reportPrivateUsage = false diff --git a/advanced/samples/python/wit/stream_service/__init__.py b/advanced/samples/python/wit/stream_service/__init__.py new file mode 100644 index 000000000..a1ed1a074 --- /dev/null +++ b/advanced/samples/python/wit/stream_service/__init__.py @@ -0,0 +1,81 @@ +# Generated by nexgen latest. DO NOT EDIT! + +from __future__ import annotations + +import collections.abc +import typing + +import nexusrpc +import temporalio.converter +from . import services as _services +from .operations.create_stream import create_stream +from ._resources.stream import append_records +from ._resources.stream import close_stream +from ._resources.stream import read_records +from ._resources.stream import describe_stream + +__all__ = [ + "append_records", + "close_stream", + "create_stream", + "describe_stream", + "read_records", +] + + +_InputT = typing.TypeVar("_InputT") +_OutputT = typing.TypeVar("_OutputT") + + +_SerializationContextFactory = collections.abc.Callable[ + [_InputT], temporalio.converter.SerializationContext +] + + +class _NexusOperationInfo(typing.Generic[_InputT, _OutputT]): + def __init__( + self, + *, + operation: nexusrpc.Operation[_InputT, _OutputT], + serialization_context: _SerializationContextFactory[_InputT] | None = None, + ) -> None: + self.operation: nexusrpc.Operation[_InputT, _OutputT] = operation + self.serialization_context: _SerializationContextFactory[_InputT] | None = ( + serialization_context + ) + + +__nexus_operation_registry__: dict[ + tuple[str, str], _NexusOperationInfo[typing.Any, typing.Any] +] = { + ( + "temporal.api.streamservice.v1.StreamService", + "CreateStream", + ): _NexusOperationInfo( + operation=_services.StreamService.create_stream, + ), + ( + "temporal.api.streamservice.v1.StreamService", + "AppendRecords", + ): _NexusOperationInfo( + operation=_services.StreamService.append_records, + ), + ( + "temporal.api.streamservice.v1.StreamService", + "CloseStream", + ): _NexusOperationInfo( + operation=_services.StreamService.close_stream, + ), + ( + "temporal.api.streamservice.v1.StreamService", + "ReadRecords", + ): _NexusOperationInfo( + operation=_services.StreamService.read_records, + ), + ( + "temporal.api.streamservice.v1.StreamService", + "DescribeStream", + ): _NexusOperationInfo( + operation=_services.StreamService.describe_stream, + ), +} diff --git a/advanced/samples/python/wit/stream_service/_resources/__init__.py b/advanced/samples/python/wit/stream_service/_resources/__init__.py new file mode 100644 index 000000000..06bb70ddd --- /dev/null +++ b/advanced/samples/python/wit/stream_service/_resources/__init__.py @@ -0,0 +1,9 @@ +# Generated by nexgen latest. DO NOT EDIT! + +from __future__ import annotations + +from .stream import Stream + +__all__ = [ + "Stream", +] diff --git a/advanced/samples/python/wit/stream_service/_resources/stream.py b/advanced/samples/python/wit/stream_service/_resources/stream.py new file mode 100644 index 000000000..ecd798701 --- /dev/null +++ b/advanced/samples/python/wit/stream_service/_resources/stream.py @@ -0,0 +1,230 @@ +# Generated by nexgen latest. DO NOT EDIT! + +from __future__ import annotations + +import dataclasses +import typing +import datetime +import temporalio.workflow + +from ..models import ( + AppendRecord, + AppendRecordsRequest, + AppendRecordsResponse, + CloseStreamRequest, + CloseStreamResponse, + DescribeStreamRequest, + DescribeStreamResponse, + Producer, + ReadRecordsRequest, + ReadRecordsResponse, + StreamReadStart, + StreamRef, +) + + +@dataclasses.dataclass +class Stream: + """A handle on one Stream. Build it from a stream-ref, or get it from a create.""" + + stream: StreamRef + created: bool | None + + async def publish( + self, + records: list[AppendRecord] | None = None, + ) -> AppendRecordsResponse: + """Publish values, and return the offset of the last one.""" + request = AppendRecordsRequest( + stream=self.stream, + records=records or [], + ) + handle = await _append_records(request) + return await handle + + async def close( + self, + result: typing.Any | None = None, + failure: BaseException | None = None, + ) -> None: + """Close the Stream. Readers drain what is left, then end.""" + request = CloseStreamRequest( + stream=self.stream, + result=result, + failure=failure, + ) + handle = await _close_stream(request) + await handle + + async def read( + self, + from_offset: int | None = None, + start: StreamReadStart | None = None, + max_records: int | None = None, + wait: datetime.timedelta | None = None, + create_wait: datetime.timedelta | None = None, + max_bytes: int | None = None, + ) -> ReadRecordsResponse: + """Read one page of Records. The poll iterator reads them all.""" + request = ReadRecordsRequest( + stream=self.stream, + from_offset=from_offset, + start=start, + max_records=max_records, + wait=wait, + create_wait=create_wait, + max_bytes=max_bytes, + ) + handle = await _read_records(request) + return await handle + + async def describe( + self, + ) -> DescribeStreamResponse: + """Describe the Stream's status, newest offset and Retention.""" + request = DescribeStreamRequest( + stream=self.stream, + ) + handle = await _describe_stream(request) + return await handle + + +async def _append_records( + request: AppendRecordsRequest, +) -> temporalio.workflow.NexusOperationHandle[AppendRecordsResponse,]: + nexus_client = temporalio.workflow.create_nexus_client( + service="temporal.api.streamservice.v1.StreamService", + endpoint="__temporal_system", + ) + return await nexus_client.start_operation( + operation="AppendRecords", + input=request, + output_type=AppendRecordsResponse, + ) + + +async def append_records( + *, + stream: StreamRef, + records: list[AppendRecord] | None = None, + producer: Producer | None = None, +) -> temporalio.workflow.NexusOperationHandle[AppendRecordsResponse,]: + """Append one batch. Workflow code makes the call from a Local Activity. + + Args: + producer: Required outside a Workflow. In a Workflow the transport sets it. + """ + request = AppendRecordsRequest( + stream=stream, + records=records or [], + producer=producer, + ) + return await _append_records(request) + + +async def _close_stream( + request: CloseStreamRequest, +) -> temporalio.workflow.NexusOperationHandle[CloseStreamResponse,]: + nexus_client = temporalio.workflow.create_nexus_client( + service="temporal.api.streamservice.v1.StreamService", + endpoint="__temporal_system", + ) + return await nexus_client.start_operation( + operation="CloseStream", + input=request, + output_type=CloseStreamResponse, + ) + + +async def close_stream( + *, + stream: StreamRef, + producer: Producer | None = None, + result: typing.Any | None = None, + failure: BaseException | None = None, +) -> temporalio.workflow.NexusOperationHandle[CloseStreamResponse,]: + request = CloseStreamRequest( + stream=stream, + producer=producer, + result=result, + failure=failure, + ) + return await _close_stream(request) + + +async def _read_records( + request: ReadRecordsRequest, +) -> temporalio.workflow.NexusOperationHandle[ReadRecordsResponse,]: + nexus_client = temporalio.workflow.create_nexus_client( + service="temporal.api.streamservice.v1.StreamService", + endpoint="__temporal_system", + ) + return await nexus_client.start_operation( + operation="ReadRecords", + input=request, + output_type=ReadRecordsResponse, + ) + + +async def read_records( + *, + stream: StreamRef, + from_offset: int | None = None, + start: StreamReadStart | None = None, + max_records: int | None = None, + wait: datetime.timedelta | None = None, + create_wait: datetime.timedelta | None = None, + last: int | None = None, + max_bytes: int | None = None, +) -> temporalio.workflow.NexusOperationHandle[ReadRecordsResponse,]: + """A long poll. NOT_FOUND for an uncreated Stream. Workflow code reads from a Local + Activity. + + Args: + from_offset: Start offset. Empty uses start. OUT_OF_RANGE past the newest offset + + 1. + start: Where to start when from-offset is empty. Unspecified is the beginning. + max_records: About this many items per page, 1 to 1000. A Superseded item stays + with its Record. + wait: How long to wait for a new Record before an empty answer. At most 60 s. + create_wait: How long to wait for an uncreated Stream. Empty fails at once. At + most 60 s. + last: Read the newest this many Records, oldest first. No from-offset. No wait. + max_bytes: The maximum Record bytes in a page. A larger Record is still returned + alone. + """ + request = ReadRecordsRequest( + stream=stream, + from_offset=from_offset, + start=start, + max_records=max_records, + wait=wait, + create_wait=create_wait, + last=last, + max_bytes=max_bytes, + ) + return await _read_records(request) + + +async def _describe_stream( + request: DescribeStreamRequest, +) -> temporalio.workflow.NexusOperationHandle[DescribeStreamResponse,]: + nexus_client = temporalio.workflow.create_nexus_client( + service="temporal.api.streamservice.v1.StreamService", + endpoint="__temporal_system", + ) + return await nexus_client.start_operation( + operation="DescribeStream", + input=request, + output_type=DescribeStreamResponse, + ) + + +async def describe_stream( + *, + stream: StreamRef, +) -> temporalio.workflow.NexusOperationHandle[DescribeStreamResponse,]: + request = DescribeStreamRequest( + stream=stream, + ) + return await _describe_stream(request) diff --git a/advanced/samples/python/wit/stream_service/_system_nexus_interceptor.py b/advanced/samples/python/wit/stream_service/_system_nexus_interceptor.py new file mode 100644 index 000000000..bec9fbe39 --- /dev/null +++ b/advanced/samples/python/wit/stream_service/_system_nexus_interceptor.py @@ -0,0 +1,266 @@ +# Generated by nexgen latest. DO NOT EDIT! + +from __future__ import annotations + + +import abc +import typing + +from . import models + +if typing.TYPE_CHECKING: + import temporalio.workflow + from temporalio.worker._interceptor import StartNexusOperationInput + + +__all__ = [ + "_start_system_nexus_operation", + "_SystemNexusWorkflowOutboundInterceptorBase", + "_SystemNexusWorkflowOutboundInterceptorTerminal", +] + + +_InputT = typing.TypeVar("_InputT") +_OutputT = typing.TypeVar("_OutputT") + + +async def _start_system_nexus_operation( + interceptor: _SystemNexusWorkflowOutboundInterceptorBase, + input: StartNexusOperationInput[_InputT, _OutputT], +) -> temporalio.workflow.NexusOperationHandle[_OutputT]: + if ( + input.service == "temporal.api.streamservice.v1.StreamService" + and input.operation_name == "CreateStream" + ): + typed_input = typing.cast( + "StartNexusOperationInput[models.CreateStreamRequest, models.CreateStreamResult]", + input, + ) + # The dispatch check above establishes that this operation's response type is _OutputT. + return typing.cast( + "temporalio.workflow.NexusOperationHandle[_OutputT]", + await interceptor.start_create_stream(typed_input.input), + ) + if ( + input.service == "temporal.api.streamservice.v1.StreamService" + and input.operation_name == "AppendRecords" + ): + typed_input = typing.cast( + "StartNexusOperationInput[models.AppendRecordsRequest, models.AppendRecordsResponse]", + input, + ) + # The dispatch check above establishes that this operation's response type is _OutputT. + return typing.cast( + "temporalio.workflow.NexusOperationHandle[_OutputT]", + await interceptor.start_append_records(typed_input.input), + ) + if ( + input.service == "temporal.api.streamservice.v1.StreamService" + and input.operation_name == "CloseStream" + ): + typed_input = typing.cast( + "StartNexusOperationInput[models.CloseStreamRequest, models.CloseStreamResponse]", + input, + ) + # The dispatch check above establishes that this operation's response type is _OutputT. + return typing.cast( + "temporalio.workflow.NexusOperationHandle[_OutputT]", + await interceptor.start_close_stream(typed_input.input), + ) + if ( + input.service == "temporal.api.streamservice.v1.StreamService" + and input.operation_name == "ReadRecords" + ): + typed_input = typing.cast( + "StartNexusOperationInput[models.ReadRecordsRequest, models.ReadRecordsResponse]", + input, + ) + # The dispatch check above establishes that this operation's response type is _OutputT. + return typing.cast( + "temporalio.workflow.NexusOperationHandle[_OutputT]", + await interceptor.start_read_records(typed_input.input), + ) + if ( + input.service == "temporal.api.streamservice.v1.StreamService" + and input.operation_name == "DescribeStream" + ): + typed_input = typing.cast( + "StartNexusOperationInput[models.DescribeStreamRequest, models.DescribeStreamResponse]", + input, + ) + # The dispatch check above establishes that this operation's response type is _OutputT. + return typing.cast( + "temporalio.workflow.NexusOperationHandle[_OutputT]", + await interceptor.start_describe_stream(typed_input.input), + ) + return await interceptor.start_system_nexus_operation(input) + + +class _SystemNexusWorkflowOutboundInterceptorBase(abc.ABC): + @abc.abstractmethod + def _next_system_nexus_interceptor( + self, + ) -> _SystemNexusWorkflowOutboundInterceptorBase: ... + + @abc.abstractmethod + async def start_system_nexus_operation( + self, + input: StartNexusOperationInput[_InputT, _OutputT], + ) -> temporalio.workflow.NexusOperationHandle[_OutputT]: + """Intercept a System Nexus operation through the generic hook.""" + ... + + async def start_create_stream( + self, request: models.CreateStreamRequest + ) -> temporalio.workflow.NexusOperationHandle[models.CreateStreamResult]: + """Intercept the CreateStream operation.""" + return await self._next_system_nexus_interceptor().start_create_stream(request) + + async def start_append_records( + self, request: models.AppendRecordsRequest + ) -> temporalio.workflow.NexusOperationHandle[models.AppendRecordsResponse]: + """Intercept the AppendRecords operation.""" + return await self._next_system_nexus_interceptor().start_append_records(request) + + async def start_close_stream( + self, request: models.CloseStreamRequest + ) -> temporalio.workflow.NexusOperationHandle[models.CloseStreamResponse]: + """Intercept the CloseStream operation.""" + return await self._next_system_nexus_interceptor().start_close_stream(request) + + async def start_read_records( + self, request: models.ReadRecordsRequest + ) -> temporalio.workflow.NexusOperationHandle[models.ReadRecordsResponse]: + """Intercept the ReadRecords operation.""" + return await self._next_system_nexus_interceptor().start_read_records(request) + + async def start_describe_stream( + self, request: models.DescribeStreamRequest + ) -> temporalio.workflow.NexusOperationHandle[models.DescribeStreamResponse]: + """Intercept the DescribeStream operation.""" + return await self._next_system_nexus_interceptor().start_describe_stream( + request + ) + + +class _SystemNexusWorkflowOutboundInterceptorTerminal(abc.ABC): + @abc.abstractmethod + async def _intercept_system_nexus_operation( + self, + input: StartNexusOperationInput[_InputT, _OutputT], + ) -> temporalio.workflow.NexusOperationHandle[_OutputT]: ... + + async def start_create_stream( + self, request: models.CreateStreamRequest + ) -> temporalio.workflow.NexusOperationHandle[models.CreateStreamResult]: + from temporalio.worker._interceptor import StartNexusOperationInput + from temporalio.nexus.system import TEMPORAL_SYSTEM_ENDPOINT + from temporalio.workflow import NexusOperationCancellationType + + return await self._intercept_system_nexus_operation( + StartNexusOperationInput( + endpoint=TEMPORAL_SYSTEM_ENDPOINT, + service="temporal.api.streamservice.v1.StreamService", + operation="CreateStream", + input=request, + output_type=models.CreateStreamResult, + schedule_to_close_timeout=None, + schedule_to_start_timeout=None, + start_to_close_timeout=None, + cancellation_type=NexusOperationCancellationType.WAIT_COMPLETED, + headers=None, + summary=None, + ) + ) + + async def start_append_records( + self, request: models.AppendRecordsRequest + ) -> temporalio.workflow.NexusOperationHandle[models.AppendRecordsResponse]: + from temporalio.worker._interceptor import StartNexusOperationInput + from temporalio.nexus.system import TEMPORAL_SYSTEM_ENDPOINT + from temporalio.workflow import NexusOperationCancellationType + + return await self._intercept_system_nexus_operation( + StartNexusOperationInput( + endpoint=TEMPORAL_SYSTEM_ENDPOINT, + service="temporal.api.streamservice.v1.StreamService", + operation="AppendRecords", + input=request, + output_type=models.AppendRecordsResponse, + schedule_to_close_timeout=None, + schedule_to_start_timeout=None, + start_to_close_timeout=None, + cancellation_type=NexusOperationCancellationType.WAIT_COMPLETED, + headers=None, + summary=None, + ) + ) + + async def start_close_stream( + self, request: models.CloseStreamRequest + ) -> temporalio.workflow.NexusOperationHandle[models.CloseStreamResponse]: + from temporalio.worker._interceptor import StartNexusOperationInput + from temporalio.nexus.system import TEMPORAL_SYSTEM_ENDPOINT + from temporalio.workflow import NexusOperationCancellationType + + return await self._intercept_system_nexus_operation( + StartNexusOperationInput( + endpoint=TEMPORAL_SYSTEM_ENDPOINT, + service="temporal.api.streamservice.v1.StreamService", + operation="CloseStream", + input=request, + output_type=models.CloseStreamResponse, + schedule_to_close_timeout=None, + schedule_to_start_timeout=None, + start_to_close_timeout=None, + cancellation_type=NexusOperationCancellationType.WAIT_COMPLETED, + headers=None, + summary=None, + ) + ) + + async def start_read_records( + self, request: models.ReadRecordsRequest + ) -> temporalio.workflow.NexusOperationHandle[models.ReadRecordsResponse]: + from temporalio.worker._interceptor import StartNexusOperationInput + from temporalio.nexus.system import TEMPORAL_SYSTEM_ENDPOINT + from temporalio.workflow import NexusOperationCancellationType + + return await self._intercept_system_nexus_operation( + StartNexusOperationInput( + endpoint=TEMPORAL_SYSTEM_ENDPOINT, + service="temporal.api.streamservice.v1.StreamService", + operation="ReadRecords", + input=request, + output_type=models.ReadRecordsResponse, + schedule_to_close_timeout=None, + schedule_to_start_timeout=None, + start_to_close_timeout=None, + cancellation_type=NexusOperationCancellationType.WAIT_COMPLETED, + headers=None, + summary=None, + ) + ) + + async def start_describe_stream( + self, request: models.DescribeStreamRequest + ) -> temporalio.workflow.NexusOperationHandle[models.DescribeStreamResponse]: + from temporalio.worker._interceptor import StartNexusOperationInput + from temporalio.nexus.system import TEMPORAL_SYSTEM_ENDPOINT + from temporalio.workflow import NexusOperationCancellationType + + return await self._intercept_system_nexus_operation( + StartNexusOperationInput( + endpoint=TEMPORAL_SYSTEM_ENDPOINT, + service="temporal.api.streamservice.v1.StreamService", + operation="DescribeStream", + input=request, + output_type=models.DescribeStreamResponse, + schedule_to_close_timeout=None, + schedule_to_start_timeout=None, + start_to_close_timeout=None, + cancellation_type=NexusOperationCancellationType.WAIT_COMPLETED, + headers=None, + summary=None, + ) + ) diff --git a/advanced/samples/python/wit/stream_service/models.py b/advanced/samples/python/wit/stream_service/models.py new file mode 100644 index 000000000..4afe7aa2d --- /dev/null +++ b/advanced/samples/python/wit/stream_service/models.py @@ -0,0 +1,1395 @@ +# Generated by nexgen latest. DO NOT EDIT! + +from __future__ import annotations + +import dataclasses +import enum +import typing +import typing_extensions +import datetime +import google.protobuf.timestamp_pb2 +import temporalio.api.common.v1.message_pb2 +import temporalio.api.enums.v1.common_pb2 +import temporalio.api.enums.v1.stream_pb2 +import temporalio.api.stream.v1.message_pb2 +import temporalio.api.streamservice.v1.request_response_pb2 +import temporalio.converter + +import temporal_support as _support + + +class StreamProducerKind(enum.IntEnum): + StreamProducerKindUnspecified = 0 + StreamProducerKindWorkflow = 1 + StreamProducerKindLocalActivity = 2 + StreamProducerKindActivity = 3 + StreamProducerKindClient = 4 + StreamProducerKindStandaloneActivity = 5 + + +class ExecutionType(enum.IntEnum): + ExecutionTypeUnspecified = 0 + ExecutionTypeWorkflow = 1 + ExecutionTypeActivity = 2 + ExecutionTypeNexusOperation = 3 + + +class StreamReadStart(enum.IntEnum): + StreamReadStartUnspecified = 0 + StreamReadStartBeginning = 1 + StreamReadStartEnd = 2 + + +class StreamCloseStatus(enum.IntEnum): + StreamCloseStatusUnspecified = 0 + StreamCloseStatusCompleted = 1 + StreamCloseStatusFailed = 2 + StreamCloseStatusCanceled = 3 + StreamCloseStatusTerminated = 4 + StreamCloseStatusTimedOut = 5 + + +class StreamStatus(enum.IntEnum): + StreamStatusUnspecified = 0 + StreamStatusOpen = 1 + StreamStatusClosed = 2 + StreamStatusExpired = 3 + + +@dataclasses.dataclass(slots=True) +class CreateStreamRequest: + """Create a Stream, or get the open one. A closed, expired or another owner's Stream + fails. + """ + + id: str + retention: datetime.timedelta | None = None + withdraw_run: str | None = None + owner: Producer | None = None + ref: StreamRef | None = None + + +class _CreateStreamRequestTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + CreateStreamRequest, + temporalio.api.streamservice.v1.request_response_pb2.CreateStreamRequest, + ] +): + transfer_type: ( + type[temporalio.api.streamservice.v1.request_response_pb2.CreateStreamRequest] + | None + ) = temporalio.api.streamservice.v1.request_response_pb2.CreateStreamRequest + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.streamservice.v1.request_response_pb2.CreateStreamRequest, + type_hint: type[CreateStreamRequest], + ) -> CreateStreamRequest: + if not value.stream_id: + raise ValueError("missing required field CreateStreamRequest.id") + id = value.stream_id + return CreateStreamRequest( + id=id, + retention=_support.duration_from_proto(value.retention) + if value.HasField("retention") + else None, + withdraw_run=value.withdraw_run if bool(value.withdraw_run) else None, + owner=_ProducerTransferTypeConverter().from_transfer_type( + value.owner, Producer + ) + if value.HasField("owner") + else None, + ref=_StreamRefTransferTypeConverter().from_transfer_type( + value.ref, StreamRef + ) + if value.HasField("ref") + else None, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: CreateStreamRequest, + ) -> temporalio.api.streamservice.v1.request_response_pb2.CreateStreamRequest: + message = ( + temporalio.api.streamservice.v1.request_response_pb2.CreateStreamRequest() + ) + message.stream_id = value.id + if value.retention is not None: + message.retention.CopyFrom(_support.duration_to_proto(value.retention)) + if value.withdraw_run is not None: + message.withdraw_run = value.withdraw_run + if value.owner is not None: + message.owner.CopyFrom( + _ProducerTransferTypeConverter().to_transfer_type(value.owner) + ) + if value.ref is not None: + message.ref.CopyFrom( + _StreamRefTransferTypeConverter().to_transfer_type(value.ref) + ) + return message + + +temporalio.converter.transfer_type_convertible( + _CreateStreamRequestTransferTypeConverter +)(CreateStreamRequest) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class Producer: + """The Producer of a batch.""" + + id: str + kind: StreamProducerKind | None = None + epoch: ProducerEpoch | None = None + run_id: str | None = None + + +class _ProducerTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + Producer, temporalio.api.stream.v1.message_pb2.Producer + ] +): + transfer_type: type[temporalio.api.stream.v1.message_pb2.Producer] | None = ( + temporalio.api.stream.v1.message_pb2.Producer + ) + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.stream.v1.message_pb2.Producer, + type_hint: type[Producer], + ) -> Producer: + if not value.id: + raise ValueError("missing required field Producer.id") + id = value.id + return Producer( + id=id, + kind=StreamProducerKind(value.kind) if value.kind != 0 else None, + epoch=_ProducerEpochTransferTypeConverter().from_transfer_type( + value.epoch, ProducerEpoch + ) + if value.HasField("epoch") + else None, + run_id=value.run_id if bool(value.run_id) else None, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: Producer, + ) -> temporalio.api.stream.v1.message_pb2.Producer: + message = temporalio.api.stream.v1.message_pb2.Producer() + message.id = value.id + if value.kind is not None: + message.kind = ( + temporalio.api.enums.v1.stream_pb2.StreamProducerKind.ValueType( + value.kind + ) + ) + if value.epoch is not None: + message.epoch.CopyFrom( + _ProducerEpochTransferTypeConverter().to_transfer_type(value.epoch) + ) + if value.run_id is not None: + message.run_id = value.run_id + return message + + +temporalio.converter.transfer_type_convertible(_ProducerTransferTypeConverter)(Producer) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class ProducerEpoch: + """The ordered attempt of a Producer. A higher epoch supersedes a lower one.""" + + task_started_event_id: int | None = None + task_attempt: int | None = None + task_started_time: Timestamp | None = None + attempt_scheduled_time: Timestamp | None = None + attempt: int | None = None + + +class _ProducerEpochTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + ProducerEpoch, temporalio.api.stream.v1.message_pb2.ProducerEpoch + ] +): + transfer_type: type[temporalio.api.stream.v1.message_pb2.ProducerEpoch] | None = ( + temporalio.api.stream.v1.message_pb2.ProducerEpoch + ) + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.stream.v1.message_pb2.ProducerEpoch, + type_hint: type[ProducerEpoch], + ) -> ProducerEpoch: + return ProducerEpoch( + task_started_event_id=value.task_started_event_id + if value.task_started_event_id != 0 + else None, + task_attempt=value.task_attempt if value.task_attempt != 0 else None, + task_started_time=_TimestampTransferTypeConverter().from_transfer_type( + value.task_started_time, Timestamp + ) + if value.HasField("task_started_time") + else None, + attempt_scheduled_time=_TimestampTransferTypeConverter().from_transfer_type( + value.attempt_scheduled_time, Timestamp + ) + if value.HasField("attempt_scheduled_time") + else None, + attempt=value.attempt if value.attempt != 0 else None, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: ProducerEpoch, + ) -> temporalio.api.stream.v1.message_pb2.ProducerEpoch: + message = temporalio.api.stream.v1.message_pb2.ProducerEpoch() + if value.task_started_event_id is not None: + message.task_started_event_id = value.task_started_event_id + if value.task_attempt is not None: + message.task_attempt = value.task_attempt + if value.task_started_time is not None: + message.task_started_time.CopyFrom( + _TimestampTransferTypeConverter().to_transfer_type( + value.task_started_time + ) + ) + if value.attempt_scheduled_time is not None: + message.attempt_scheduled_time.CopyFrom( + _TimestampTransferTypeConverter().to_transfer_type( + value.attempt_scheduled_time + ) + ) + if value.attempt is not None: + message.attempt = value.attempt + return message + + +temporalio.converter.transfer_type_convertible(_ProducerEpochTransferTypeConverter)( + ProducerEpoch +) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class Timestamp: + seconds: int | None = None + nanos: int | None = None + + +class _TimestampTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + Timestamp, google.protobuf.timestamp_pb2.Timestamp + ] +): + transfer_type: type[google.protobuf.timestamp_pb2.Timestamp] | None = ( + google.protobuf.timestamp_pb2.Timestamp + ) + + @typing_extensions.override + def from_transfer_type( + self, + value: google.protobuf.timestamp_pb2.Timestamp, + type_hint: type[Timestamp], + ) -> Timestamp: + return Timestamp( + seconds=value.seconds if value.seconds != 0 else None, + nanos=value.nanos if value.nanos != 0 else None, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: Timestamp, + ) -> google.protobuf.timestamp_pb2.Timestamp: + message = google.protobuf.timestamp_pb2.Timestamp() + if value.seconds is not None: + message.seconds = value.seconds + if value.nanos is not None: + message.nanos = value.nanos + return message + + +temporalio.converter.transfer_type_convertible(_TimestampTransferTypeConverter)( + Timestamp +) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class StreamRef: + """Identifies one Stream. The id is opaque to callers.""" + + id: str + owner: Execution | None = None + backend: str | None = None + + +class _StreamRefTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + StreamRef, temporalio.api.stream.v1.message_pb2.StreamRef + ] +): + transfer_type: type[temporalio.api.stream.v1.message_pb2.StreamRef] | None = ( + temporalio.api.stream.v1.message_pb2.StreamRef + ) + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.stream.v1.message_pb2.StreamRef, + type_hint: type[StreamRef], + ) -> StreamRef: + if not value.id: + raise ValueError("missing required field StreamRef.id") + id = value.id + return StreamRef( + id=id, + owner=_ExecutionTransferTypeConverter().from_transfer_type( + value.owner, Execution + ) + if value.HasField("owner") + else None, + backend=value.backend if bool(value.backend) else None, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: StreamRef, + ) -> temporalio.api.stream.v1.message_pb2.StreamRef: + message = temporalio.api.stream.v1.message_pb2.StreamRef() + message.id = value.id + if value.owner is not None: + message.owner.CopyFrom( + _ExecutionTransferTypeConverter().to_transfer_type(value.owner) + ) + if value.backend is not None: + message.backend = value.backend + return message + + +temporalio.converter.transfer_type_convertible(_StreamRefTransferTypeConverter)( + StreamRef +) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class Execution: + """A Workflow run chain or a Standalone Activity run.""" + + type: ExecutionType | None = None + business_id: str | None = None + run_id: str | None = None + + +class _ExecutionTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + Execution, temporalio.api.common.v1.message_pb2.Execution + ] +): + transfer_type: type[temporalio.api.common.v1.message_pb2.Execution] | None = ( + temporalio.api.common.v1.message_pb2.Execution + ) + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.common.v1.message_pb2.Execution, + type_hint: type[Execution], + ) -> Execution: + return Execution( + type=ExecutionType(value.type) if value.type != 0 else None, + business_id=value.business_id if bool(value.business_id) else None, + run_id=value.run_id if bool(value.run_id) else None, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: Execution, + ) -> temporalio.api.common.v1.message_pb2.Execution: + message = temporalio.api.common.v1.message_pb2.Execution() + if value.type is not None: + message.type = temporalio.api.enums.v1.common_pb2.ExecutionType.ValueType( + value.type + ) + if value.business_id is not None: + message.business_id = value.business_id + if value.run_id is not None: + message.run_id = value.run_id + return message + + +temporalio.converter.transfer_type_convertible(_ExecutionTransferTypeConverter)( + Execution +) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class CreateStreamResult: + stream: StreamRef | None = None + created: bool | None = None + + +class _CreateStreamResultTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + CreateStreamResult, + temporalio.api.streamservice.v1.request_response_pb2.CreateStreamResponse, + ] +): + transfer_type: ( + type[temporalio.api.streamservice.v1.request_response_pb2.CreateStreamResponse] + | None + ) = temporalio.api.streamservice.v1.request_response_pb2.CreateStreamResponse + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.streamservice.v1.request_response_pb2.CreateStreamResponse, + type_hint: type[CreateStreamResult], + ) -> CreateStreamResult: + return CreateStreamResult( + stream=_StreamRefTransferTypeConverter().from_transfer_type( + value.stream, StreamRef + ) + if value.HasField("stream") + else None, + created=value.created if bool(value.created) else None, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: CreateStreamResult, + ) -> temporalio.api.streamservice.v1.request_response_pb2.CreateStreamResponse: + message = ( + temporalio.api.streamservice.v1.request_response_pb2.CreateStreamResponse() + ) + if value.stream is not None: + message.stream.CopyFrom( + _StreamRefTransferTypeConverter().to_transfer_type(value.stream) + ) + if value.created is not None: + message.created = value.created + return message + + +temporalio.converter.transfer_type_convertible( + _CreateStreamResultTransferTypeConverter +)(CreateStreamResult) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class AppendRecordsRequest: + """Append. FAILED_PRECONDITION if closed, fenced or format too new. NOT_FOUND if + absent. + """ + + stream: StreamRef + records: list[AppendRecord] | None = dataclasses.field(default_factory=list) + producer: Producer | None = None + + +class _AppendRecordsRequestTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + AppendRecordsRequest, + temporalio.api.streamservice.v1.request_response_pb2.AppendRecordsRequest, + ] +): + transfer_type: ( + type[temporalio.api.streamservice.v1.request_response_pb2.AppendRecordsRequest] + | None + ) = temporalio.api.streamservice.v1.request_response_pb2.AppendRecordsRequest + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.streamservice.v1.request_response_pb2.AppendRecordsRequest, + type_hint: type[AppendRecordsRequest], + ) -> AppendRecordsRequest: + if not value.HasField("stream"): + raise ValueError("missing required field AppendRecordsRequest.stream") + stream = _StreamRefTransferTypeConverter().from_transfer_type( + value.stream, StreamRef + ) + return AppendRecordsRequest( + stream=stream, + records=[ + _AppendRecordTransferTypeConverter().from_transfer_type( + item, AppendRecord + ) + for item in value.records + ], + producer=_ProducerTransferTypeConverter().from_transfer_type( + value.producer, Producer + ) + if value.HasField("producer") + else None, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: AppendRecordsRequest, + ) -> temporalio.api.streamservice.v1.request_response_pb2.AppendRecordsRequest: + message = ( + temporalio.api.streamservice.v1.request_response_pb2.AppendRecordsRequest() + ) + message.stream.CopyFrom( + _StreamRefTransferTypeConverter().to_transfer_type(value.stream) + ) + if value.records: + for element in value.records: + item = message.records.add() + item.CopyFrom( + _AppendRecordTransferTypeConverter().to_transfer_type(element) + ) + if value.producer is not None: + message.producer.CopyFrom( + _ProducerTransferTypeConverter().to_transfer_type(value.producer) + ) + return message + + +temporalio.converter.transfer_type_convertible( + _AppendRecordsRequestTransferTypeConverter +)(AppendRecordsRequest) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class AppendRecord: + """One Record to append.""" + + body: typing.Any + + +class _AppendRecordTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + AppendRecord, temporalio.api.stream.v1.message_pb2.AppendRecord + ] +): + transfer_type: type[temporalio.api.stream.v1.message_pb2.AppendRecord] | None = ( + temporalio.api.stream.v1.message_pb2.AppendRecord + ) + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.stream.v1.message_pb2.AppendRecord, + type_hint: type[AppendRecord], + ) -> AppendRecord: + if not value.HasField("body"): + raise ValueError("missing required field AppendRecord.body") + body = _support.payload_from_proto(value.body) + return AppendRecord( + body=body, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: AppendRecord, + ) -> temporalio.api.stream.v1.message_pb2.AppendRecord: + message = temporalio.api.stream.v1.message_pb2.AppendRecord() + message.body.CopyFrom(_support.payload_to_proto(value.body)) + return message + + +temporalio.converter.transfer_type_convertible(_AppendRecordTransferTypeConverter)( + AppendRecord +) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class AppendRecordsResponse: + first_offset: int | None = None + last_offset: int | None = None + + +class _AppendRecordsResponseTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + AppendRecordsResponse, + temporalio.api.streamservice.v1.request_response_pb2.AppendRecordsResponse, + ] +): + transfer_type: ( + type[temporalio.api.streamservice.v1.request_response_pb2.AppendRecordsResponse] + | None + ) = temporalio.api.streamservice.v1.request_response_pb2.AppendRecordsResponse + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.streamservice.v1.request_response_pb2.AppendRecordsResponse, + type_hint: type[AppendRecordsResponse], + ) -> AppendRecordsResponse: + return AppendRecordsResponse( + first_offset=value.first_offset if value.first_offset != 0 else None, + last_offset=value.last_offset if value.last_offset != 0 else None, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: AppendRecordsResponse, + ) -> temporalio.api.streamservice.v1.request_response_pb2.AppendRecordsResponse: + message = ( + temporalio.api.streamservice.v1.request_response_pb2.AppendRecordsResponse() + ) + if value.first_offset is not None: + message.first_offset = value.first_offset + if value.last_offset is not None: + message.last_offset = value.last_offset + return message + + +temporalio.converter.transfer_type_convertible( + _AppendRecordsResponseTransferTypeConverter +)(AppendRecordsResponse) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class CloseStreamRequest: + """Close a Stream. Readers drain what is left, then end. Later appends are refused.""" + + stream: StreamRef + producer: Producer | None = None + result: typing.Any | None = None + failure: BaseException | None = None + + +class _CloseStreamRequestTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + CloseStreamRequest, + temporalio.api.streamservice.v1.request_response_pb2.CloseStreamRequest, + ] +): + transfer_type: ( + type[temporalio.api.streamservice.v1.request_response_pb2.CloseStreamRequest] + | None + ) = temporalio.api.streamservice.v1.request_response_pb2.CloseStreamRequest + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.streamservice.v1.request_response_pb2.CloseStreamRequest, + type_hint: type[CloseStreamRequest], + ) -> CloseStreamRequest: + if not value.HasField("stream"): + raise ValueError("missing required field CloseStreamRequest.stream") + stream = _StreamRefTransferTypeConverter().from_transfer_type( + value.stream, StreamRef + ) + return CloseStreamRequest( + stream=stream, + producer=_ProducerTransferTypeConverter().from_transfer_type( + value.producer, Producer + ) + if value.HasField("producer") + else None, + result=_support.payload_from_proto(value.result) + if value.HasField("result") + else None, + failure=_support.failure_from_proto(value.failure) + if value.HasField("failure") + else None, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: CloseStreamRequest, + ) -> temporalio.api.streamservice.v1.request_response_pb2.CloseStreamRequest: + message = ( + temporalio.api.streamservice.v1.request_response_pb2.CloseStreamRequest() + ) + message.stream.CopyFrom( + _StreamRefTransferTypeConverter().to_transfer_type(value.stream) + ) + if value.producer is not None: + message.producer.CopyFrom( + _ProducerTransferTypeConverter().to_transfer_type(value.producer) + ) + if value.result is not None: + message.result.CopyFrom(_support.payload_to_proto(value.result)) + if value.failure is not None: + message.failure.CopyFrom(_support.failure_to_proto(value.failure)) + return message + + +temporalio.converter.transfer_type_convertible( + _CloseStreamRequestTransferTypeConverter +)(CloseStreamRequest) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class CloseStreamResponse: + pass + + +class _CloseStreamResponseTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + CloseStreamResponse, + temporalio.api.streamservice.v1.request_response_pb2.CloseStreamResponse, + ] +): + transfer_type: ( + type[temporalio.api.streamservice.v1.request_response_pb2.CloseStreamResponse] + | None + ) = temporalio.api.streamservice.v1.request_response_pb2.CloseStreamResponse + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.streamservice.v1.request_response_pb2.CloseStreamResponse, + type_hint: type[CloseStreamResponse], + ) -> CloseStreamResponse: + return CloseStreamResponse() + + @typing_extensions.override + def to_transfer_type( + self, + value: CloseStreamResponse, + ) -> temporalio.api.streamservice.v1.request_response_pb2.CloseStreamResponse: + message = ( + temporalio.api.streamservice.v1.request_response_pb2.CloseStreamResponse() + ) + return message + + +temporalio.converter.transfer_type_convertible( + _CloseStreamResponseTransferTypeConverter +)(CloseStreamResponse) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class ReadRecordsRequest: + """Read Records from an offset.""" + + stream: StreamRef + from_offset: int | None = None + start: StreamReadStart | None = None + max_records: int | None = None + wait: datetime.timedelta | None = None + create_wait: datetime.timedelta | None = None + last: int | None = None + max_bytes: int | None = None + + +class _ReadRecordsRequestTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + ReadRecordsRequest, + temporalio.api.streamservice.v1.request_response_pb2.ReadRecordsRequest, + ] +): + transfer_type: ( + type[temporalio.api.streamservice.v1.request_response_pb2.ReadRecordsRequest] + | None + ) = temporalio.api.streamservice.v1.request_response_pb2.ReadRecordsRequest + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.streamservice.v1.request_response_pb2.ReadRecordsRequest, + type_hint: type[ReadRecordsRequest], + ) -> ReadRecordsRequest: + if not value.HasField("stream"): + raise ValueError("missing required field ReadRecordsRequest.stream") + stream = _StreamRefTransferTypeConverter().from_transfer_type( + value.stream, StreamRef + ) + return ReadRecordsRequest( + stream=stream, + from_offset=value.from_offset if value.from_offset != 0 else None, + start=StreamReadStart(value.start) if value.start != 0 else None, + max_records=value.max_records if value.max_records != 0 else None, + wait=_support.duration_from_proto(value.wait) + if value.HasField("wait") + else None, + create_wait=_support.duration_from_proto(value.create_wait) + if value.HasField("create_wait") + else None, + last=value.last if value.last != 0 else None, + max_bytes=value.max_bytes if value.max_bytes != 0 else None, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: ReadRecordsRequest, + ) -> temporalio.api.streamservice.v1.request_response_pb2.ReadRecordsRequest: + message = ( + temporalio.api.streamservice.v1.request_response_pb2.ReadRecordsRequest() + ) + message.stream.CopyFrom( + _StreamRefTransferTypeConverter().to_transfer_type(value.stream) + ) + if value.from_offset is not None: + message.from_offset = value.from_offset + if value.start is not None: + message.start = ( + temporalio.api.enums.v1.stream_pb2.StreamReadStart.ValueType( + value.start + ) + ) + if value.max_records is not None: + message.max_records = value.max_records + if value.wait is not None: + message.wait.CopyFrom(_support.duration_to_proto(value.wait)) + if value.create_wait is not None: + message.create_wait.CopyFrom(_support.duration_to_proto(value.create_wait)) + if value.last is not None: + message.last = value.last + if value.max_bytes is not None: + message.max_bytes = value.max_bytes + return message + + +temporalio.converter.transfer_type_convertible( + _ReadRecordsRequestTransferTypeConverter +)(ReadRecordsRequest) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class ReadRecordsResponse: + records: list[StreamRecord] | None = dataclasses.field(default_factory=list) + next_offset: int | None = None + done: bool | None = None + + +class _ReadRecordsResponseTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + ReadRecordsResponse, + temporalio.api.streamservice.v1.request_response_pb2.ReadRecordsResponse, + ] +): + transfer_type: ( + type[temporalio.api.streamservice.v1.request_response_pb2.ReadRecordsResponse] + | None + ) = temporalio.api.streamservice.v1.request_response_pb2.ReadRecordsResponse + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.streamservice.v1.request_response_pb2.ReadRecordsResponse, + type_hint: type[ReadRecordsResponse], + ) -> ReadRecordsResponse: + return ReadRecordsResponse( + records=[ + _StreamRecordTransferTypeConverter().from_transfer_type( + item, StreamRecord + ) + for item in value.records + ], + next_offset=value.next_offset if value.next_offset != 0 else None, + done=value.done if bool(value.done) else None, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: ReadRecordsResponse, + ) -> temporalio.api.streamservice.v1.request_response_pb2.ReadRecordsResponse: + message = ( + temporalio.api.streamservice.v1.request_response_pb2.ReadRecordsResponse() + ) + if value.records: + for element in value.records: + item = message.records.add() + item.CopyFrom( + _StreamRecordTransferTypeConverter().to_transfer_type(element) + ) + if value.next_offset is not None: + message.next_offset = value.next_offset + if value.done is not None: + message.done = value.done + return message + + +temporalio.converter.transfer_type_convertible( + _ReadRecordsResponseTransferTypeConverter +)(ReadRecordsResponse) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class StreamRecord: + """One item that a read gives.""" + + item: StreamItem | None = None + offset: int | None = None + + +class _StreamRecordTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + StreamRecord, temporalio.api.stream.v1.message_pb2.StreamRecord + ] +): + transfer_type: type[temporalio.api.stream.v1.message_pb2.StreamRecord] | None = ( + temporalio.api.stream.v1.message_pb2.StreamRecord + ) + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.stream.v1.message_pb2.StreamRecord, + type_hint: type[StreamRecord], + ) -> StreamRecord: + _oneof_item: StreamItemData | StreamItemSuperseded | StreamItemClose | None + match value.WhichOneof("item"): + case "data": + _oneof_item = StreamItemData( + _StreamDataTransferTypeConverter().from_transfer_type( + value.data, StreamData + ) + ) + case "superseded": + _oneof_item = StreamItemSuperseded( + _SupersessionTransferTypeConverter().from_transfer_type( + value.superseded, Supersession + ) + ) + case "close": + _oneof_item = StreamItemClose( + _StreamCloseTransferTypeConverter().from_transfer_type( + value.close, StreamClose + ) + ) + case None: + _oneof_item = None + return StreamRecord( + item=_oneof_item, + offset=value.offset if value.offset != 0 else None, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: StreamRecord, + ) -> temporalio.api.stream.v1.message_pb2.StreamRecord: + message = temporalio.api.stream.v1.message_pb2.StreamRecord() + if value.item is not None: + match value.item: + case StreamItemData(): + message.data.CopyFrom( + _StreamDataTransferTypeConverter().to_transfer_type( + value.item.value + ) + ) + case StreamItemSuperseded(): + message.superseded.CopyFrom( + _SupersessionTransferTypeConverter().to_transfer_type( + value.item.value + ) + ) + case StreamItemClose(): + message.close.CopyFrom( + _StreamCloseTransferTypeConverter().to_transfer_type( + value.item.value + ) + ) + if value.offset is not None: + message.offset = value.offset + return message + + +temporalio.converter.transfer_type_convertible(_StreamRecordTransferTypeConverter)( + StreamRecord +) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class StreamData: + """One Record that a Producer appended.""" + + body: typing.Any + producer: Producer | None = None + + +class _StreamDataTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + StreamData, temporalio.api.stream.v1.message_pb2.StreamData + ] +): + transfer_type: type[temporalio.api.stream.v1.message_pb2.StreamData] | None = ( + temporalio.api.stream.v1.message_pb2.StreamData + ) + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.stream.v1.message_pb2.StreamData, + type_hint: type[StreamData], + ) -> StreamData: + if not value.HasField("body"): + raise ValueError("missing required field StreamData.body") + body = _support.payload_from_proto(value.body) + return StreamData( + body=body, + producer=_ProducerTransferTypeConverter().from_transfer_type( + value.producer, Producer + ) + if value.HasField("producer") + else None, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: StreamData, + ) -> temporalio.api.stream.v1.message_pb2.StreamData: + message = temporalio.api.stream.v1.message_pb2.StreamData() + message.body.CopyFrom(_support.payload_to_proto(value.body)) + if value.producer is not None: + message.producer.CopyFrom( + _ProducerTransferTypeConverter().to_transfer_type(value.producer) + ) + return message + + +temporalio.converter.transfer_type_convertible(_StreamDataTransferTypeConverter)( + StreamData +) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class Supersession: + """Which earlier Records a Superseded item withdraws.""" + + producer_id: str + failed_tasks: list[int] | None = dataclasses.field(default_factory=list) + producer_kind: StreamProducerKind | None = None + epoch: ProducerEpoch | None = None + previous_completed_task: WorkflowTaskEpoch | None = None + withdraw_run: str | None = None + run_id: str | None = None + + +class _SupersessionTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + Supersession, temporalio.api.stream.v1.message_pb2.Supersession + ] +): + transfer_type: type[temporalio.api.stream.v1.message_pb2.Supersession] | None = ( + temporalio.api.stream.v1.message_pb2.Supersession + ) + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.stream.v1.message_pb2.Supersession, + type_hint: type[Supersession], + ) -> Supersession: + if not value.producer_id: + raise ValueError("missing required field Supersession.producer_id") + producer_id = value.producer_id + return Supersession( + producer_id=producer_id, + failed_tasks=list(value.failed_tasks), + producer_kind=StreamProducerKind(value.producer_kind) + if value.producer_kind != 0 + else None, + epoch=_ProducerEpochTransferTypeConverter().from_transfer_type( + value.epoch, ProducerEpoch + ) + if value.HasField("epoch") + else None, + previous_completed_task=_WorkflowTaskEpochTransferTypeConverter().from_transfer_type( + value.previous_completed_task, WorkflowTaskEpoch + ) + if value.HasField("previous_completed_task") + else None, + withdraw_run=value.withdraw_run if bool(value.withdraw_run) else None, + run_id=value.run_id if bool(value.run_id) else None, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: Supersession, + ) -> temporalio.api.stream.v1.message_pb2.Supersession: + message = temporalio.api.stream.v1.message_pb2.Supersession() + message.producer_id = value.producer_id + if value.failed_tasks: + message.failed_tasks.extend(value.failed_tasks) + if value.producer_kind is not None: + message.producer_kind = ( + temporalio.api.enums.v1.stream_pb2.StreamProducerKind.ValueType( + value.producer_kind + ) + ) + if value.epoch is not None: + message.epoch.CopyFrom( + _ProducerEpochTransferTypeConverter().to_transfer_type(value.epoch) + ) + if value.previous_completed_task is not None: + message.previous_completed_task.CopyFrom( + _WorkflowTaskEpochTransferTypeConverter().to_transfer_type( + value.previous_completed_task + ) + ) + if value.withdraw_run is not None: + message.withdraw_run = value.withdraw_run + if value.run_id is not None: + message.run_id = value.run_id + return message + + +temporalio.converter.transfer_type_convertible(_SupersessionTransferTypeConverter)( + Supersession +) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class WorkflowTaskEpoch: + """A Workflow Task attempt of the run.""" + + started_event_id: int | None = None + attempt: int | None = None + started_time: Timestamp | None = None + + +class _WorkflowTaskEpochTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + WorkflowTaskEpoch, temporalio.api.stream.v1.message_pb2.WorkflowTaskEpoch + ] +): + transfer_type: ( + type[temporalio.api.stream.v1.message_pb2.WorkflowTaskEpoch] | None + ) = temporalio.api.stream.v1.message_pb2.WorkflowTaskEpoch + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.stream.v1.message_pb2.WorkflowTaskEpoch, + type_hint: type[WorkflowTaskEpoch], + ) -> WorkflowTaskEpoch: + return WorkflowTaskEpoch( + started_event_id=value.started_event_id + if value.started_event_id != 0 + else None, + attempt=value.attempt if value.attempt != 0 else None, + started_time=_TimestampTransferTypeConverter().from_transfer_type( + value.started_time, Timestamp + ) + if value.HasField("started_time") + else None, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: WorkflowTaskEpoch, + ) -> temporalio.api.stream.v1.message_pb2.WorkflowTaskEpoch: + message = temporalio.api.stream.v1.message_pb2.WorkflowTaskEpoch() + if value.started_event_id is not None: + message.started_event_id = value.started_event_id + if value.attempt is not None: + message.attempt = value.attempt + if value.started_time is not None: + message.started_time.CopyFrom( + _TimestampTransferTypeConverter().to_transfer_type(value.started_time) + ) + return message + + +temporalio.converter.transfer_type_convertible(_WorkflowTaskEpochTransferTypeConverter)( + WorkflowTaskEpoch +) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class StreamClose: + """The close of a Stream. A read gives it last.""" + + producer: Producer | None = None + result: typing.Any | None = None + failure: BaseException | None = None + status: StreamCloseStatus | None = None + + +class _StreamCloseTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + StreamClose, temporalio.api.stream.v1.message_pb2.StreamClose + ] +): + transfer_type: type[temporalio.api.stream.v1.message_pb2.StreamClose] | None = ( + temporalio.api.stream.v1.message_pb2.StreamClose + ) + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.stream.v1.message_pb2.StreamClose, + type_hint: type[StreamClose], + ) -> StreamClose: + return StreamClose( + producer=_ProducerTransferTypeConverter().from_transfer_type( + value.producer, Producer + ) + if value.HasField("producer") + else None, + result=_support.payload_from_proto(value.result) + if value.HasField("result") + else None, + failure=_support.failure_from_proto(value.failure) + if value.HasField("failure") + else None, + status=StreamCloseStatus(value.status) if value.status != 0 else None, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: StreamClose, + ) -> temporalio.api.stream.v1.message_pb2.StreamClose: + message = temporalio.api.stream.v1.message_pb2.StreamClose() + if value.producer is not None: + message.producer.CopyFrom( + _ProducerTransferTypeConverter().to_transfer_type(value.producer) + ) + if value.result is not None: + message.result.CopyFrom(_support.payload_to_proto(value.result)) + if value.failure is not None: + message.failure.CopyFrom(_support.failure_to_proto(value.failure)) + if value.status is not None: + message.status = ( + temporalio.api.enums.v1.stream_pb2.StreamCloseStatus.ValueType( + value.status + ) + ) + return message + + +temporalio.converter.transfer_type_convertible(_StreamCloseTransferTypeConverter)( + StreamClose +) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class DescribeStreamRequest: + stream: StreamRef + + +class _DescribeStreamRequestTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + DescribeStreamRequest, + temporalio.api.streamservice.v1.request_response_pb2.DescribeStreamRequest, + ] +): + transfer_type: ( + type[temporalio.api.streamservice.v1.request_response_pb2.DescribeStreamRequest] + | None + ) = temporalio.api.streamservice.v1.request_response_pb2.DescribeStreamRequest + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.streamservice.v1.request_response_pb2.DescribeStreamRequest, + type_hint: type[DescribeStreamRequest], + ) -> DescribeStreamRequest: + if not value.HasField("stream"): + raise ValueError("missing required field DescribeStreamRequest.stream") + stream = _StreamRefTransferTypeConverter().from_transfer_type( + value.stream, StreamRef + ) + return DescribeStreamRequest( + stream=stream, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: DescribeStreamRequest, + ) -> temporalio.api.streamservice.v1.request_response_pb2.DescribeStreamRequest: + message = ( + temporalio.api.streamservice.v1.request_response_pb2.DescribeStreamRequest() + ) + message.stream.CopyFrom( + _StreamRefTransferTypeConverter().to_transfer_type(value.stream) + ) + return message + + +temporalio.converter.transfer_type_convertible( + _DescribeStreamRequestTransferTypeConverter +)(DescribeStreamRequest) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class DescribeStreamResponse: + status: StreamStatus | None = None + last_offset: int | None = None + retention: datetime.timedelta | None = None + + +class _DescribeStreamResponseTransferTypeConverter( + temporalio.converter.TransferTypeConverter[ + DescribeStreamResponse, + temporalio.api.streamservice.v1.request_response_pb2.DescribeStreamResponse, + ] +): + transfer_type: ( + type[ + temporalio.api.streamservice.v1.request_response_pb2.DescribeStreamResponse + ] + | None + ) = temporalio.api.streamservice.v1.request_response_pb2.DescribeStreamResponse + + @typing_extensions.override + def from_transfer_type( + self, + value: temporalio.api.streamservice.v1.request_response_pb2.DescribeStreamResponse, + type_hint: type[DescribeStreamResponse], + ) -> DescribeStreamResponse: + return DescribeStreamResponse( + status=StreamStatus(value.status) if value.status != 0 else None, + last_offset=value.last_offset if value.last_offset != 0 else None, + retention=_support.duration_from_proto(value.retention) + if value.HasField("retention") + else None, + ) + + @typing_extensions.override + def to_transfer_type( + self, + value: DescribeStreamResponse, + ) -> temporalio.api.streamservice.v1.request_response_pb2.DescribeStreamResponse: + message = temporalio.api.streamservice.v1.request_response_pb2.DescribeStreamResponse() + if value.status is not None: + message.status = temporalio.api.enums.v1.stream_pb2.StreamStatus.ValueType( + value.status + ) + if value.last_offset is not None: + message.last_offset = value.last_offset + if value.retention is not None: + message.retention.CopyFrom(_support.duration_to_proto(value.retention)) + return message + + +temporalio.converter.transfer_type_convertible( + _DescribeStreamResponseTransferTypeConverter +)(DescribeStreamResponse) # pyright: ignore[reportUnusedCallResult] + + +@dataclasses.dataclass(slots=True) +class StreamItemData: + value: StreamData + + +@dataclasses.dataclass(slots=True) +class StreamItemSuperseded: + value: Supersession + + +@dataclasses.dataclass(slots=True) +class StreamItemClose: + value: StreamClose + + +StreamItem = StreamItemData | StreamItemSuperseded | StreamItemClose +"""A read item: a Record, a Superseded item or the close.""" diff --git a/advanced/samples/python/wit/stream_service/operations/__init__.py b/advanced/samples/python/wit/stream_service/operations/__init__.py new file mode 100644 index 000000000..ca23418da --- /dev/null +++ b/advanced/samples/python/wit/stream_service/operations/__init__.py @@ -0,0 +1,3 @@ +# Generated by nexgen latest. DO NOT EDIT! + +from __future__ import annotations diff --git a/advanced/samples/python/wit/stream_service/operations/create_stream.py b/advanced/samples/python/wit/stream_service/operations/create_stream.py new file mode 100644 index 000000000..20afe2b12 --- /dev/null +++ b/advanced/samples/python/wit/stream_service/operations/create_stream.py @@ -0,0 +1,62 @@ +# Generated by nexgen latest. DO NOT EDIT! + +from __future__ import annotations + +import datetime +import temporalio.workflow + +from ..models import ( + CreateStreamRequest, + CreateStreamResult, + Producer, + StreamRef, +) +from .._resources import Stream + + +async def _create_stream( + request: CreateStreamRequest, +) -> Stream: + nexus_client = temporalio.workflow.create_nexus_client( + service="temporal.api.streamservice.v1.StreamService", + endpoint="__temporal_system", + ) + handle = await nexus_client.start_operation( + operation="CreateStream", + input=request, + output_type=CreateStreamResult, + ) + result = await handle + if result.stream is None: + raise RuntimeError("CreateStream answered without `stream`") + return Stream( + stream=result.stream, + created=result.created or None, + ) + + +async def create_stream( + *, + id: str, + retention: datetime.timedelta | None = None, + withdraw_run: str | None = None, + owner: Producer | None = None, + ref: StreamRef | None = None, +) -> Stream: + """ + Args: + id: The Stream id. Required, and never empty. + retention: The age after which a Record is dropped. Empty is the Store default. + withdraw_run: Retry run only. The previous run, whose Records this withdraws. + owner: The Workflow run that owns the Stream. The Workflow transport sets it. + ref: The ref, whose id is stream-id. A create with an owner needs the ref's + owner. + """ + request = CreateStreamRequest( + id=id, + retention=retention, + withdraw_run=withdraw_run, + owner=owner, + ref=ref, + ) + return await _create_stream(request) diff --git a/advanced/samples/python/wit/stream_service/services.py b/advanced/samples/python/wit/stream_service/services.py new file mode 100644 index 000000000..c22151607 --- /dev/null +++ b/advanced/samples/python/wit/stream_service/services.py @@ -0,0 +1,50 @@ +# Generated by nexgen latest. DO NOT EDIT! + +from __future__ import annotations + +from nexusrpc import Operation, service +from .models import ( + AppendRecordsRequest, + AppendRecordsResponse, + CloseStreamRequest, + CloseStreamResponse, + CreateStreamRequest, + CreateStreamResult, + DescribeStreamRequest, + DescribeStreamResponse, + ReadRecordsRequest, + ReadRecordsResponse, +) + + +@service(name="temporal.api.streamservice.v1.StreamService") +class StreamService: + """ + .. warning:: + This API is experimental and subject to change. + """ + + create_stream: Operation[ + CreateStreamRequest, + CreateStreamResult, + ] = Operation(name="CreateStream") + + append_records: Operation[ + AppendRecordsRequest, + AppendRecordsResponse, + ] = Operation(name="AppendRecords") + + close_stream: Operation[ + CloseStreamRequest, + CloseStreamResponse, + ] = Operation(name="CloseStream") + + read_records: Operation[ + ReadRecordsRequest, + ReadRecordsResponse, + ] = Operation(name="ReadRecords") + + describe_stream: Operation[ + DescribeStreamRequest, + DescribeStreamResponse, + ] = Operation(name="DescribeStream")