diff --git a/pyproject.toml b/pyproject.toml index 7eb3c5df3e..76fc34b294 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -185,6 +185,7 @@ exclude = [ 'temporalio/api', 'temporalio/bridge/proto', 'temporalio/nexus/system/workflow_service', + 'temporalio/streams/proto', # The Core submodule, which has a script of the crate's test vectors 'temporalio/bridge/sdk-core', ] @@ -193,6 +194,12 @@ exclude = [ module = "temporalio.nexus.system.workflow_service.*" ignore_errors = true +[[tool.mypy.overrides]] +module = [ + "temporalio.streams.proto.*", +] +ignore_errors = true + [tool.pydocstyle] convention = "google" # https://github.com/PyCQA/pydocstyle/issues/363#issuecomment-625563088 @@ -217,6 +224,7 @@ intersphinx = [ ] privacy = [ "PRIVATE:temporalio.bridge", + "PRIVATE:temporalio.streams.proto", "PRIVATE:temporalio.types", "PRIVATE:temporalio.worker.workflow_sandbox.restrictions", "PRIVATE:temporalio.worker.workflow_sandbox.runner", @@ -262,6 +270,7 @@ exclude = [ "temporalio/api", "temporalio/bridge/proto", "temporalio/bridge/_visitor.py", + "temporalio/streams/proto", "tests/worker/workflow_sandbox/testmodules/proto", # The Core submodule, which has a script of the crate's test vectors "temporalio/bridge/sdk-core", diff --git a/scripts/_proto/Dockerfile b/scripts/_proto/Dockerfile index 0f0fb74b06..2b6fdfe738 100644 --- a/scripts/_proto/Dockerfile +++ b/scripts/_proto/Dockerfile @@ -3,7 +3,7 @@ FROM python:3.10 RUN curl -LsSf https://astral.sh/uv/install.sh | sh ENV PATH="$PATH:/root/.local/bin" RUN uv tool install poethepoet -VOLUME ["/api_new", "/bridge_new"] +VOLUME ["/api_new", "/bridge_new", "/streams_new"] COPY ./ ./ @@ -13,4 +13,4 @@ RUN uv add "protobuf<4" RUN uv sync --all-extras RUN uv run scripts/gen_protos.py -CMD ["sh", "-c", "cp -r ./temporalio/api/* /api_new && cp -r ./temporalio/bridge/proto/* /bridge_new"] +CMD ["sh", "-c", "cp -r ./temporalio/api/* /api_new && cp -r ./temporalio/bridge/proto/* /bridge_new && cp -r ./temporalio/streams/proto/* /streams_new"] diff --git a/scripts/_proto/temporal/sdk/streams/v1/redis.proto b/scripts/_proto/temporal/sdk/streams/v1/redis.proto new file mode 100644 index 0000000000..8f9f252998 --- /dev/null +++ b/scripts/_proto/temporal/sdk/streams/v1/redis.proto @@ -0,0 +1,33 @@ +syntax = "proto3"; + +// The config of the Redis Stream runtime. It's the Redis Store's own config, so it isn't in +// sdk-core's `coresdk.streams` package. Its shared fields have the numbers and types of +// `coresdk.streams.MemoryStreamConfig`, so a backend can pass either config. Every SDK uses these +// field numbers. +package temporal.sdk.streams.v1; + +import "google/protobuf/duration.proto"; + +// The settings of the Redis Stream runtime. Lang builds it from the backend options. +message RedisStreamConfig { + // One URL, or the seed URLs of a cluster. Credentials and TLS come from the URL. + repeated string urls = 1; + bool cluster = 2; + // Only letters, digits, `-`, `.`, `_` and `~`. Empty is `temporal-streams`. + string key_prefix = 3; + // The default Retention of a new Stream. Empty is 7 days. + google.protobuf.Duration retention = 4; + // How long an open Workflow-owned Stream stays after its last write. Empty is 90 days. + google.protobuf.Duration safety_ttl = 5; + // Zero is 256. + int32 blocking_reads_per_node = 8; + // Empty is 5 s. + google.protobuf.Duration response_timeout = 9; + // The backend name that lang registered this runtime under. The crate sets it on each + // StreamRef that it builds, such as the one in a create response. + string backend = 10; + // How many replicas must have each write before it answers. Zero sends no WAIT. + int32 wait_replicas = 11; + // How long a write waits for its replicas, at most half of response_timeout. Empty is 1 s. + google.protobuf.Duration wait_timeout = 12; +} diff --git a/scripts/gen_protos.py b/scripts/gen_protos.py index 4b2ea04569..e1b60d025f 100644 --- a/scripts/gen_protos.py +++ b/scripts/gen_protos.py @@ -33,6 +33,7 @@ api_out_dir = base_dir / "temporalio" / "api" sdk_out_dir = base_dir / "temporalio" / "bridge" / "proto" +streams_out_dir = base_dir / "temporalio" / "streams" / "proto" py_fixes = [ partial(re.compile(r"from temporal\.api\.").sub, r"from temporalio.api."), @@ -50,6 +51,14 @@ partial( re.compile(r"from temporal\.sdk\.core\.").sub, r"from temporalio.bridge.proto." ), + partial( + re.compile(r"from temporal\.sdk\.streams\.").sub, + r"from temporalio.streams.proto.", + ), + partial( + re.compile(r"'__module__' : 'temporal\.sdk\.streams\.").sub, + r"'__module__' : 'temporalio.streams.proto.", + ), partial( re.compile(r"'__module__' : 'temporal\.api\.").sub, r"'__module__' : 'temporalio.api.", @@ -71,6 +80,10 @@ r"temporalio.api.dependencies.nexusannotations.", ), partial(re.compile(r"temporal\.sdk\.core\.").sub, r"temporalio.bridge.proto."), + partial( + re.compile(r"temporal\.sdk\.streams\.").sub, + r"temporalio.streams.proto.", + ), ] find_class_re = re.compile(r"\nclass ([^_\(\:]+)") @@ -218,6 +231,11 @@ def generate_protos(output_dir: Path): for p in (output_dir / "temporal" / "sdk" / "core").iterdir(): shutil.rmtree(sdk_out_dir / p.name, ignore_errors=True) p.replace(sdk_out_dir / p.name) + # The Redis Store's config isn't a Core proto, so it goes in the Streams package + streams_out_dir.mkdir(parents=True, exist_ok=True) + for p in (output_dir / "temporal" / "sdk" / "streams").iterdir(): + shutil.rmtree(streams_out_dir / p.name, ignore_errors=True) + p.replace(streams_out_dir / p.name) shutil.rmtree(sdk_out_dir / "health", ignore_errors=True) (output_dir / "health").replace(sdk_out_dir / "health") # Move test protos diff --git a/scripts/gen_protos_docker.py b/scripts/gen_protos_docker.py index 6735bdf90d..600b0923c1 100644 --- a/scripts/gen_protos_docker.py +++ b/scripts/gen_protos_docker.py @@ -34,6 +34,8 @@ os.path.join(os.getcwd(), "temporalio", "api") + ":/api_new", "-v", os.path.join(os.getcwd(), "temporalio", "bridge", "proto") + ":/bridge_new", + "-v", + os.path.join(os.getcwd(), "temporalio", "streams", "proto") + ":/streams_new", image_id, ] ) diff --git a/temporalio/streams/__init__.py b/temporalio/streams/__init__.py new file mode 100644 index 0000000000..23e86ba97f --- /dev/null +++ b/temporalio/streams/__init__.py @@ -0,0 +1 @@ +"""The Streams API, which Workflows, Activities and clients share.""" diff --git a/temporalio/streams/proto/__init__.py b/temporalio/streams/proto/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/temporalio/streams/proto/v1/__init__.py b/temporalio/streams/proto/v1/__init__.py new file mode 100644 index 0000000000..ccb1b9acd4 --- /dev/null +++ b/temporalio/streams/proto/v1/__init__.py @@ -0,0 +1,5 @@ +from .redis_pb2 import RedisStreamConfig + +__all__ = [ + "RedisStreamConfig", +] diff --git a/temporalio/streams/proto/v1/redis_pb2.py b/temporalio/streams/proto/v1/redis_pb2.py new file mode 100644 index 0000000000..bba567f752 --- /dev/null +++ b/temporalio/streams/proto/v1/redis_pb2.py @@ -0,0 +1,40 @@ +# -*- coding: utf-8 -*- +# Generated by the protocol buffer compiler. DO NOT EDIT! +# source: temporal/sdk/streams/v1/redis.proto +"""Generated protocol buffer code.""" + +from google.protobuf import descriptor as _descriptor +from google.protobuf import descriptor_pool as _descriptor_pool +from google.protobuf import message as _message +from google.protobuf import reflection as _reflection +from google.protobuf import symbol_database as _symbol_database + +# @@protoc_insertion_point(imports) + +_sym_db = _symbol_database.Default() + + +from google.protobuf import duration_pb2 as google_dot_protobuf_dot_duration__pb2 + +DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile( + b'\n#temporal/sdk/streams/v1/redis.proto\x12\x17temporal.sdk.streams.v1\x1a\x1egoogle/protobuf/duration.proto"\xd2\x02\n\x11RedisStreamConfig\x12\x0c\n\x04urls\x18\x01 \x03(\t\x12\x0f\n\x07\x63luster\x18\x02 \x01(\x08\x12\x12\n\nkey_prefix\x18\x03 \x01(\t\x12,\n\tretention\x18\x04 \x01(\x0b\x32\x19.google.protobuf.Duration\x12-\n\nsafety_ttl\x18\x05 \x01(\x0b\x32\x19.google.protobuf.Duration\x12\x1f\n\x17\x62locking_reads_per_node\x18\x08 \x01(\x05\x12\x33\n\x10response_timeout\x18\t \x01(\x0b\x32\x19.google.protobuf.Duration\x12\x0f\n\x07\x62\x61\x63kend\x18\n \x01(\t\x12\x15\n\rwait_replicas\x18\x0b \x01(\x05\x12/\n\x0cwait_timeout\x18\x0c \x01(\x0b\x32\x19.google.protobuf.Durationb\x06proto3' +) + + +_REDISSTREAMCONFIG = DESCRIPTOR.message_types_by_name["RedisStreamConfig"] +RedisStreamConfig = _reflection.GeneratedProtocolMessageType( + "RedisStreamConfig", + (_message.Message,), + { + "DESCRIPTOR": _REDISSTREAMCONFIG, + "__module__": "temporalio.streams.proto.v1.redis_pb2", + # @@protoc_insertion_point(class_scope:temporal.sdk.streams.v1.RedisStreamConfig) + }, +) +_sym_db.RegisterMessage(RedisStreamConfig) + +if _descriptor._USE_C_DESCRIPTORS == False: + DESCRIPTOR._options = None + _REDISSTREAMCONFIG._serialized_start = 97 + _REDISSTREAMCONFIG._serialized_end = 435 +# @@protoc_insertion_point(module_scope) diff --git a/temporalio/streams/proto/v1/redis_pb2.pyi b/temporalio/streams/proto/v1/redis_pb2.pyi new file mode 100644 index 0000000000..28e0dd956e --- /dev/null +++ b/temporalio/streams/proto/v1/redis_pb2.pyi @@ -0,0 +1,122 @@ +""" +@generated by mypy-protobuf. Do not edit manually! +isort:skip_file +The config of the Redis Stream runtime. It's the Redis Store's own config, so it isn't in +sdk-core's `coresdk.streams` package. Its shared fields have the numbers and types of +`coresdk.streams.MemoryStreamConfig`, so a backend can pass either config. Every SDK uses these +field numbers. +""" + +import builtins +import collections.abc +import sys + +import google.protobuf.descriptor +import google.protobuf.duration_pb2 +import google.protobuf.internal.containers +import google.protobuf.message + +if sys.version_info >= (3, 8): + import typing as typing_extensions +else: + import typing_extensions + +DESCRIPTOR: google.protobuf.descriptor.FileDescriptor + +class RedisStreamConfig(google.protobuf.message.Message): + """The settings of the Redis Stream runtime. Lang builds it from the backend options.""" + + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + URLS_FIELD_NUMBER: builtins.int + CLUSTER_FIELD_NUMBER: builtins.int + KEY_PREFIX_FIELD_NUMBER: builtins.int + RETENTION_FIELD_NUMBER: builtins.int + SAFETY_TTL_FIELD_NUMBER: builtins.int + BLOCKING_READS_PER_NODE_FIELD_NUMBER: builtins.int + RESPONSE_TIMEOUT_FIELD_NUMBER: builtins.int + BACKEND_FIELD_NUMBER: builtins.int + WAIT_REPLICAS_FIELD_NUMBER: builtins.int + WAIT_TIMEOUT_FIELD_NUMBER: builtins.int + @property + def urls( + self, + ) -> google.protobuf.internal.containers.RepeatedScalarFieldContainer[builtins.str]: + """One URL, or the seed URLs of a cluster. Credentials and TLS come from the URL.""" + cluster: builtins.bool + key_prefix: builtins.str + """Only letters, digits, `-`, `.`, `_` and `~`. Empty is `temporal-streams`.""" + @property + def retention(self) -> google.protobuf.duration_pb2.Duration: + """The default Retention of a new Stream. Empty is 7 days.""" + @property + def safety_ttl(self) -> google.protobuf.duration_pb2.Duration: + """How long an open Workflow-owned Stream stays after its last write. Empty is 90 days.""" + blocking_reads_per_node: builtins.int + """Zero is 256.""" + @property + def response_timeout(self) -> google.protobuf.duration_pb2.Duration: + """Empty is 5 s.""" + backend: builtins.str + """The backend name that lang registered this runtime under. The crate sets it on each + StreamRef that it builds, such as the one in a create response. + """ + wait_replicas: builtins.int + """How many replicas must have each write before it answers. Zero sends no WAIT.""" + @property + def wait_timeout(self) -> google.protobuf.duration_pb2.Duration: + """How long a write waits for its replicas, at most half of response_timeout. Empty is 1 s.""" + def __init__( + self, + *, + urls: collections.abc.Iterable[builtins.str] | None = ..., + cluster: builtins.bool = ..., + key_prefix: builtins.str = ..., + retention: google.protobuf.duration_pb2.Duration | None = ..., + safety_ttl: google.protobuf.duration_pb2.Duration | None = ..., + blocking_reads_per_node: builtins.int = ..., + response_timeout: google.protobuf.duration_pb2.Duration | None = ..., + backend: builtins.str = ..., + wait_replicas: builtins.int = ..., + wait_timeout: google.protobuf.duration_pb2.Duration | None = ..., + ) -> None: ... + def HasField( + self, + field_name: typing_extensions.Literal[ + "response_timeout", + b"response_timeout", + "retention", + b"retention", + "safety_ttl", + b"safety_ttl", + "wait_timeout", + b"wait_timeout", + ], + ) -> builtins.bool: ... + def ClearField( + self, + field_name: typing_extensions.Literal[ + "backend", + b"backend", + "blocking_reads_per_node", + b"blocking_reads_per_node", + "cluster", + b"cluster", + "key_prefix", + b"key_prefix", + "response_timeout", + b"response_timeout", + "retention", + b"retention", + "safety_ttl", + b"safety_ttl", + "urls", + b"urls", + "wait_replicas", + b"wait_replicas", + "wait_timeout", + b"wait_timeout", + ], + ) -> None: ... + +global___RedisStreamConfig = RedisStreamConfig