From a260c117a82b599c7a775d7d27db7c1722a34d0f Mon Sep 17 00:00:00 2001 From: Glenn Jocher Date: Fri, 30 Oct 2026 11:58:59 +0100 Subject: [PATCH 1/2] Wake the local On Premise worker from every SDK client --- .github/workflows/ci.yml | 10 ++--- AGENTS.md | 8 ++-- README.md | 4 +- cli.py | 24 +---------- hooks.py | 43 +++++++++++++++++++ openapi.config.json | 3 +- .../src/ultralytics_platform/_client.py | 7 ++- sdk/python/src/ultralytics_platform/_hooks.py | 43 +++++++++++++++++++ .../src/ultralytics_platform/async_client.py | 2 +- sdk/python/src/ultralytics_platform/cli.py | 24 +---------- sdk/python/src/ultralytics_platform/client.py | 2 +- tests/test_python_sdk.py | 21 +++++++++ 12 files changed, 131 insertions(+), 60 deletions(-) create mode 100644 hooks.py create mode 100644 sdk/python/src/ultralytics_platform/_hooks.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 6c77102..aebe55c 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -72,8 +72,8 @@ jobs: - name: Validate Python SDK run: | - uvx ruff@0.16.2 format --check --line-length 120 sdk/python tests cli.py auth.py - uvx ruff@0.16.2 check sdk/python tests cli.py auth.py + uvx ruff@0.16.2 format --check --line-length 120 sdk/python tests cli.py auth.py hooks.py + uvx ruff@0.16.2 check sdk/python tests cli.py auth.py hooks.py python -m compileall -q sdk/python/src uv build --project sdk/python unzip -l sdk/python/dist/*.whl | grep --quiet '\.dist-info/licenses/LICENSE$' @@ -134,13 +134,13 @@ jobs: exit 0 fi echo "changed=true" >> "$GITHUB_OUTPUT" - echo "hash=$(printf '%s\n%s\n%s\n' "$(sha256sum openapi.json)" "$(git -C .generator rev-parse HEAD)" "$(git ls-files cli.py auth.py openapi.config.json README.python.md | xargs sha256sum)" | sha256sum | cut -c1-12)" >> "$GITHUB_OUTPUT" + echo "hash=$(printf '%s\n%s\n%s\n' "$(sha256sum openapi.json)" "$(git -C .generator rev-parse HEAD)" "$(git ls-files cli.py auth.py hooks.py openapi.config.json README.python.md | xargs sha256sum)" | sha256sum | cut -c1-12)" >> "$GITHUB_OUTPUT" - name: Validate regenerated Python SDK if: steps.sync.outputs.changed == 'true' run: | - uvx ruff@0.16.2 format --check --line-length 120 sdk/python tests cli.py auth.py - uvx ruff@0.16.2 check sdk/python tests cli.py auth.py + uvx ruff@0.16.2 format --check --line-length 120 sdk/python tests cli.py auth.py hooks.py + uvx ruff@0.16.2 check sdk/python tests cli.py auth.py hooks.py python -m compileall -q sdk/python/src uv build --project sdk/python uv pip install pytest jsonschema referencing "./sdk/python" diff --git a/AGENTS.md b/AGENTS.md index 7a894fd..ff81dda 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -57,15 +57,15 @@ export OPENAPI_CONFIG="$PWD/openapi.config.json" diff --recursive --unified sdk/python .generator/generated/python pytest tests -v -uvx ruff@0.16.2 format --check --line-length 120 sdk/python tests cli.py auth.py -uvx ruff@0.16.2 check sdk/python tests cli.py auth.py +uvx ruff@0.16.2 format --check --line-length 120 sdk/python tests cli.py auth.py hooks.py +uvx ruff@0.16.2 check sdk/python tests cli.py auth.py hooks.py ``` Use the generator's `main` branch, never a pinned SHA or tag; update an existing `.generator` checkout before regenerating. Check drift before running tests/builds, which leave ignored artifacts in the generated tree. After intentional source changes, inspect the diff, then copy with `rsync --archive --delete .generator/generated/python/ sdk/python/`. Never repair generated files by hand. Install the package in the same interpreter that runs pytest. Use `python -m ultralytics_platform.cli` if the system `ul` command shadows the console script. ## Where to look -- CLI and credential customization → `cli.py`, `auth.py`. +- CLI, credential, and response hook customization → `cli.py`, `auth.py`, `hooks.py`. - Generator configuration and README template → `openapi.config.json`, `README.python.md`. - Pinned contract → `openapi.json`, `openapi.sha256`. - Generated package → `sdk/python/`. @@ -76,7 +76,7 @@ Use the generator's `main` branch, never a pinned SHA or tag; update an existing - Ultralytics-owned PyPI packages use `MAJOR.MINOR.PATCH` versions only; no suffixes. - License headers (`# Ultralytics 🚀 AGPL-3.0 License - https://ultralytics.com/license`) are added automatically by Ultralytics Actions — don't add or revert them manually. Generated files receive the same header from the generator (`header` in `openapi.config.json`). -- Google-style docstrings, `from __future__ import annotations` for modern type hints, line length 120. `format.yml` sets `python: false` and `prettier: false`, so nothing reformats Python or Markdown on PR branches: run `uvx ruff@0.16.2 format --line-length 120 cli.py auth.py tests` yourself before pushing; the generator formats `sdk/python`. +- Google-style docstrings, `from __future__ import annotations` for modern type hints, line length 120. `format.yml` sets `python: false` and `prettier: false`, so nothing reformats Python or Markdown on PR branches: run `uvx ruff@0.16.2 format --line-length 120 cli.py auth.py hooks.py tests` yourself before pushing; the generator formats `sdk/python`. ## Pitfalls diff --git a/README.md b/README.md index 8b2087a..00e2c66 100644 --- a/README.md +++ b/README.md @@ -22,7 +22,7 @@ Typed SDKs for the [Ultralytics Platform API](https://platform.ultralytics.com), ## 🐍 Python -The package includes the unified `ul` CLI for local YOLO and Platform SDK commands; see the [CLI guide](README.python.md#unified-ul-cli). Maintain `cli.py` and `auth.py` at the repository root, then regenerate their installed copies through the configured `python.cli.source` and `python.authProvider` hooks. +The package includes the unified `ul` CLI for local YOLO and Platform SDK commands; see the [CLI guide](README.python.md#unified-ul-cli). Maintain `cli.py`, `auth.py`, and `hooks.py` at the repository root, then regenerate their installed copies through the configured `python.cli.source`, `python.authProvider`, and `python.hooksProvider` hooks. [![PyPI - Version](https://img.shields.io/pypi/v/ultralytics-platform?logo=pypi&logoColor=white)](https://pypi.org/project/ultralytics-platform/) [![Ultralytics Downloads](https://static.pepy.tech/badge/ultralytics-platform)](https://clickpy.clickhouse.com/dashboard/ultralytics-platform) [![PyPI - Python Version](https://img.shields.io/pypi/pyversions/ultralytics-platform?logo=python&logoColor=gold)](https://pypi.org/project/ultralytics-platform/) @@ -87,7 +87,7 @@ Platform OpenAPI contract └── java/ # coming soon ``` -`openapi.config.json` contains product and package configuration, including the source path for `README.python.md`. `auth.py` owns the Python SDK's Ultralytics credential discovery and is included by the generic `python.authProvider` hook on every regeneration. `openapi.json` and `openapi.sha256` pin the exact consumed contract. Generated files are never edited manually; update the contract snapshot, configuration, package README source, or generator and regenerate. +`openapi.config.json` contains product and package configuration, including the source path for `README.python.md`. `auth.py` owns the Python SDK's Ultralytics credential discovery and is included by the generic `python.authProvider` hook on every regeneration; `hooks.py`, included by `python.hooksProvider`, wakes a local On Premise worker after the SDK creates an ingest or training job. `openapi.json` and `openapi.sha256` pin the exact consumed contract. Generated files are never edited manually; update the contract snapshot, configuration, package README source, or generator and regenerate. ## 🛠️ Validation diff --git a/cli.py b/cli.py index f7e8431..2eb0c0c 100644 --- a/cli.py +++ b/cli.py @@ -13,15 +13,13 @@ import typing import zipfile from collections.abc import Sequence -from contextlib import ExitStack, suppress +from contextlib import ExitStack from importlib.metadata import PackageNotFoundError, version from pathlib import Path from tarfile import is_tarfile from tempfile import TemporaryDirectory from typing import Any, BinaryIO, Literal, get_args, get_origin, get_type_hints -import httpx - from . import NOT_GIVEN, APIConnectionError, APIError, NotGiven, Platform from ._cli_metadata import MULTIPART_FILES @@ -35,23 +33,6 @@ def platform_url() -> str: return os.getenv("ULTRALYTICS_PLATFORM_URL", "https://platform.ultralytics.com").rstrip("/") -def wake_worker(response: httpx.Response) -> None: - """Have an On Premise worker on this machine claim a just-created ingest or training job instead of polling. - - Responses do not say where a job runs, so every one wakes; without a worker, or an old one, nothing changes. - """ - request = response.request - if response.is_success and request.method == "POST" and request.url.path.endswith(("/ingest", "/training/start")): - # stream returns after the headers, so a listener that drips a body can't hold the CLI open - with ( - suppress(httpx.HTTPError), - httpx.stream( - "GET", "http://127.0.0.1:8765/wake", headers={"Origin": platform_url()}, timeout=0.5, trust_env=False - ), - ): - pass - - def kinds(annotation: Any) -> tuple[set[str], list | None]: """Map a generated SDK annotation to JSON kinds and literal choices; empty kinds accept any JSON value.""" if annotation is Any: @@ -734,8 +715,7 @@ def dispatch(tokens: list[str]) -> int: """Execute one SDK operation, inferring only a missing resource-path owner.""" tokens = list(tokens) with ExitStack() as stack: - http_client = httpx.Client(timeout=60.0, event_hooks={"response": [wake_worker]}) - client = stack.enter_context(Platform(base_url=platform_url(), http_client=http_client)) + client = stack.enter_context(Platform(base_url=platform_url())) if tokens and tokens[0] in CLOUD_COMMANDS: if len(tokens) == 1 or any(token in {"help", "--help", "-h"} for token in tokens): print(inspect.getdoc(CLOUD_COMMANDS[tokens[0]])) diff --git a/hooks.py b/hooks.py new file mode 100644 index 0000000..7a2d939 --- /dev/null +++ b/hooks.py @@ -0,0 +1,43 @@ +# Ultralytics 🚀 AGPL-3.0 License - https://ultralytics.com/license +"""Response hooks that Platform and AsyncPlatform register on the HTTP clients they create; they must never raise.""" + +from __future__ import annotations + +from contextlib import suppress + +import httpx + +WAKE_URL = "http://127.0.0.1:8765/wake" # the On Premise worker's fixed local port + + +def job_origin(response: httpx.Response) -> str | None: + """Return the Platform origin of a just-created ingest or training job, else None. + + Responses do not say where a job runs, so every one wakes; without a worker, or an old one, nothing changes. + """ + request = response.request + if response.is_success and request.method == "POST" and request.url.path.endswith(("/ingest", "/training/start")): + return f"{request.url.scheme}://{request.url.netloc.decode()}" # workers accept only their Platform's origin + return None + + +def response_hook(response: httpx.Response) -> None: + """Have an On Premise worker on this machine claim a just-created ingest or training job instead of polling.""" + if origin := job_origin(response): + # stream returns after the headers, so a listener that drips a body can't hold the caller open + with ( + suppress(httpx.HTTPError), + httpx.stream("GET", WAKE_URL, headers={"Origin": origin}, timeout=0.5, trust_env=False), + ): + pass + + +async def async_response_hook(response: httpx.Response) -> None: + """Wake the local On Premise worker like response_hook, without blocking the event loop.""" + if origin := job_origin(response): + with suppress(httpx.HTTPError): + async with ( + httpx.AsyncClient(trust_env=False) as client, + client.stream("GET", WAKE_URL, headers={"Origin": origin}, timeout=0.5), + ): + pass diff --git a/openapi.config.json b/openapi.config.json index 956c879..53c7647 100644 --- a/openapi.config.json +++ b/openapi.config.json @@ -57,6 +57,7 @@ "project": "ultralytics-platform", "readme": "README.python.md", "requiresPython": ">=3.11", - "authProvider": "auth.py" + "authProvider": "auth.py", + "hooksProvider": "hooks.py" } } diff --git a/sdk/python/src/ultralytics_platform/_client.py b/sdk/python/src/ultralytics_platform/_client.py index 006986c..bf24fce 100644 --- a/sdk/python/src/ultralytics_platform/_client.py +++ b/sdk/python/src/ultralytics_platform/_client.py @@ -14,6 +14,7 @@ from ._auth import get_api_key from ._exceptions import APIConnectionError, APIError +from ._hooks import async_response_hook, response_hook def _resolve_api_key(api_key: str | None) -> str | None: @@ -126,7 +127,7 @@ def __init__( retry_methods: Sequence[str], http_client: httpx.Client | None, ) -> None: - self._client = http_client or httpx.Client(timeout=timeout) + self._client = http_client or httpx.Client(timeout=timeout, event_hooks={"response": [response_hook]}) self._base_url = httpx.URL(f"{base_url.rstrip('/')}/") self._api_key = api_key self._max_retries = max_retries @@ -192,7 +193,9 @@ def __init__( retry_methods: Sequence[str], http_client: httpx.AsyncClient | None, ) -> None: - self._client = http_client or httpx.AsyncClient(timeout=timeout) + self._client = http_client or httpx.AsyncClient( + timeout=timeout, event_hooks={"response": [async_response_hook]} + ) self._base_url = httpx.URL(f"{base_url.rstrip('/')}/") self._api_key = api_key self._max_retries = max_retries diff --git a/sdk/python/src/ultralytics_platform/_hooks.py b/sdk/python/src/ultralytics_platform/_hooks.py new file mode 100644 index 0000000..7a2d939 --- /dev/null +++ b/sdk/python/src/ultralytics_platform/_hooks.py @@ -0,0 +1,43 @@ +# Ultralytics 🚀 AGPL-3.0 License - https://ultralytics.com/license +"""Response hooks that Platform and AsyncPlatform register on the HTTP clients they create; they must never raise.""" + +from __future__ import annotations + +from contextlib import suppress + +import httpx + +WAKE_URL = "http://127.0.0.1:8765/wake" # the On Premise worker's fixed local port + + +def job_origin(response: httpx.Response) -> str | None: + """Return the Platform origin of a just-created ingest or training job, else None. + + Responses do not say where a job runs, so every one wakes; without a worker, or an old one, nothing changes. + """ + request = response.request + if response.is_success and request.method == "POST" and request.url.path.endswith(("/ingest", "/training/start")): + return f"{request.url.scheme}://{request.url.netloc.decode()}" # workers accept only their Platform's origin + return None + + +def response_hook(response: httpx.Response) -> None: + """Have an On Premise worker on this machine claim a just-created ingest or training job instead of polling.""" + if origin := job_origin(response): + # stream returns after the headers, so a listener that drips a body can't hold the caller open + with ( + suppress(httpx.HTTPError), + httpx.stream("GET", WAKE_URL, headers={"Origin": origin}, timeout=0.5, trust_env=False), + ): + pass + + +async def async_response_hook(response: httpx.Response) -> None: + """Wake the local On Premise worker like response_hook, without blocking the event loop.""" + if origin := job_origin(response): + with suppress(httpx.HTTPError): + async with ( + httpx.AsyncClient(trust_env=False) as client, + client.stream("GET", WAKE_URL, headers={"Origin": origin}, timeout=0.5), + ): + pass diff --git a/sdk/python/src/ultralytics_platform/async_client.py b/sdk/python/src/ultralytics_platform/async_client.py index 45061fd..82c2aff 100644 --- a/sdk/python/src/ultralytics_platform/async_client.py +++ b/sdk/python/src/ultralytics_platform/async_client.py @@ -46,7 +46,7 @@ def __init__( timeout (float | httpx.Timeout): Request timeout. max_retries (int): Retries for connection errors and retryable responses. retry_methods (Sequence[str]): HTTP methods safe to repeat after network errors or HTTP 408, 409, 429, and 5xx responses. Defaults to GET, HEAD, OPTIONS. Opt in only when the operation is safe to repeat and its request body can be resent. - http_client (httpx.AsyncClient, optional): Custom HTTP client. + http_client (httpx.AsyncClient, optional): Custom HTTP client, used as is without the configured response hooks. """ resolved_api_key = _resolve_api_key(api_key) self._client = AsyncAPIClient( diff --git a/sdk/python/src/ultralytics_platform/cli.py b/sdk/python/src/ultralytics_platform/cli.py index f7e8431..2eb0c0c 100644 --- a/sdk/python/src/ultralytics_platform/cli.py +++ b/sdk/python/src/ultralytics_platform/cli.py @@ -13,15 +13,13 @@ import typing import zipfile from collections.abc import Sequence -from contextlib import ExitStack, suppress +from contextlib import ExitStack from importlib.metadata import PackageNotFoundError, version from pathlib import Path from tarfile import is_tarfile from tempfile import TemporaryDirectory from typing import Any, BinaryIO, Literal, get_args, get_origin, get_type_hints -import httpx - from . import NOT_GIVEN, APIConnectionError, APIError, NotGiven, Platform from ._cli_metadata import MULTIPART_FILES @@ -35,23 +33,6 @@ def platform_url() -> str: return os.getenv("ULTRALYTICS_PLATFORM_URL", "https://platform.ultralytics.com").rstrip("/") -def wake_worker(response: httpx.Response) -> None: - """Have an On Premise worker on this machine claim a just-created ingest or training job instead of polling. - - Responses do not say where a job runs, so every one wakes; without a worker, or an old one, nothing changes. - """ - request = response.request - if response.is_success and request.method == "POST" and request.url.path.endswith(("/ingest", "/training/start")): - # stream returns after the headers, so a listener that drips a body can't hold the CLI open - with ( - suppress(httpx.HTTPError), - httpx.stream( - "GET", "http://127.0.0.1:8765/wake", headers={"Origin": platform_url()}, timeout=0.5, trust_env=False - ), - ): - pass - - def kinds(annotation: Any) -> tuple[set[str], list | None]: """Map a generated SDK annotation to JSON kinds and literal choices; empty kinds accept any JSON value.""" if annotation is Any: @@ -734,8 +715,7 @@ def dispatch(tokens: list[str]) -> int: """Execute one SDK operation, inferring only a missing resource-path owner.""" tokens = list(tokens) with ExitStack() as stack: - http_client = httpx.Client(timeout=60.0, event_hooks={"response": [wake_worker]}) - client = stack.enter_context(Platform(base_url=platform_url(), http_client=http_client)) + client = stack.enter_context(Platform(base_url=platform_url())) if tokens and tokens[0] in CLOUD_COMMANDS: if len(tokens) == 1 or any(token in {"help", "--help", "-h"} for token in tokens): print(inspect.getdoc(CLOUD_COMMANDS[tokens[0]])) diff --git a/sdk/python/src/ultralytics_platform/client.py b/sdk/python/src/ultralytics_platform/client.py index 9acfe4d..928fe4c 100644 --- a/sdk/python/src/ultralytics_platform/client.py +++ b/sdk/python/src/ultralytics_platform/client.py @@ -46,7 +46,7 @@ def __init__( timeout (float | httpx.Timeout): Request timeout. max_retries (int): Retries for connection errors and retryable responses. retry_methods (Sequence[str]): HTTP methods safe to repeat after network errors or HTTP 408, 409, 429, and 5xx responses. Defaults to GET, HEAD, OPTIONS. Opt in only when the operation is safe to repeat and its request body can be resent. - http_client (httpx.Client, optional): Custom HTTP client. + http_client (httpx.Client, optional): Custom HTTP client, used as is without the configured response hooks. """ resolved_api_key = _resolve_api_key(api_key) self._client = SyncAPIClient( diff --git a/tests/test_python_sdk.py b/tests/test_python_sdk.py index 55c3d6d..da17bc5 100644 --- a/tests/test_python_sdk.py +++ b/tests/test_python_sdk.py @@ -12,6 +12,7 @@ import httpx import pytest +from test_cli import loopback from ultralytics_platform import APIError, AsyncPlatform, Platform ROOT = Path(__file__).parents[1] @@ -159,3 +160,23 @@ def handle(request: httpx.Request) -> httpx.Response: assert raised.value.status_code == 422 assert raised.value.body == '{"detail":"invalid"}' assert raised.value.request_id == "request-1" + + +def test_async_client_wakes_local_worker() -> None: + wakes = [] + + def wake(handler, body): + wakes.append((handler.path, handler.headers["Origin"])) + return 204, b"" + + async def run(url: str) -> None: + async with AsyncPlatform(api_key="", base_url=url) as client: + await client.datasets.list("user") + with pytest.raises(APIError): + await client.datasets.ingest("user", "fail", body={}) + await client.training.start(model_id="model-id", train_args={}) + await client.datasets.ingest("user", "dataset", body={}) + + with loopback(lambda handler, body: (422 if "/fail/" in handler.path else 200, {})) as url, loopback(wake, 8765): + asyncio.run(run(url)) + assert wakes == [("/wake", url)] * 2 # not reads or failures; the CLI wire test covers the sync Platform From 194be8d237421d4676ed6ff7bb4f524c3ed192c9 Mon Sep 17 00:00:00 2001 From: Glenn Jocher Date: Fri, 30 Oct 2026 12:03:48 +0100 Subject: [PATCH 2/2] Trim the wake hooks, drop the extra test, document the wake --- README.python.md | 2 + hooks.py | 38 +++++++------------ sdk/python/README.md | 2 + sdk/python/src/ultralytics_platform/_hooks.py | 38 +++++++------------ tests/test_python_sdk.py | 21 ---------- 5 files changed, 32 insertions(+), 69 deletions(-) diff --git a/README.python.md b/README.python.md index 2036c5e..adfdfe6 100644 --- a/README.python.md +++ b/README.python.md @@ -43,6 +43,8 @@ asyncio.run(main()) The package includes typed responses, multipart uploads, retries for temporary failures, structured API errors, custom HTTP clients, and context-manager cleanup. +After a successful ingest or training start, both clients send a best-effort `GET http://127.0.0.1:8765/wake` so an On Premise worker on the same machine claims the job at once; a custom `http_client` skips it. + ## Unified `ul` CLI This package installs `ul`. API resource commands work without the ML package. Local commands and `ul cloud train/predict/export` require `ultralytics` in the same environment. Existing `yolo` behavior, including local training with `ul://` inputs, is unchanged. diff --git a/hooks.py b/hooks.py index 7a2d939..749b07a 100644 --- a/hooks.py +++ b/hooks.py @@ -1,5 +1,5 @@ # Ultralytics 🚀 AGPL-3.0 License - https://ultralytics.com/license -"""Response hooks that Platform and AsyncPlatform register on the HTTP clients they create; they must never raise.""" +"""Response hooks for Platform and AsyncPlatform: wake a local On Premise worker after creating its job.""" from __future__ import annotations @@ -7,37 +7,27 @@ import httpx -WAKE_URL = "http://127.0.0.1:8765/wake" # the On Premise worker's fixed local port - -def job_origin(response: httpx.Response) -> str | None: - """Return the Platform origin of a just-created ingest or training job, else None. - - Responses do not say where a job runs, so every one wakes; without a worker, or an old one, nothing changes. - """ - request = response.request - if response.is_success and request.method == "POST" and request.url.path.endswith(("/ingest", "/training/start")): - return f"{request.url.scheme}://{request.url.netloc.decode()}" # workers accept only their Platform's origin +def wake_request(response: httpx.Response) -> dict | None: + """Return the local wake request for a created ingest or training job; responses don't say where a job runs.""" + url = response.request.url + if response.is_success and response.request.method == "POST" and url.path.endswith(("/ingest", "/training/start")): + # Workers accept only their Platform's origin; reading headers only keeps a dripping listener from blocking + origin = f"{url.scheme}://{url.netloc.decode()}" + return {"method": "GET", "url": "http://127.0.0.1:8765/wake", "headers": {"Origin": origin}, "timeout": 0.5} return None def response_hook(response: httpx.Response) -> None: - """Have an On Premise worker on this machine claim a just-created ingest or training job instead of polling.""" - if origin := job_origin(response): - # stream returns after the headers, so a listener that drips a body can't hold the caller open - with ( - suppress(httpx.HTTPError), - httpx.stream("GET", WAKE_URL, headers={"Origin": origin}, timeout=0.5, trust_env=False), - ): + """Wake the local worker; never raises.""" + if request := wake_request(response): + with suppress(httpx.HTTPError), httpx.stream(**request, trust_env=False): pass async def async_response_hook(response: httpx.Response) -> None: - """Wake the local On Premise worker like response_hook, without blocking the event loop.""" - if origin := job_origin(response): + """Wake the local worker without blocking the event loop; never raises.""" + if request := wake_request(response): with suppress(httpx.HTTPError): - async with ( - httpx.AsyncClient(trust_env=False) as client, - client.stream("GET", WAKE_URL, headers={"Origin": origin}, timeout=0.5), - ): + async with httpx.AsyncClient(trust_env=False) as client, client.stream(**request): pass diff --git a/sdk/python/README.md b/sdk/python/README.md index 2036c5e..adfdfe6 100644 --- a/sdk/python/README.md +++ b/sdk/python/README.md @@ -43,6 +43,8 @@ asyncio.run(main()) The package includes typed responses, multipart uploads, retries for temporary failures, structured API errors, custom HTTP clients, and context-manager cleanup. +After a successful ingest or training start, both clients send a best-effort `GET http://127.0.0.1:8765/wake` so an On Premise worker on the same machine claims the job at once; a custom `http_client` skips it. + ## Unified `ul` CLI This package installs `ul`. API resource commands work without the ML package. Local commands and `ul cloud train/predict/export` require `ultralytics` in the same environment. Existing `yolo` behavior, including local training with `ul://` inputs, is unchanged. diff --git a/sdk/python/src/ultralytics_platform/_hooks.py b/sdk/python/src/ultralytics_platform/_hooks.py index 7a2d939..749b07a 100644 --- a/sdk/python/src/ultralytics_platform/_hooks.py +++ b/sdk/python/src/ultralytics_platform/_hooks.py @@ -1,5 +1,5 @@ # Ultralytics 🚀 AGPL-3.0 License - https://ultralytics.com/license -"""Response hooks that Platform and AsyncPlatform register on the HTTP clients they create; they must never raise.""" +"""Response hooks for Platform and AsyncPlatform: wake a local On Premise worker after creating its job.""" from __future__ import annotations @@ -7,37 +7,27 @@ import httpx -WAKE_URL = "http://127.0.0.1:8765/wake" # the On Premise worker's fixed local port - -def job_origin(response: httpx.Response) -> str | None: - """Return the Platform origin of a just-created ingest or training job, else None. - - Responses do not say where a job runs, so every one wakes; without a worker, or an old one, nothing changes. - """ - request = response.request - if response.is_success and request.method == "POST" and request.url.path.endswith(("/ingest", "/training/start")): - return f"{request.url.scheme}://{request.url.netloc.decode()}" # workers accept only their Platform's origin +def wake_request(response: httpx.Response) -> dict | None: + """Return the local wake request for a created ingest or training job; responses don't say where a job runs.""" + url = response.request.url + if response.is_success and response.request.method == "POST" and url.path.endswith(("/ingest", "/training/start")): + # Workers accept only their Platform's origin; reading headers only keeps a dripping listener from blocking + origin = f"{url.scheme}://{url.netloc.decode()}" + return {"method": "GET", "url": "http://127.0.0.1:8765/wake", "headers": {"Origin": origin}, "timeout": 0.5} return None def response_hook(response: httpx.Response) -> None: - """Have an On Premise worker on this machine claim a just-created ingest or training job instead of polling.""" - if origin := job_origin(response): - # stream returns after the headers, so a listener that drips a body can't hold the caller open - with ( - suppress(httpx.HTTPError), - httpx.stream("GET", WAKE_URL, headers={"Origin": origin}, timeout=0.5, trust_env=False), - ): + """Wake the local worker; never raises.""" + if request := wake_request(response): + with suppress(httpx.HTTPError), httpx.stream(**request, trust_env=False): pass async def async_response_hook(response: httpx.Response) -> None: - """Wake the local On Premise worker like response_hook, without blocking the event loop.""" - if origin := job_origin(response): + """Wake the local worker without blocking the event loop; never raises.""" + if request := wake_request(response): with suppress(httpx.HTTPError): - async with ( - httpx.AsyncClient(trust_env=False) as client, - client.stream("GET", WAKE_URL, headers={"Origin": origin}, timeout=0.5), - ): + async with httpx.AsyncClient(trust_env=False) as client, client.stream(**request): pass diff --git a/tests/test_python_sdk.py b/tests/test_python_sdk.py index da17bc5..55c3d6d 100644 --- a/tests/test_python_sdk.py +++ b/tests/test_python_sdk.py @@ -12,7 +12,6 @@ import httpx import pytest -from test_cli import loopback from ultralytics_platform import APIError, AsyncPlatform, Platform ROOT = Path(__file__).parents[1] @@ -160,23 +159,3 @@ def handle(request: httpx.Request) -> httpx.Response: assert raised.value.status_code == 422 assert raised.value.body == '{"detail":"invalid"}' assert raised.value.request_id == "request-1" - - -def test_async_client_wakes_local_worker() -> None: - wakes = [] - - def wake(handler, body): - wakes.append((handler.path, handler.headers["Origin"])) - return 204, b"" - - async def run(url: str) -> None: - async with AsyncPlatform(api_key="", base_url=url) as client: - await client.datasets.list("user") - with pytest.raises(APIError): - await client.datasets.ingest("user", "fail", body={}) - await client.training.start(model_id="model-id", train_args={}) - await client.datasets.ingest("user", "dataset", body={}) - - with loopback(lambda handler, body: (422 if "/fail/" in handler.path else 200, {})) as url, loopback(wake, 8765): - asyncio.run(run(url)) - assert wakes == [("/wake", url)] * 2 # not reads or failures; the CLI wire test covers the sync Platform