diff --git a/README.md b/README.md index bf6a0168..9f5498e7 100644 --- a/README.md +++ b/README.md @@ -69,6 +69,11 @@ execution = client.durable.start( ) print(execution.id, execution.status) +execution = client.durable.get_from_application( + "order-pipeline", execution.id, client_token=execution.client_token +) +print(execution.is_terminal, execution.result) + owner_client = VolcanoClient( api_url="https://api.volcano.dev", anon_key="your-anon-key", @@ -245,26 +250,36 @@ Anonymous and service keys do not refresh. `durable.start()` begins an execution of a deployed durable function and returns a handle rather than a result: an execution can run for up to 366 days, so its -result is read back with `durable.get()`. It takes the credential `functions.invoke()` -takes, and is, with `durable.messages`, the only durable operation an -application credential may perform. +result is read back later. It takes the credential `functions.invoke()` takes. An `execution_name` makes the start idempotent — starting again under the same name returns the execution that already exists rather than beginning a second -one, and is charged once. - -`durable.get()`, `durable.list()` and `durable.stop()` are owner-scoped and need -the project owner's platform user token or a project access token, because an -execution is addressed by its id alone and an anonymous key is held by everyone -who loads the page. A `read_only` project access token can read and list but not -stop. Auth-user sessions from `sign_in()`, anonymous keys, and service keys are -not accepted. Poll them from a trusted backend. -`get()` carries `result` once the execution has succeeded and `error` when it +one, and is charged once. Only the start that creates the execution returns its +`client_token`. + +`durable.get_from_application()` and `durable.stop_from_application()` read and +stop an execution with the credential `durable.start()` takes. A service key +holding `*`, the default, reaches any execution of the project and a signed-in +user the executions they started; an anonymous key, a service key scoped without +`*`, or a user who did not start the execution, passes the `client_token`. An +anonymous key or a service key scoped without `*` also needs `functions.invoke`; +without it the call raises `PermissionDeniedError`. The function's visibility +must also admit the caller: an anonymous key on an `authenticated` function +raises `PermissionDeniedError`, even with the token, and a `private` function +raises `NotFoundError` for every credential but a service key. An execution the +caller cannot reach raises `NotFoundError`, as a missing one does. + +`durable.get()`, `durable.list()` and `durable.stop()` are owner-scoped, reach any +execution of the project, and need the project owner's platform user token or a +project access token. A `read_only` project access token can read and list but +not stop. Auth-user sessions from `sign_in()`, anonymous keys, and service keys +are not accepted. Call them from a trusted backend. +Every read carries `result` once the execution has succeeded and `error` when it failed; `result_expired` separates a result the platform has discarded from a function that returned nothing. `is_terminal` reports whether the execution has stopped changing, and counts `unknown` — the status the platform writes for an -outcome it could not determine — as finished. `stop()` is accepted rather than +outcome it could not determine — as finished. Either stop is accepted rather than awaited: what it returns is the execution read back after asking, often still -`running`, so poll `get()` to see it reach `stopped`. Repeating a stop is safe. +`running`, so poll a read to see it reach `stopped`. Repeating a stop is safe. `durable.approvals` reads and decides the approvals a durable function waits on: diff --git a/docs/functions.md b/docs/functions.md index 787dc1e0..becad2a9 100644 --- a/docs/functions.md +++ b/docs/functions.md @@ -74,6 +74,12 @@ handle = client.durable.start( {"order_id": "order-9"}, execution_name="order-9", ) +execution = client.durable.get_from_application( + "charge-order", handle.id, client_token=handle.client_token +) +client.durable.stop_from_application( + "charge-order", handle.id, client_token=handle.client_token +) owner_client = VolcanoClient( anon_key=os.environ["VOLCANO_ANON_KEY"], @@ -85,9 +91,11 @@ page = owner_client.durable.list(project_id, "charge-order", status="running") owner_client.durable.stop(project_id, "charge-order", handle.id) ``` -`start()` accepts the same active session, service key, or anonymous key as `functions.invoke()`. Besides [sending and following messages](#send-and-follow-messages), it is the only durable operation available to application credentials. An execution name makes a start idempotent. +`start()` accepts the same active session, service key, or anonymous key as `functions.invoke()`. An execution name makes a start idempotent. Only the start that creates the execution returns its `client_token`; a repeat under the same name and a read never do, so keep it with the execution id. + +`get_from_application()` and `stop_from_application()` read and stop an execution with the credential `start()` takes, as [sending and following messages](#send-and-follow-messages) does. A service key holding `*`, the default, reaches any execution of the project, and a signed-in user reaches the executions they started. An anonymous key, a service key scoped without `*`, or a user who did not start the execution, passes the `client_token`. An anonymous key or a service key scoped without `*` also needs `functions.invoke`; without it the call raises `PermissionDeniedError`. The function's visibility must also admit the caller: an anonymous key on an `authenticated` function raises `PermissionDeniedError`, even with the token, and a `private` function raises `NotFoundError` for every credential but a service key. An execution the caller cannot reach raises `NotFoundError`, as a missing one does, and until an execution leaves `pending` only a service key holding `*` reaches it; that key stopping one then raises `ConflictError`. Reads and stops share the function's rate limit with starts, so poll every few seconds. A `ServerError` from a stop can follow a stop that was already sent, so read the execution before stopping it again. -`get()`, `list()`, and `stop()` are owner-scoped. Call them from a trusted backend with the project owner's platform user token or a project access token; a `read_only` token can read and list but not stop. Auth-user sessions, anonymous keys, and service keys are not accepted. `stop()` returns after the stop request is accepted, so poll `get()` until `is_terminal` is true. +`get()`, `list()`, and `stop()` are owner-scoped and reach any execution of the project. Call them from a trusted backend with the project owner's platform user token or a project access token; a `read_only` token can read and list but not stop. Auth-user sessions, anonymous keys, and service keys are not accepted. Listing has no application equivalent. Either stop returns once the request is accepted, so poll a read until `is_terminal` is true. ## Write a durable function diff --git a/openapi/openapi.yaml b/openapi/openapi.yaml index 641fe9aa..795f3330 100644 --- a/openapi/openapi.yaml +++ b/openapi/openapi.yaml @@ -4383,10 +4383,13 @@ paths: A `private` durable function answers every credential but a service key exactly as a missing one, with 404. - Starting is all this endpoint does. Reading a result or stopping an - execution requires the project owner's token, because an anon key is - shared by everyone who loads the page and an execution is addressed by - id alone. + The start returns a `client_token`. An anon key, or a user who did not + start the execution, sends it as `X-Volcano-Client-Token` to read the + execution, stop it, and reach its messages. A service key scoped to a + list without `*` sends it to read and stop the execution. An anon key is shared by + everyone who loads the page, so the id alone never reaches an + execution. Listing executions takes the project owner's token or a + project access token. Send `X-Volcano-Execution-Name` to make the start idempotent: repeating a start with the same name returns the existing execution instead of @@ -4505,6 +4508,200 @@ paths: application/json: schema: $ref: '#/components/schemas/Error' + /durable-functions/{functionId}/executions/{executionId}: + get: + tags: + - Durable Functions + summary: Get a durable execution from an application + description: | + Returns the execution's current state with the credential that started + it, including its `result` once it has succeeded. Poll this to wait for + an execution to finish. + + **With a service key:** any execution in the project when the key holds + `*`, the default. A key scoped to a list without `*` reaches an + execution whose `X-Volcano-Client-Token` the request carries, as an + anon key does; the start returns that token. + + **With an auth user token:** an execution the user started, or one + whose `X-Volcano-Client-Token` the request carries, of a durable + function whose `visibility` is `authenticated` or `public`. + + **With an anon key:** an execution whose `X-Volcano-Client-Token` the + request carries, of a function with `visibility: public`. + + Only a service key holding `*` reaches an execution that is still + `pending`. Any other execution answers `404`, as a missing one does. + + The response is the one `GET + /projects/{id}/durable-functions/{functionId}/executions/{executionId}` + returns. It never carries `client_token`. + operationId: getDurableExecutionFromApplication + security: + - AnonKey: [] + - ServiceRoleKey: [] + - AuthUserAccessToken: [] + parameters: + - $ref: '#/components/parameters/DurableFunctionId' + - $ref: '#/components/parameters/DurableExecutionId' + - $ref: '#/components/parameters/DurableClientToken' + responses: + '200': + description: Successful response + content: + application/json: + schema: + $ref: '#/components/schemas/DurableExecution' + '401': + description: Missing or invalid credential + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + '403': + description: | + The anon key lacks `functions.invoke`, the service key is scoped + without it, or the function is `authenticated` and the request + carries an anon key. Also returned when the request's origin is + refused by the project's CORS policy. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + '404': + description: | + Durable function or execution not found, or the credential cannot + reach the execution. A `private` function answers every credential + but a service key this way. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + '429': + description: | + The function's or the project's function rate limit, which reads + share with invocations and starts, or the project is over its + bandwidth cap. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + '503': + description: | + Durable execution is not available in this environment, the plan + terms for the request could not be read, or reads of the + execution's state in its region are throttled. The first is not one + to retry in a loop; the others are transient, and a throttled read + carries `Retry-After`. + headers: + Retry-After: + description: | + Seconds to wait before reading again. Sent only when reads in + the execution's region are throttled. + schema: + type: integer + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + /durable-functions/{functionId}/executions/{executionId}/stop: + post: + tags: + - Durable Functions + summary: Stop a durable execution from an application + description: | + Cancels a running execution with the credential that started it. Its + completed steps are not undone. + + Access follows the same rules as reading the execution. + + The call is accepted rather than awaited: the response reports the + execution as it was read back and may still say `running`. Poll + `GET /durable-functions/{functionId}/executions/{executionId}` to see + it settle into `stopped`. + + Stopping an execution that already finished is not an error: the + response carries the state it settled in. + operationId: stopDurableExecutionFromApplication + security: + - AnonKey: [] + - ServiceRoleKey: [] + - AuthUserAccessToken: [] + parameters: + - $ref: '#/components/parameters/DurableFunctionId' + - $ref: '#/components/parameters/DurableExecutionId' + - $ref: '#/components/parameters/DurableClientToken' + responses: + '200': + description: | + Stop accepted. The body is the execution as it was read back, which + may still report `running`. + content: + application/json: + schema: + $ref: '#/components/schemas/DurableExecution' + '401': + description: Missing or invalid credential + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + '403': + description: | + The anon key lacks `functions.invoke`, the service key is scoped + without it, or the function is `authenticated` and the request + carries an anon key. Also returned when the request's origin is + refused by the project's CORS policy. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + '404': + description: | + Durable function or execution not found, or the credential cannot + reach the execution. A `private` function answers every credential + but a service key this way. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + '409': + description: | + The execution has not started yet. Only a service key holding `*` + can reach an execution in that state. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + '429': + description: | + The function's or the project's function rate limit, which stops + share with invocations and starts, or the project is over its + bandwidth cap. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + '503': + description: | + Durable execution is not available in this environment, the plan + terms for the request could not be read, or the execution's state + could not be read back because reads in its region are throttled. + The first is not one to retry in a loop; the others are transient, + and a throttled read carries `Retry-After`. A stop answered with a + throttle may already have been sent, so read the execution before + stopping it again. + headers: + Retry-After: + description: | + Seconds to wait before reading again. Sent only when reads in + the execution's region are throttled. + schema: + type: integer + content: + application/json: + schema: + $ref: '#/components/schemas/Error' /durable-functions/{functionId}/executions/{executionId}/messages: post: tags: @@ -16926,7 +17123,8 @@ components: required: false description: | The `client_token` the execution's start returned. Lets an anon key, - or a user who did not start the execution, reach it. + or a user who did not start the execution, reach it, and a service key + scoped to a list without `*` read and stop it. schema: type: string maxLength: 128 @@ -21591,7 +21789,8 @@ components: Whatever the function returned, verbatim. Absent while the execution is still running, absent when the result was too large to return and was checkpointed instead, and absent once the retention period has - lapsed. + lapsed. A result Volcano has just read can still be returned for up + to five minutes after that. result_expired: type: boolean description: | @@ -21614,9 +21813,11 @@ components: client_token: type: string description: | - Lets a browser read and send the execution's messages. Returned only - by the start whose input the execution runs, never by a repeat under - the same name or by a read. Send it as `X-Volcano-Client-Token`. + Lets a browser read the execution, stop it, and read and send its + messages, and a service key scoped to a list without `*` read and + stop it. Returned only by the start whose input the execution runs, + never by a repeat under the same name or by a read. Send it as + `X-Volcano-Client-Token`. required: - id - function_id diff --git a/scripts/generate_openapi.py b/scripts/generate_openapi.py index 13494207..7c01df74 100755 --- a/scripts/generate_openapi.py +++ b/scripts/generate_openapi.py @@ -26,6 +26,7 @@ "get_durable_approval.py", "get_durable_approval_stats.py", "get_durable_execution.py", + "get_durable_execution_from_application.py", "force_release_project_lock.py", "get_project_log_activity.py", "get_project_lock.py", @@ -45,6 +46,7 @@ "send_durable_message_from_application.py", "start_durable_execution_from_application.py", "stop_durable_execution.py", + "stop_durable_execution_from_application.py", "update_storage_object_visibility.py", "upload_storage_object.py", } diff --git a/src/volcano_sdk/_durable_arguments.py b/src/volcano_sdk/_durable_arguments.py index 869c4e90..27140ba0 100644 --- a/src/volcano_sdk/_durable_arguments.py +++ b/src/volcano_sdk/_durable_arguments.py @@ -24,6 +24,8 @@ "execution_id": "execution_id must be a UUID", "approval_id": "approval_id must be a UUID", } +# The spec's maxLength on X-Volcano-Client-Token. +MAX_CLIENT_TOKEN_LENGTH = 128 def identifier(value: object, field: str) -> str: @@ -67,3 +69,41 @@ def known_options(options: Mapping[str, object], known: frozenset[str]) -> None: if unknown: message = f"unexpected option: {', '.join(unknown)}" raise TypeError(message) + + +def header_value(value: object, field: str, limit: int) -> str | None: + """Require a value an optional request header can carry. + + Returns + ------- + str | None + The value, or None when it is unset. + + Raises + ------ + ValueError + If the value is empty, is not a string, is not printable ASCII, has + surrounding whitespace, or exceeds `limit` characters. + + """ + if value is None: + return None + if not isinstance(value, str) or not value: + message = f"{field} must be a non-empty string" + raise ValueError(message) + _require_header_text(value, field) + if len(value) > limit: + message = f"{field} must be at most {limit} characters" + raise ValueError(message) + return value + + +def _require_header_text(value: str, field: str) -> None: + # Headers carry printable ASCII only, and the HTTP stack refuses a value + # with surrounding whitespace. + if not (value.isascii() and value.isprintable()): + message = f"{field} must be printable ASCII" + raise ValueError(message) + if value != value.strip(): + message = f"{field} must not start or end with whitespace" + raise ValueError(message) diff --git a/src/volcano_sdk/_durable_messages.py b/src/volcano_sdk/_durable_messages.py index db05b9cc..547a7d23 100644 --- a/src/volcano_sdk/_durable_messages.py +++ b/src/volcano_sdk/_durable_messages.py @@ -7,7 +7,12 @@ from typing import TYPE_CHECKING, Protocol, Unpack, runtime_checkable from ._callbacks import require_callable -from ._durable_arguments import identifier, known_options +from ._durable_arguments import ( + MAX_CLIENT_TOKEN_LENGTH, + header_value, + identifier, + known_options, +) from ._durable_callback import operation_name from ._durable_message_response import durable_message, durable_message_page from ._durable_message_stream import DurableMessageStream @@ -45,9 +50,8 @@ "The server ended the durable message subscription; subscribe again to resume" ) _LEAVE_FAILED = "Volcano could not leave the ended durable message channel" -# The spec's bounds on Idempotency-Key, X-Volcano-Client-Token, and a page. +# The spec's bounds on Idempotency-Key and a page. _MAX_IDEMPOTENCY_KEY_LENGTH = 255 -_MAX_CLIENT_TOKEN_LENGTH = 128 _MAX_LIST_LIMIT = 500 _TARGET_OPTIONS = frozenset({"client_token", "project_id"}) _SEND_OPTIONS = _TARGET_OPTIONS | {"idempotency_key"} @@ -230,7 +234,7 @@ def send( request = DurableMessageSendRequest( name=operation_name(name), data=data, - idempotency_key=_bounded_text( + idempotency_key=header_value( options.get("idempotency_key"), "idempotency_key", _MAX_IDEMPOTENCY_KEY_LENGTH, @@ -461,36 +465,12 @@ def _target( function_id=identifier(function_name, "function_name"), execution_id=identifier(execution_id, "execution_id"), project_id=None if project_id is None else identifier(project_id, "project_id"), - client_token=_bounded_text( - options.get("client_token"), "client_token", _MAX_CLIENT_TOKEN_LENGTH + client_token=header_value( + options.get("client_token"), "client_token", MAX_CLIENT_TOKEN_LENGTH ), ) -def _bounded_text(value: object, field: str, limit: int) -> str | None: - if value is None: - return None - if not isinstance(value, str) or not value: - message = f"{field} must be a non-empty string" - raise ValueError(message) - _require_header_value(value, field) - if len(value) > limit: - message = f"{field} must be at most {limit} characters" - raise ValueError(message) - return value - - -def _require_header_value(value: str, field: str) -> None: - # Headers carry printable ASCII only, and the HTTP stack refuses a value - # with surrounding whitespace. - if not (value.isascii() and value.isprintable()): - message = f"{field} must be printable ASCII" - raise ValueError(message) - if value != value.strip(): - message = f"{field} must not start or end with whitespace" - raise ValueError(message) - - def _after(value: object) -> int | None: if value is None: return None diff --git a/src/volcano_sdk/_generated/api/durable_functions/get_durable_execution_from_application.py b/src/volcano_sdk/_generated/api/durable_functions/get_durable_execution_from_application.py new file mode 100644 index 00000000..f4f26717 --- /dev/null +++ b/src/volcano_sdk/_generated/api/durable_functions/get_durable_execution_from_application.py @@ -0,0 +1,333 @@ +from http import HTTPStatus +from typing import Any, cast +from urllib.parse import quote + +import httpx + +from ...client import AuthenticatedClient, Client +from ...types import Response, UNSET +from ... import errors + +from ...models.durable_execution import DurableExecution +from ...models.error import Error +from ...types import UNSET, Unset +from typing import cast +from uuid import UUID + + + +def request_kwargs( + function_id: str, + execution_id: UUID | str, + *, + x_volcano_client_token: str | Unset = UNSET, + +) -> dict[str, Any]: + headers: dict[str, Any] = {} + if not isinstance(x_volcano_client_token, Unset): + headers["X-Volcano-Client-Token"] = x_volcano_client_token + + + + + + + + _kwargs: dict[str, Any] = { + "method": "get", + "url": "/durable-functions/{function_id}/executions/{execution_id}".format(function_id=quote(str(function_id), safe=""),execution_id=quote(str(execution_id), safe=""),), + } + + + _kwargs["headers"] = headers + return _kwargs + + + +def _parse_response(*, client: AuthenticatedClient | Client, response: httpx.Response) -> DurableExecution | Error | None: + if response.status_code == 200: + response_200 = DurableExecution.from_dict(response.json()) + + + + return response_200 + + if response.status_code == 401: + response_401 = Error.from_dict(response.json()) + + + + return response_401 + + if response.status_code == 403: + response_403 = Error.from_dict(response.json()) + + + + return response_403 + + if response.status_code == 404: + response_404 = Error.from_dict(response.json()) + + + + return response_404 + + if response.status_code == 429: + response_429 = Error.from_dict(response.json()) + + + + return response_429 + + if response.status_code == 503: + response_503 = Error.from_dict(response.json()) + + + + return response_503 + + if client.raise_on_unexpected_status: + raise errors.UnexpectedStatus(response.status_code, response.content) + else: + return None + + +def build_response(*, client: AuthenticatedClient | Client, response: httpx.Response) -> Response[DurableExecution | Error]: + return Response( + status_code=HTTPStatus(response.status_code), + content=response.content, + headers=response.headers, + parsed=_parse_response(client=client, response=response), + ) + + +def sync_detailed( + function_id: str, + execution_id: UUID | str, + *, + client: AuthenticatedClient, + x_volcano_client_token: str | Unset = UNSET, + +) -> Response[DurableExecution | Error]: + """ Get a durable execution from an application + + Returns the execution's current state with the credential that started + it, including its `result` once it has succeeded. Poll this to wait for + an execution to finish. + + **With a service key:** any execution in the project when the key holds + `*`, the default. A key scoped to a list without `*` reaches an + execution whose `X-Volcano-Client-Token` the request carries, as an + anon key does; the start returns that token. + + **With an auth user token:** an execution the user started, or one + whose `X-Volcano-Client-Token` the request carries, of a durable + function whose `visibility` is `authenticated` or `public`. + + **With an anon key:** an execution whose `X-Volcano-Client-Token` the + request carries, of a function with `visibility: public`. + + Only a service key holding `*` reaches an execution that is still + `pending`. Any other execution answers `404`, as a missing one does. + + The response is the one `GET + /projects/{id}/durable-functions/{functionId}/executions/{executionId}` + returns. It never carries `client_token`. + + Args: + function_id (str): + execution_id (UUID): + x_volcano_client_token (str | Unset): + + Raises: + errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True. + httpx.TimeoutException: If the request takes longer than Client.timeout. + + Returns: + Response[DurableExecution | Error] + """ + + + kwargs = request_kwargs( + function_id=function_id, +execution_id=execution_id, +x_volcano_client_token=x_volcano_client_token, + + ) + + response = client.get_httpx_client().request( + **kwargs, + ) + + return build_response(client=client, response=response) + +def sync( + function_id: str, + execution_id: UUID | str, + *, + client: AuthenticatedClient, + x_volcano_client_token: str | Unset = UNSET, + +) -> DurableExecution | Error | None: + """ Get a durable execution from an application + + Returns the execution's current state with the credential that started + it, including its `result` once it has succeeded. Poll this to wait for + an execution to finish. + + **With a service key:** any execution in the project when the key holds + `*`, the default. A key scoped to a list without `*` reaches an + execution whose `X-Volcano-Client-Token` the request carries, as an + anon key does; the start returns that token. + + **With an auth user token:** an execution the user started, or one + whose `X-Volcano-Client-Token` the request carries, of a durable + function whose `visibility` is `authenticated` or `public`. + + **With an anon key:** an execution whose `X-Volcano-Client-Token` the + request carries, of a function with `visibility: public`. + + Only a service key holding `*` reaches an execution that is still + `pending`. Any other execution answers `404`, as a missing one does. + + The response is the one `GET + /projects/{id}/durable-functions/{functionId}/executions/{executionId}` + returns. It never carries `client_token`. + + Args: + function_id (str): + execution_id (UUID): + x_volcano_client_token (str | Unset): + + Raises: + errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True. + httpx.TimeoutException: If the request takes longer than Client.timeout. + + Returns: + DurableExecution | Error + """ + + + return sync_detailed( + function_id=function_id, +execution_id=execution_id, +client=client, +x_volcano_client_token=x_volcano_client_token, + + ).parsed + +async def asyncio_detailed( + function_id: str, + execution_id: UUID | str, + *, + client: AuthenticatedClient, + x_volcano_client_token: str | Unset = UNSET, + +) -> Response[DurableExecution | Error]: + """ Get a durable execution from an application + + Returns the execution's current state with the credential that started + it, including its `result` once it has succeeded. Poll this to wait for + an execution to finish. + + **With a service key:** any execution in the project when the key holds + `*`, the default. A key scoped to a list without `*` reaches an + execution whose `X-Volcano-Client-Token` the request carries, as an + anon key does; the start returns that token. + + **With an auth user token:** an execution the user started, or one + whose `X-Volcano-Client-Token` the request carries, of a durable + function whose `visibility` is `authenticated` or `public`. + + **With an anon key:** an execution whose `X-Volcano-Client-Token` the + request carries, of a function with `visibility: public`. + + Only a service key holding `*` reaches an execution that is still + `pending`. Any other execution answers `404`, as a missing one does. + + The response is the one `GET + /projects/{id}/durable-functions/{functionId}/executions/{executionId}` + returns. It never carries `client_token`. + + Args: + function_id (str): + execution_id (UUID): + x_volcano_client_token (str | Unset): + + Raises: + errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True. + httpx.TimeoutException: If the request takes longer than Client.timeout. + + Returns: + Response[DurableExecution | Error] + """ + + + kwargs = request_kwargs( + function_id=function_id, +execution_id=execution_id, +x_volcano_client_token=x_volcano_client_token, + + ) + + response = await client.get_async_httpx_client().request( + **kwargs + ) + + return build_response(client=client, response=response) + +async def asyncio( + function_id: str, + execution_id: UUID | str, + *, + client: AuthenticatedClient, + x_volcano_client_token: str | Unset = UNSET, + +) -> DurableExecution | Error | None: + """ Get a durable execution from an application + + Returns the execution's current state with the credential that started + it, including its `result` once it has succeeded. Poll this to wait for + an execution to finish. + + **With a service key:** any execution in the project when the key holds + `*`, the default. A key scoped to a list without `*` reaches an + execution whose `X-Volcano-Client-Token` the request carries, as an + anon key does; the start returns that token. + + **With an auth user token:** an execution the user started, or one + whose `X-Volcano-Client-Token` the request carries, of a durable + function whose `visibility` is `authenticated` or `public`. + + **With an anon key:** an execution whose `X-Volcano-Client-Token` the + request carries, of a function with `visibility: public`. + + Only a service key holding `*` reaches an execution that is still + `pending`. Any other execution answers `404`, as a missing one does. + + The response is the one `GET + /projects/{id}/durable-functions/{functionId}/executions/{executionId}` + returns. It never carries `client_token`. + + Args: + function_id (str): + execution_id (UUID): + x_volcano_client_token (str | Unset): + + Raises: + errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True. + httpx.TimeoutException: If the request takes longer than Client.timeout. + + Returns: + DurableExecution | Error + """ + + + return (await asyncio_detailed( + function_id=function_id, +execution_id=execution_id, +client=client, +x_volcano_client_token=x_volcano_client_token, + + )).parsed diff --git a/src/volcano_sdk/_generated/api/durable_functions/start_durable_execution_from_application.py b/src/volcano_sdk/_generated/api/durable_functions/start_durable_execution_from_application.py index 282b2377..c300cb59 100644 --- a/src/volcano_sdk/_generated/api/durable_functions/start_durable_execution_from_application.py +++ b/src/volcano_sdk/_generated/api/durable_functions/start_durable_execution_from_application.py @@ -157,10 +157,13 @@ def sync_detailed( A `private` durable function answers every credential but a service key exactly as a missing one, with 404. - Starting is all this endpoint does. Reading a result or stopping an - execution requires the project owner's token, because an anon key is - shared by everyone who loads the page and an execution is addressed by - id alone. + The start returns a `client_token`. An anon key, or a user who did not + start the execution, sends it as `X-Volcano-Client-Token` to read the + execution, stop it, and reach its messages. A service key scoped to a + list without `*` sends it to read and stop the execution. An anon key is shared by + everyone who loads the page, so the id alone never reaches an + execution. Listing executions takes the project owner's token or a + project access token. Send `X-Volcano-Execution-Name` to make the start idempotent: repeating a start with the same name returns the existing execution instead of @@ -230,10 +233,13 @@ def sync( A `private` durable function answers every credential but a service key exactly as a missing one, with 404. - Starting is all this endpoint does. Reading a result or stopping an - execution requires the project owner's token, because an anon key is - shared by everyone who loads the page and an execution is addressed by - id alone. + The start returns a `client_token`. An anon key, or a user who did not + start the execution, sends it as `X-Volcano-Client-Token` to read the + execution, stop it, and reach its messages. A service key scoped to a + list without `*` sends it to read and stop the execution. An anon key is shared by + everyone who loads the page, so the id alone never reaches an + execution. Listing executions takes the project owner's token or a + project access token. Send `X-Volcano-Execution-Name` to make the start idempotent: repeating a start with the same name returns the existing execution instead of @@ -298,10 +304,13 @@ async def asyncio_detailed( A `private` durable function answers every credential but a service key exactly as a missing one, with 404. - Starting is all this endpoint does. Reading a result or stopping an - execution requires the project owner's token, because an anon key is - shared by everyone who loads the page and an execution is addressed by - id alone. + The start returns a `client_token`. An anon key, or a user who did not + start the execution, sends it as `X-Volcano-Client-Token` to read the + execution, stop it, and reach its messages. A service key scoped to a + list without `*` sends it to read and stop the execution. An anon key is shared by + everyone who loads the page, so the id alone never reaches an + execution. Listing executions takes the project owner's token or a + project access token. Send `X-Volcano-Execution-Name` to make the start idempotent: repeating a start with the same name returns the existing execution instead of @@ -371,10 +380,13 @@ async def asyncio( A `private` durable function answers every credential but a service key exactly as a missing one, with 404. - Starting is all this endpoint does. Reading a result or stopping an - execution requires the project owner's token, because an anon key is - shared by everyone who loads the page and an execution is addressed by - id alone. + The start returns a `client_token`. An anon key, or a user who did not + start the execution, sends it as `X-Volcano-Client-Token` to read the + execution, stop it, and reach its messages. A service key scoped to a + list without `*` sends it to read and stop the execution. An anon key is shared by + everyone who loads the page, so the id alone never reaches an + execution. Listing executions takes the project owner's token or a + project access token. Send `X-Volcano-Execution-Name` to make the start idempotent: repeating a start with the same name returns the existing execution instead of diff --git a/src/volcano_sdk/_generated/api/durable_functions/stop_durable_execution_from_application.py b/src/volcano_sdk/_generated/api/durable_functions/stop_durable_execution_from_application.py new file mode 100644 index 00000000..06387e68 --- /dev/null +++ b/src/volcano_sdk/_generated/api/durable_functions/stop_durable_execution_from_application.py @@ -0,0 +1,300 @@ +from http import HTTPStatus +from typing import Any, cast +from urllib.parse import quote + +import httpx + +from ...client import AuthenticatedClient, Client +from ...types import Response, UNSET +from ... import errors + +from ...models.durable_execution import DurableExecution +from ...models.error import Error +from ...types import UNSET, Unset +from typing import cast +from uuid import UUID + + + +def request_kwargs( + function_id: str, + execution_id: UUID | str, + *, + x_volcano_client_token: str | Unset = UNSET, + +) -> dict[str, Any]: + headers: dict[str, Any] = {} + if not isinstance(x_volcano_client_token, Unset): + headers["X-Volcano-Client-Token"] = x_volcano_client_token + + + + + + + + _kwargs: dict[str, Any] = { + "method": "post", + "url": "/durable-functions/{function_id}/executions/{execution_id}/stop".format(function_id=quote(str(function_id), safe=""),execution_id=quote(str(execution_id), safe=""),), + } + + + _kwargs["headers"] = headers + return _kwargs + + + +def _parse_response(*, client: AuthenticatedClient | Client, response: httpx.Response) -> DurableExecution | Error | None: + if response.status_code == 200: + response_200 = DurableExecution.from_dict(response.json()) + + + + return response_200 + + if response.status_code == 401: + response_401 = Error.from_dict(response.json()) + + + + return response_401 + + if response.status_code == 403: + response_403 = Error.from_dict(response.json()) + + + + return response_403 + + if response.status_code == 404: + response_404 = Error.from_dict(response.json()) + + + + return response_404 + + if response.status_code == 409: + response_409 = Error.from_dict(response.json()) + + + + return response_409 + + if response.status_code == 429: + response_429 = Error.from_dict(response.json()) + + + + return response_429 + + if response.status_code == 503: + response_503 = Error.from_dict(response.json()) + + + + return response_503 + + if client.raise_on_unexpected_status: + raise errors.UnexpectedStatus(response.status_code, response.content) + else: + return None + + +def build_response(*, client: AuthenticatedClient | Client, response: httpx.Response) -> Response[DurableExecution | Error]: + return Response( + status_code=HTTPStatus(response.status_code), + content=response.content, + headers=response.headers, + parsed=_parse_response(client=client, response=response), + ) + + +def sync_detailed( + function_id: str, + execution_id: UUID | str, + *, + client: AuthenticatedClient, + x_volcano_client_token: str | Unset = UNSET, + +) -> Response[DurableExecution | Error]: + """ Stop a durable execution from an application + + Cancels a running execution with the credential that started it. Its + completed steps are not undone. + + Access follows the same rules as reading the execution. + + The call is accepted rather than awaited: the response reports the + execution as it was read back and may still say `running`. Poll + `GET /durable-functions/{functionId}/executions/{executionId}` to see + it settle into `stopped`. + + Stopping an execution that already finished is not an error: the + response carries the state it settled in. + + Args: + function_id (str): + execution_id (UUID): + x_volcano_client_token (str | Unset): + + Raises: + errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True. + httpx.TimeoutException: If the request takes longer than Client.timeout. + + Returns: + Response[DurableExecution | Error] + """ + + + kwargs = request_kwargs( + function_id=function_id, +execution_id=execution_id, +x_volcano_client_token=x_volcano_client_token, + + ) + + response = client.get_httpx_client().request( + **kwargs, + ) + + return build_response(client=client, response=response) + +def sync( + function_id: str, + execution_id: UUID | str, + *, + client: AuthenticatedClient, + x_volcano_client_token: str | Unset = UNSET, + +) -> DurableExecution | Error | None: + """ Stop a durable execution from an application + + Cancels a running execution with the credential that started it. Its + completed steps are not undone. + + Access follows the same rules as reading the execution. + + The call is accepted rather than awaited: the response reports the + execution as it was read back and may still say `running`. Poll + `GET /durable-functions/{functionId}/executions/{executionId}` to see + it settle into `stopped`. + + Stopping an execution that already finished is not an error: the + response carries the state it settled in. + + Args: + function_id (str): + execution_id (UUID): + x_volcano_client_token (str | Unset): + + Raises: + errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True. + httpx.TimeoutException: If the request takes longer than Client.timeout. + + Returns: + DurableExecution | Error + """ + + + return sync_detailed( + function_id=function_id, +execution_id=execution_id, +client=client, +x_volcano_client_token=x_volcano_client_token, + + ).parsed + +async def asyncio_detailed( + function_id: str, + execution_id: UUID | str, + *, + client: AuthenticatedClient, + x_volcano_client_token: str | Unset = UNSET, + +) -> Response[DurableExecution | Error]: + """ Stop a durable execution from an application + + Cancels a running execution with the credential that started it. Its + completed steps are not undone. + + Access follows the same rules as reading the execution. + + The call is accepted rather than awaited: the response reports the + execution as it was read back and may still say `running`. Poll + `GET /durable-functions/{functionId}/executions/{executionId}` to see + it settle into `stopped`. + + Stopping an execution that already finished is not an error: the + response carries the state it settled in. + + Args: + function_id (str): + execution_id (UUID): + x_volcano_client_token (str | Unset): + + Raises: + errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True. + httpx.TimeoutException: If the request takes longer than Client.timeout. + + Returns: + Response[DurableExecution | Error] + """ + + + kwargs = request_kwargs( + function_id=function_id, +execution_id=execution_id, +x_volcano_client_token=x_volcano_client_token, + + ) + + response = await client.get_async_httpx_client().request( + **kwargs + ) + + return build_response(client=client, response=response) + +async def asyncio( + function_id: str, + execution_id: UUID | str, + *, + client: AuthenticatedClient, + x_volcano_client_token: str | Unset = UNSET, + +) -> DurableExecution | Error | None: + """ Stop a durable execution from an application + + Cancels a running execution with the credential that started it. Its + completed steps are not undone. + + Access follows the same rules as reading the execution. + + The call is accepted rather than awaited: the response reports the + execution as it was read back and may still say `running`. Poll + `GET /durable-functions/{functionId}/executions/{executionId}` to see + it settle into `stopped`. + + Stopping an execution that already finished is not an error: the + response carries the state it settled in. + + Args: + function_id (str): + execution_id (UUID): + x_volcano_client_token (str | Unset): + + Raises: + errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True. + httpx.TimeoutException: If the request takes longer than Client.timeout. + + Returns: + DurableExecution | Error + """ + + + return (await asyncio_detailed( + function_id=function_id, +execution_id=execution_id, +client=client, +x_volcano_client_token=x_volcano_client_token, + + )).parsed diff --git a/src/volcano_sdk/_generated/models/durable_execution.py b/src/volcano_sdk/_generated/models/durable_execution.py index c45737cf..291acd58 100644 --- a/src/volcano_sdk/_generated/models/durable_execution.py +++ b/src/volcano_sdk/_generated/models/durable_execution.py @@ -58,7 +58,8 @@ class DurableExecution: result (Any | Unset): Whatever the function returned, verbatim. Absent while the execution is still running, absent when the result was too large to return and was checkpointed instead, and absent once the retention period has - lapsed. + lapsed. A result Volcano has just read can still be returned for up + to five minutes after that. result_expired (bool | Unset): `true` when the execution is terminal but its result is no longer retained, which distinguishes a discarded result from an empty one. Shortly after that the execution itself is dropped and reads answer @@ -68,9 +69,11 @@ class DurableExecution: unset, so it reads like a function that returned nothing. error (DurableExecutionError | Unset): Why a failed or timed-out execution ended. completed_at (datetime.datetime | Unset): Present once the execution has reached a terminal status. - client_token (str | Unset): Lets a browser read and send the execution's messages. Returned only - by the start whose input the execution runs, never by a repeat under - the same name or by a read. Send it as `X-Volcano-Client-Token`. + client_token (str | Unset): Lets a browser read the execution, stop it, and read and send its + messages, and a service key scoped to a list without `*` read and + stop it. Returned only by the start whose input the execution runs, + never by a repeat under the same name or by a read. Send it as + `X-Volcano-Client-Token`. """ id: UUID diff --git a/src/volcano_sdk/_tests/test_durable.py b/src/volcano_sdk/_tests/test_durable.py index 755a3c32..76fc4f37 100644 --- a/src/volcano_sdk/_tests/test_durable.py +++ b/src/volcano_sdk/_tests/test_durable.py @@ -1,24 +1,35 @@ from __future__ import annotations +import re from dataclasses import dataclass from datetime import UTC, datetime from types import MappingProxyType +from typing import TYPE_CHECKING +import httpx import pytest from volcano_sdk import ( ConflictError, + DurableExecution, NotFoundError, ServerError, Session, VolcanoClient, ) -from volcano_sdk._transport import DurableExecutionListRequest +from volcano_sdk._transport import DurableExecutionListRequest, GeneratedTransport from .transport_fixtures import RejectingTransport +if TYPE_CHECKING: + from collections.abc import Callable + + from volcano_sdk.durable import Durable + EXECUTION_ID = "00000000-0000-4000-8000-0000000000e1" PROJECT_ID = "00000000-0000-4000-8000-000000000001" +API_URL = "https://api.volcano.test" +APPLICATION_PATH = f"/durable-functions/order-pipeline/executions/{EXECUTION_ID}" @dataclass(frozen=True) @@ -86,6 +97,14 @@ def stop_durable_execution(self, **kwargs: object) -> FakeResponse: self.calls.append(("stopDurableExecution", kwargs)) return self.stop_response + def get_durable_execution_from_application(self, **kwargs: object) -> FakeResponse: + self.calls.append(("getDurableExecutionFromApplication", kwargs)) + return self.get_response + + def stop_durable_execution_from_application(self, **kwargs: object) -> FakeResponse: + self.calls.append(("stopDurableExecutionFromApplication", kwargs)) + return self.stop_response + def durable_client( transport: FakeDurableTransport, @@ -642,3 +661,360 @@ def test_execution_rejects_an_invalid_client_token(value: object) -> None: with pytest.raises(TypeError, match="complete durable execution"): _ = durable_client(transport).durable.start("order-pipeline") + + +def test_get_from_application_reads_with_the_start_credential() -> None: + transport = FakeDurableTransport() + + execution = durable_client(transport, session=False).durable.get_from_application( + "order-pipeline", EXECUTION_ID, client_token="client-token" + ) + + assert execution.status == "succeeded" + assert execution.result == {"charged": "ch_1", "items": (1, 2)} + assert execution.completed_at == datetime(2026, 9, 2, 12, 4, 30, tzinfo=UTC) + assert transport.calls == [ + ( + "getDurableExecutionFromApplication", + { + "authorization": "service-key", + "function_id": "order-pipeline", + "execution_id": EXECUTION_ID, + "client_token": "client-token", + }, + ) + ] + + +def test_stop_from_application_returns_the_execution_read_back() -> None: + transport = FakeDurableTransport() + + execution = durable_client(transport, session=False).durable.stop_from_application( + "order-pipeline", EXECUTION_ID, client_token="client-token" + ) + + assert execution.status == "running" + assert transport.calls == [ + ( + "stopDurableExecutionFromApplication", + { + "authorization": "service-key", + "function_id": "order-pipeline", + "execution_id": EXECUTION_ID, + "client_token": "client-token", + }, + ) + ] + + +APPLICATION_CALLS = { + "get": "getDurableExecutionFromApplication", + "stop": "stopDurableExecutionFromApplication", +} + + +def from_application( + durable: Durable, + operation: str, + *, + function_name: str = "order-pipeline", + execution_id: str = EXECUTION_ID, + client_token: str | None = None, +) -> DurableExecution: + if operation == "get": + return durable.get_from_application( + function_name, execution_id, client_token=client_token + ) + return durable.stop_from_application( + function_name, execution_id, client_token=client_token + ) + + +@pytest.mark.parametrize("operation", APPLICATION_CALLS) +@pytest.mark.parametrize( + ("anon_key", "service_key", "session", "expected"), + [ + pytest.param("anon-key", "service-key", True, "access-token", id="session"), + pytest.param("anon-key", "service-key", False, "service-key", id="service"), + pytest.param("anon-key", None, False, "anon-key", id="anon"), + ], +) +def test_application_routes_send_the_credential_start_sends( + operation: str, + anon_key: str, + service_key: str | None, + *, + session: bool, + expected: str, +) -> None: + transport = FakeDurableTransport() + client = durable_client( + transport, anon_key=anon_key, service_key=service_key, session=session + ) + + _ = from_application(client.durable, operation) + + assert [called for called, _ in transport.calls] == [APPLICATION_CALLS[operation]] + assert transport.calls[0][1]["authorization"] == expected + assert transport.calls[0][1]["client_token"] is None + + +@pytest.mark.parametrize("operation", APPLICATION_CALLS) +def test_application_routes_canonicalize_the_execution_id(operation: str) -> None: + transport = FakeDurableTransport() + + _ = from_application( + durable_client(transport).durable, + operation, + function_name=" order-pipeline ", + execution_id=f" {EXECUTION_ID.upper()} ", + ) + + assert transport.calls[0][1]["function_id"] == "order-pipeline" + assert transport.calls[0][1]["execution_id"] == EXECUTION_ID + + +def test_application_routes_accept_a_client_token_at_the_platform_limit() -> None: + transport = FakeDurableTransport() + + _ = durable_client(transport).durable.get_from_application( + "order-pipeline", EXECUTION_ID, client_token="t" * 128 + ) + + assert transport.calls[0][1]["client_token"] == "t" * 128 + + +INVALID_APPLICATION_ARGUMENTS: dict[str, tuple[Callable[[Durable], object], str]] = { + "get function": ( + lambda d: d.get_from_application(" ", EXECUTION_ID), + "function_name must be a non-empty string", + ), + "get empty execution": ( + lambda d: d.get_from_application("order-pipeline", ""), + "execution_id must be a non-empty string", + ), + "get execution": ( + lambda d: d.get_from_application("order-pipeline", "execution-1"), + "execution_id must be a UUID", + ), + "get empty client token": ( + lambda d: d.get_from_application( + "order-pipeline", EXECUTION_ID, client_token="" + ), + "client_token must be a non-empty string", + ), + "get long client token": ( + lambda d: d.get_from_application( + "order-pipeline", EXECUTION_ID, client_token="t" * 129 + ), + "client_token must be at most 128 characters", + ), + "get non-ASCII client token": ( + lambda d: d.get_from_application( + "order-pipeline", EXECUTION_ID, client_token="tök" + ), + "client_token must be printable ASCII", + ), + "get padded client token": ( + lambda d: d.get_from_application( + "order-pipeline", EXECUTION_ID, client_token=" token" + ), + "client_token must not start or end with whitespace", + ), + "stop function": ( + lambda d: d.stop_from_application("", EXECUTION_ID), + "function_name must be a non-empty string", + ), + "stop execution": ( + lambda d: d.stop_from_application("order-pipeline", "execution-1"), + "execution_id must be a UUID", + ), + "stop client token": ( + lambda d: d.stop_from_application( + "order-pipeline", EXECUTION_ID, client_token="t" * 129 + ), + "client_token must be at most 128 characters", + ), +} + + +@pytest.mark.parametrize( + ("call", "message"), + [ + pytest.param(*case, id=name) + for name, case in INVALID_APPLICATION_ARGUMENTS.items() + ], +) +def test_application_routes_refuse_invalid_arguments_before_transport( + call: Callable[[Durable], object], message: str +) -> None: + transport = FakeDurableTransport() + + with pytest.raises(ValueError, match=f"^{re.escape(message)}$"): + _ = call(durable_client(transport).durable) + + assert transport.calls == [] + + +@pytest.mark.parametrize("operation", APPLICATION_CALLS) +def test_application_routes_raise_not_found_for_an_unreachable_execution( + operation: str, +) -> None: + # The platform answers an execution the credential cannot reach exactly + # as a missing one, so the id cannot be probed. + transport = FakeDurableTransport() + missing = FakeResponse( + 404, {"error": "execution not found", "code": "not_found"}, {} + ) + transport.get_response = missing + transport.stop_response = missing + + with pytest.raises(NotFoundError, match="execution not found") as raised: + _ = from_application( + durable_client(transport, session=False).durable, operation + ) + + assert raised.value.status == 404 + assert raised.value.code == "not_found" + + +def test_stop_from_application_raises_conflict_for_a_pending_execution() -> None: + transport = FakeDurableTransport() + transport.stop_response = FakeResponse( + 409, {"error": "execution has not started", "code": "conflict"}, {} + ) + + with pytest.raises(ConflictError, match="execution has not started") as raised: + _ = durable_client(transport, session=False).durable.stop_from_application( + "order-pipeline", EXECUTION_ID + ) + + assert raised.value.status == 409 + + +def wire_durable( + respond: Callable[[httpx.Request], httpx.Response], + *, + service_key: str | None = None, + session: bool = False, +) -> tuple[Durable, list[httpx.Request]]: + requests: list[httpx.Request] = [] + + def handle(request: httpx.Request) -> httpx.Response: + requests.append(request) + return respond(request) + + client = VolcanoClient( + anon_key="anon-key", + service_key=service_key, + _transport=GeneratedTransport( + api_url=API_URL, httpx_transport=httpx.MockTransport(handle) + ), + ) + if session: + _ = client.auth.set_session( + Session( + access_token="access-token", + refresh_token="refresh-token", + user_id="user-1", + ) + ) + return client.durable, requests + + +@pytest.mark.parametrize( + ("operation", "method", "path"), + [ + ("get", "GET", APPLICATION_PATH), + ("stop", "POST", f"{APPLICATION_PATH}/stop"), + ], +) +@pytest.mark.parametrize( + ("client_token", "headers"), + [ + pytest.param( + "client-token", {"x-volcano-client-token": "client-token"}, id="token" + ), + pytest.param(None, {}, id="no-token"), + ], +) +def test_application_routes_go_over_the_data_plane_path( + operation: str, + method: str, + path: str, + client_token: str | None, + headers: dict[str, str], +) -> None: + durable, requests = wire_durable( + lambda _r: httpx.Response( + 200, json=running_execution(status="succeeded", result={"ok": True}) + ) + ) + + execution = from_application(durable, operation, client_token=client_token) + + assert len(requests) == 1 + assert requests[0].method == method + assert str(requests[0].url) == f"{API_URL}{path}" + assert requests[0].content == b"" + assert requests[0].headers["authorization"] == "Bearer anon-key" + assert { + name: value + for name, value in requests[0].headers.items() + if name == "x-volcano-client-token" + } == headers + assert execution == DurableExecution( + id=EXECUTION_ID, + function_id="00000000-0000-4000-8000-000000000040", + name="order-42", + status="succeeded", + region="aws-us-east-1", + created_at=datetime(2026, 9, 2, 12, 0, tzinfo=UTC), + result={"ok": True}, + ) + + +@pytest.mark.parametrize("operation", APPLICATION_CALLS) +@pytest.mark.parametrize( + ("service_key", "session", "authorization"), + [ + pytest.param("service-key", True, "Bearer access-token", id="session"), + # A service key scoped without `*` reaches an execution only with its + # client token, so the SDK cannot drop the token for a service key. + pytest.param("service-key", False, "Bearer service-key", id="service"), + pytest.param(None, False, "Bearer anon-key", id="anon"), + ], +) +def test_application_routes_send_the_client_token_with_every_credential( + operation: str, + service_key: str | None, + *, + session: bool, + authorization: str, +) -> None: + durable, requests = wire_durable( + lambda _r: httpx.Response(200, json=running_execution()), + service_key=service_key, + session=session, + ) + + _ = from_application(durable, operation, client_token="client-token") + + assert len(requests) == 1 + assert requests[0].headers["authorization"] == authorization + assert requests[0].headers["x-volcano-client-token"] == "client-token" + + +@pytest.mark.parametrize("operation", APPLICATION_CALLS) +def test_application_routes_raise_not_found_over_the_wire(operation: str) -> None: + durable, requests = wire_durable( + lambda _r: httpx.Response( + 404, json={"error": "execution not found", "code": "not_found"} + ) + ) + + with pytest.raises(NotFoundError, match="execution not found") as raised: + _ = from_application(durable, operation, client_token="client-token") + + assert len(requests) == 1 + assert raised.value.code == "not_found" diff --git a/src/volcano_sdk/_transport_execution.py b/src/volcano_sdk/_transport_execution.py index 267c27be..760b6fa1 100644 --- a/src/volcano_sdk/_transport_execution.py +++ b/src/volcano_sdk/_transport_execution.py @@ -9,9 +9,11 @@ from ._generated.api.durable_functions import ( get_durable_execution, + get_durable_execution_from_application, list_durable_executions, start_durable_execution_from_application, stop_durable_execution, + stop_durable_execution_from_application, ) from ._generated.api.functions.invoke_function import ( request_kwargs as invoke_function_kwargs, @@ -209,6 +211,23 @@ def get_durable_execution( ) return parsed_response(response) + def get_durable_execution_from_application( + self, + *, + authorization: str, + function_id: str, + execution_id: str, + client_token: str | None, + ) -> TransportResponse: + with self._client(authorization) as client: + response = get_durable_execution_from_application.sync_detailed( + function_id, + UUID(execution_id), + client=client, + x_volcano_client_token=UNSET if client_token is None else client_token, + ) + return parsed_response(response) + def list_durable_executions( self, *, @@ -244,3 +263,20 @@ def stop_durable_execution( client=client, ) return parsed_response(response) + + def stop_durable_execution_from_application( + self, + *, + authorization: str, + function_id: str, + execution_id: str, + client_token: str | None, + ) -> TransportResponse: + with self._client(authorization) as client: + response = stop_durable_execution_from_application.sync_detailed( + function_id, + UUID(execution_id), + client=client, + x_volcano_client_token=UNSET if client_token is None else client_token, + ) + return parsed_response(response) diff --git a/src/volcano_sdk/durable.py b/src/volcano_sdk/durable.py index fa522e42..c6d9a6c6 100644 --- a/src/volcano_sdk/durable.py +++ b/src/volcano_sdk/durable.py @@ -12,7 +12,12 @@ durable_approval_page, durable_approval_stats, ) -from ._durable_arguments import identifier, known_options +from ._durable_arguments import ( + MAX_CLIENT_TOKEN_LENGTH, + header_value, + identifier, + known_options, +) from ._durable_messages import DurableMessageChannels, DurableMessages from ._durable_response import durable_execution, durable_execution_page from ._transport import ( @@ -98,6 +103,17 @@ def get_durable_execution( """Read one execution of a durable function.""" ... + def get_durable_execution_from_application( + self, + *, + authorization: str, + function_id: str, + execution_id: str, + client_token: str | None, + ) -> TransportResponse: + """Read one execution with the credential that started it.""" + ... + def list_durable_executions( self, *, @@ -120,6 +136,17 @@ def stop_durable_execution( """Ask a running execution to stop.""" ... + def stop_durable_execution_from_application( + self, + *, + authorization: str, + function_id: str, + execution_id: str, + client_token: str | None, + ) -> TransportResponse: + """Ask a running execution to stop with the credential that started it.""" + ... + @runtime_checkable class DurableApprovalsTransport(Protocol): @@ -181,7 +208,7 @@ def deny_durable_approval( class DurableApprovals: """Review and decide the approvals durable workflows wait on. - Owner-scoped, like reading an execution: every method takes the project id + Owner-scoped, like listing executions: every method takes the project id and the session's platform token. Only a person decides: a project access token can read approvals but is refused with `PermissionDeniedError` when it approves or denies one. @@ -361,9 +388,11 @@ def start( A durable function is never invoked synchronously: it can run for hours, so the platform accepts the start and answers with an execution - to follow. Besides messaging an execution through `durable.messages`, - starting is the only durable operation an application credential may - perform -- reading a result or stopping an execution is owner-scoped. + to follow. Follow it with `get_from_application` and + `stop_from_application`, and message it through `durable.messages`, + with the same credential. Keep the returned `client_token` with the + caller that follows it: an anonymous key, or a service key scoped + without `*`, needs it to reach the execution again. Passing `execution_name` makes the start idempotent: starting again under the same name returns the execution that already exists rather @@ -397,11 +426,10 @@ def get( """Read an execution, including its result once it has succeeded. Owner-scoped: it takes the project id and the project's own platform - token or a project access token, because an execution is addressed by - its id alone and an anonymous key is held by everyone who loads the - page. Poll it from a backend, not a browser. Neither an auth-user - session from sign-in nor a service key is accepted here -- the route - answers anything else 401. + token or a project access token, and reaches any execution of the + project. Neither an auth-user session from sign-in nor a service key is + accepted here -- the route answers anything else 401. An application + reads the executions it started with `get_from_application` instead. Returns ------- @@ -469,7 +497,8 @@ def stop( after asking, and it often still says `running`, so poll `get` to see it reach `stopped`. Completed steps are not undone. Repeating a stop is safe -- an execution that has already finished reports the state it is - in. Owner-scoped, like `get`. + in. Owner-scoped, like `get`; an application stops the executions it + started with `stop_from_application` instead. Returns ------- @@ -491,6 +520,89 @@ def stop( response_body: object = response_payload(response, _HTTP_OK) return durable_execution(response_body) + def get_from_application( + self, + function_name: str, + execution_id: str, + *, + client_token: str | None = None, + ) -> DurableExecution: + """Read an execution with the credential that started it. + + Takes the client's credential, as `start` does, and the function's + visibility must admit it. A service key holding `*`, the default, + reaches any execution of the project; a signed-in user reaches the + executions they started. An anonymous key, a service key scoped + without `*`, or a user who did not start the execution, passes the + `client_token` its start returned. An anonymous key or a service key + scoped without `*` also needs `functions.invoke`; without it the call + raises `PermissionDeniedError`. An anonymous key on an `authenticated` + function raises `PermissionDeniedError` too, even with the token, and a + `private` function raises `NotFoundError` for every credential but a + service key. An execution the caller cannot reach raises + `NotFoundError`, as a missing one does, and so does one that has not + started yet, except to a service key holding `*`. Reads share the + function's rate limit with starts, so poll every few seconds. + + Returns + ------- + DurableExecution + The execution snapshot, including any available result or failure. + + """ + return self._from_application("get", function_name, execution_id, client_token) + + def stop_from_application( + self, + function_name: str, + execution_id: str, + *, + client_token: str | None = None, + ) -> DurableExecution: + """Ask a running execution to stop, with the credential that started it. + + Reaches what `get_from_application` reaches. Accepted rather than + awaited, like `stop`: the execution it returns often still says + `running`, so poll `get_from_application` to see it reach `stopped`. + Completed steps are not undone, and repeating a stop is safe. A service + key holding `*` stopping an execution that has not started yet raises + `ConflictError`. A `ServerError` can follow a stop that was already + sent, so read the execution before stopping it again. + + Returns + ------- + DurableExecution + The execution snapshot after the stop request; it may still be running. + + """ + return self._from_application("stop", function_name, execution_id, client_token) + + def _from_application( + self, + operation: Literal["get", "stop"], + function_name: str, + execution_id: str, + client_token: object, + ) -> DurableExecution: + function = identifier(function_name, "function_name") + execution = identifier(execution_id, "execution_id") + token = header_value(client_token, "client_token", MAX_CLIENT_TOKEN_LENGTH) + transport = self._durable_transport() + request = ( + transport.get_durable_execution_from_application + if operation == "get" + else transport.stop_durable_execution_from_application + ) + response = invoke( + request, + authorization=self._client.function_token(), + function_id=function, + execution_id=execution, + client_token=token, + ) + response_body: object = response_payload(response, _HTTP_OK) + return durable_execution(response_body) + def _execution_name(value: object) -> str: """Require a name the platform will accept, including its length. diff --git a/src/volcano_sdk/models.py b/src/volcano_sdk/models.py index c463eea0..f43e1cf7 100644 --- a/src/volcano_sdk/models.py +++ b/src/volcano_sdk/models.py @@ -381,9 +381,9 @@ class DurableExecution: result_expired: bool | None = None error: DurableExecutionFailure | None = None completed_at: datetime | None = None - # Lets whoever started the execution send it messages and read what it - # publishes. Returned only by the start that created the execution, so - # keep it with the execution id. + # Lets whoever started the execution read it, stop it, send it messages, + # and read what it publishes. Returned only by the start that created the + # execution, so keep it with the execution id. client_token: str | None = field(default=None, repr=False) def __post_init__(self) -> None: