Skip to content

Commit d119d8e

Browse files
CyMuleNick Franck
andauthored
feat(etl-uvicorn): give plugins request-scoped settings and invocation context (#74)
## Summary - Resolve and bind `invocation_settings` plus `invocation_context` once per `/invoke` request for synchronous, asynchronous, and streaming plugins. - Generate local Python bindings from the ratified [`invocation-context/v1`](https://schemas.u10d.dev/invocation-context/v1.json) and [`errors/audience/v1`](https://schemas.u10d.dev/errors/audience/v1.json) contracts. - Advertise generic invocation transport capabilities and the format-specific `invoke_with_sealed_dag_node_settings_v2` capability by construction. - Map legacy `UserError` failures to a complete nested `plugin_error` envelope with `audience=user`; remove the redundant top-level `blame` spelling. - Preserve exact-format, default-deny sealed-settings dispatch during mixed-version rollout. ## Why Request limits, route registration, thread-context propagation, and HTTP responses are transport concerns shared by every FastAPI-wrapped plugin. The wrapper owns those mechanics while `utic-invocation-settings` remains the authority for document validation and field-envelope resolution. Cross-service wire contracts are referenced by schema `$id` and code-generated locally instead of being imported from another service package. The handwritten adapter retains only transport-specific error mapping and the equal-length batch invariant that JSON Schema cannot express. The generic transport capability never authorizes ciphertext. The controller dispatches on the document's own `format` and forwards sealed settings only when the plugin advertises the exact matching capability; v1 remains unchanged and v2 requires `invoke_with_sealed_dag_node_settings_v2`. ## Merge dependency Unstructured-IO/utic-public-libs#67 is merged and `utic-invocation-settings 0.5.0` is published to public PyPI. This branch now resolves that released wheel and contains no temporary source pin. - [x] Merge #67 and publish `utic-invocation-settings 0.5.0`. - [x] Remove the temporary source pin, relock against the published package, and verify a clean install. - [ ] Merge and publish this package before downstream PRs remove their remaining wrapper pin. ## Validation - `uv run --locked pytest -q` — 187 passed - `uv run --locked python scripts/generate_invocation_contracts.py --check` — generated bindings match the ratified schemas - `uv run --locked ruff check .` — clean - `git diff --check` — clean Part of DTPL-648, part of DTPL-707 --------- Co-authored-by: Nick Franck <nickfranck@unstructured.io>
1 parent 6273c41 commit d119d8e

20 files changed

Lines changed: 2444 additions & 189 deletions

.github/workflows/ci.yml

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,7 @@ jobs:
1616
runs-on: ubuntu-latest
1717
strategy:
1818
matrix:
19-
python-version: [ "3.10", "3.11", "3.12", "3.13" ]
19+
python-version: [ "3.11", "3.12", "3.13" ]
2020
steps:
2121
- uses: actions/checkout@v3
2222

@@ -47,7 +47,7 @@ jobs:
4747
runs-on: ubuntu-latest
4848
strategy:
4949
matrix:
50-
python-version: [ "3.10", "3.11", "3.12", "3.13" ]
50+
python-version: [ "3.11", "3.12", "3.13" ]
5151
steps:
5252
- uses: actions/checkout@v3
5353

.github/workflows/release.yml

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@ on:
66
- published
77

88
env:
9-
PYTHON_VERSION: "3.10"
9+
PYTHON_VERSION: "3.11"
1010

1111
jobs:
1212
release:

CHANGELOG.md

Lines changed: 57 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,60 @@
1+
## 0.1.0
2+
3+
* **This package now owns the `/invoke` transport for the reserved fields.**
4+
`unstructured_platform_plugins.invocation_settings` holds the `/invoke` binding dependency and
5+
body cap, the `/metadata` capability route, the request-scoped accessors, and `http_status_for`
6+
— the HTTP spelling of the
7+
library's normative `blame` → status rule. It sits on `utic-invocation-settings >=0.5.0`, which
8+
owns the *settings contract* — including the v2 document as the only accepted sealed
9+
`/invoke` shape, independent sealed-field resolution, and what an absent field is allowed to
10+
mean. That split is deliberate: the
11+
absence rule is a security decision and belongs next to the crypto it governs, while request
12+
handling and route registration belong here, where a web framework is already a dependency.
13+
Nothing about the sealed-settings wire format is decided in this repository.
14+
* **The wrapper consumes the ratified `invocation_context` contract locally.**
15+
The field model, reserved key, supported versions, and dimension allow-list are generated from
16+
`https://schemas.u10d.dev/invocation-context/v1.json`. The handwritten
17+
`unstructured_platform_plugins.invocation_context` adapter retains transport error mapping and
18+
the equal-length batch invariant that JSON Schema cannot express. The ratified
19+
`https://schemas.u10d.dev/errors/audience/v1.json` vocabulary also replaces the redundant
20+
top-level `blame` response field: a legacy `UserError` now carries a complete `plugin_error`
21+
metadata object with `audience=user`.
22+
* **Every wrapped app installs it at construction.** The reserved `invocation_settings` /
23+
`invocation_context` fields are handled outside the generated handler schema, the opaque
24+
settings payload is delegated to `utic-invocation-settings`, and only the final resolved mapping
25+
is exposed through `current_invocation_settings()` / `current_invocation_context()`.
26+
An absent field preserves the existing fallback behaviour; under
27+
`FF_INVOCATION_SETTINGS` missing or plaintext settings fail closed.
28+
Repeated installation is safe: the dependency installs once and the last `/metadata`
29+
registration wins.
30+
* **Extraction is a route dependency.** `bind_invocation_envelope` reads the body the framework
31+
buffered and parsed (`request.json()` is Starlette-cached), so the `/invoke` body is held and
32+
decoded exactly once per request. `install_invocation_envelope` contributes that path-aware
33+
dependency through the router's public dependency list before `/invoke` is registered; no
34+
private FastAPI dependency graph is mutated. It also registers the failure response shape and
35+
can optionally install `InvokeBodyLimitMiddleware`, a streaming byte counter that answers 413
36+
over a host-selected cap without buffering. The cap is disabled by default so a wrapper upgrade
37+
cannot impose an unvalidated fleet-wide request limit. Async-generator plugins explicitly
38+
re-enter the captured request binding inside response iteration, so streaming stays correct
39+
independently of FastAPI's yield-dependency cleanup timing.
40+
* **Sealed settings consumption remains opt-in.** Pass
41+
`invoke_with_sealed_dag_node_settings_v2=True` to `wrap_in_fastapi` / `generate_fast_api` (or
42+
`--sealed-dag-node-settings-v2` on the CLI) only for a plugin that consumes per-invoke settings;
43+
it advertises that the application accepts and acts on the versioned v2 document.
44+
Transport support alone continues to advertise only `invocation_settings` and
45+
`invocation_context`. A
46+
plugin that serves a custom `/metadata` payload must register it via `add_metadata_route` (which
47+
replaces the wrapper's route) — a plain `@app.get("/metadata")` added after construction is
48+
shadowed by the wrapper's earlier registration.
49+
* **Resolution runs off the event loop.** Resolution may perform blocking cryptography for
50+
independently sealed fields and this dependency fronts every invoke on the pod, so it is
51+
dispatched with `asyncio.to_thread` rather than blocking the loop.
52+
* **Failures map through the library's blame taxonomy**, not a flat 500: only a caller-fixable
53+
fault answers 422. Sealing drift, an envelope addressed to another recipient and a broken local
54+
mount are all 5xx, which keeps the controller's blame classification off the customer. Responses
55+
carry the error's class name and never its message, which can embed request-controlled values.
56+
* **Python floor is now 3.11** (required by `utic-invocation-settings`).
57+
158
## 0.0.46
259

360
* **Carry preflight failure categories through standard `/precheck` responses.**

pyproject.toml

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
[project]
22
name = "unstructured_platform_plugins"
33
description = "Wrapper to convert arbitrary code into a uvicorn/fastapi implementation for Unstructured Platform"
4-
requires-python = ">=3.10"
4+
requires-python = ">=3.11"
55
classifiers = [
66
"Development Status :: 4 - Beta",
77
"Intended Audience :: Developers",
@@ -10,7 +10,6 @@ classifiers = [
1010
"License :: OSI Approved :: Apache Software License",
1111
"Operating System :: OS Independent",
1212
"Programming Language :: Python :: 3",
13-
"Programming Language :: Python :: 3.10",
1413
"Programming Language :: Python :: 3.11",
1514
"Programming Language :: Python :: 3.12",
1615
"Programming Language :: Python :: 3.13",
@@ -24,6 +23,7 @@ dependencies = [
2423
"fastapi",
2524
"click",
2625
"unstructured-ingest",
26+
"utic-invocation-settings>=0.5.0,<1.0.0",
2727
"opentelemetry-instrumentation-fastapi",
2828
"opentelemetry-exporter-otlp-proto-grpc",
2929
"dataclasses-json"
Lines changed: 112 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,112 @@
1+
#!/usr/bin/env python3
2+
"""Generate local Python bindings from the ratified invocation schema IDs."""
3+
4+
from __future__ import annotations
5+
6+
import argparse
7+
import hashlib
8+
import json
9+
import subprocess
10+
import sys
11+
from pathlib import Path
12+
from typing import Any
13+
14+
ROOT = Path(__file__).resolve().parents[1]
15+
OUTPUT_DIR = ROOT / "unstructured_platform_plugins" / "generated"
16+
API_ROOT = "repos/Unstructured-IO/schemas-experimental/contents"
17+
18+
CONTRACTS = {
19+
"invocation_context_v1.py": {
20+
"schema": f"{API_ROOT}/schemas/ratified-types/invocation-context/v1.json?ref=main",
21+
"typeviz": f"{API_ROOT}/web/public/typeviz/invocation-context-v1.json?ref=main",
22+
"schema_id": "https://schemas.u10d.dev/invocation-context/v1.json",
23+
"root_type": "InvocationContext",
24+
},
25+
"error_audience_v1.py": {
26+
"schema": f"{API_ROOT}/schemas/ratified-types/errors/audience/v1.json?ref=main",
27+
"typeviz": f"{API_ROOT}/web/public/typeviz/errors-audience-v1.json?ref=main",
28+
"schema_id": "https://schemas.u10d.dev/errors/audience/v1.json",
29+
"root_type": "ErrorAudience",
30+
},
31+
}
32+
33+
34+
def _load_json(api_path: str) -> dict[str, Any]:
35+
completed = subprocess.run(
36+
["gh", "api", "-H", "Accept: application/vnd.github.raw+json", api_path],
37+
check=True,
38+
capture_output=True,
39+
text=True,
40+
)
41+
return json.loads(completed.stdout)
42+
43+
44+
def _python_binding(sidecar: dict[str, Any]) -> str:
45+
for binding in sidecar["bindings"]:
46+
if binding["key"] == "python":
47+
return binding["code"].rstrip() + "\n"
48+
raise ValueError("typeviz sidecar has no Python binding")
49+
50+
51+
def _render(spec: dict[str, str]) -> str:
52+
schema = _load_json(spec["schema"])
53+
sidecar = _load_json(spec["typeviz"])
54+
if schema["$id"] != spec["schema_id"]:
55+
raise ValueError(f"unexpected schema id: {schema['$id']}")
56+
if sidecar["root_type_name"] != spec["root_type"]:
57+
raise ValueError(f"unexpected root type: {sidecar['root_type_name']}")
58+
59+
canonical = json.dumps(schema, sort_keys=True, separators=(",", ":")).encode()
60+
schema_hash = hashlib.sha256(canonical).hexdigest()
61+
generated = _python_binding(sidecar)
62+
if spec["root_type"] == "InvocationContext":
63+
generated = generated.replace(
64+
"from typing import Any, Literal", "from typing import Literal"
65+
)
66+
else:
67+
generated = generated.replace("\nfrom pydantic import BaseModel, Field\n", "")
68+
provenance = (
69+
"\n# Generation provenance used by drift tests and reviewers.\n"
70+
f"SCHEMA_ID = {schema['$id']!r}\n"
71+
f"SCHEMA_SHA256 = {schema_hash!r}\n"
72+
)
73+
if spec["root_type"] == "InvocationContext":
74+
policy = schema["x-unstructured-version-policy"]
75+
provenance += (
76+
f"RESERVED_CONTEXT_KEY = {schema['x-unstructured-reserved-key']!r}\n"
77+
f"SUPPORTED_CONTEXT_VERSIONS = frozenset({policy['supported']!r})\n"
78+
f"DIMENSION_FIELDS = {tuple(schema['x-unstructured-dimension-fields'])!r}\n"
79+
)
80+
return (
81+
"# Generated by scripts/generate_invocation_contracts.py; do not edit by hand.\n"
82+
"# ruff: noqa: E501\n"
83+
f"# Source: {spec['schema_id']}\n"
84+
+ generated
85+
+ provenance
86+
)
87+
88+
89+
def main() -> int:
90+
parser = argparse.ArgumentParser()
91+
parser.add_argument("--check", action="store_true")
92+
args = parser.parse_args()
93+
94+
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
95+
stale: list[str] = []
96+
for filename, spec in CONTRACTS.items():
97+
output = OUTPUT_DIR / filename
98+
rendered = _render(spec)
99+
if args.check:
100+
if not output.exists() or output.read_text() != rendered:
101+
stale.append(str(output.relative_to(ROOT)))
102+
else:
103+
output.write_text(rendered)
104+
105+
if stale:
106+
print("stale generated invocation contracts:", *stale, sep="\n ", file=sys.stderr)
107+
return 1
108+
return 0
109+
110+
111+
if __name__ == "__main__":
112+
raise SystemExit(main())

test/api/test_api.py

Lines changed: 88 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,14 +18,24 @@
1818
UsageData,
1919
wrap_in_fastapi,
2020
)
21+
from unstructured_platform_plugins.generated.error_audience_v1 import ErrorAudience
2122
from unstructured_platform_plugins.schema.filedata_meta import FileDataMeta
2223

2324

25+
class PluginErrorMetadata(BaseModel):
26+
error_type: str
27+
error_reason: str
28+
dependency: Optional[str] = None
29+
audience: Optional[ErrorAudience] = None
30+
retryable: bool = False
31+
32+
2433
class InvokeResponse(BaseModel):
2534
usage: list[UsageData]
2635
status_code: int
2736
filedata_meta: FileDataMeta
2837
status_code_text: Optional[str] = None
38+
plugin_error: Optional[PluginErrorMetadata] = None
2939
output: Optional[Any] = None
3040
file_data: Optional[Union[FileData, BatchFileData]] = None
3141

@@ -222,6 +232,67 @@ def test_http_exception_handling(file_data):
222232
assert invoke_response.status_code_text == "Not found"
223233

224234

235+
@pytest.mark.parametrize(
236+
"file_data", mock_file_data, ids=[type(fd).__name__ for fd in mock_file_data]
237+
)
238+
def test_user_error_declares_canonical_user_audience(file_data):
239+
"""Only the UserError family declares a user-actionable plugin error."""
240+
from test.assets.exception_status_code import function_raises_user_error as test_fn
241+
242+
client = TestClient(wrap_in_fastapi(func=test_fn, plugin_id="mock_plugin"))
243+
244+
resp = client.post("/invoke", json={"file_data": file_data.model_dump()})
245+
invoke_response = InvokeResponse.model_validate(resp.json())
246+
247+
assert invoke_response.status_code >= 400
248+
assert invoke_response.plugin_error is not None
249+
assert invoke_response.plugin_error.audience is ErrorAudience.USER
250+
assert invoke_response.plugin_error.error_type == "configuration"
251+
assert invoke_response.plugin_error.error_reason == "invalid_input"
252+
assert invoke_response.plugin_error.retryable is False
253+
254+
255+
@pytest.mark.parametrize(
256+
"file_data", mock_file_data, ids=[type(fd).__name__ for fd in mock_file_data]
257+
)
258+
def test_non_user_failures_declare_no_plugin_error(file_data):
259+
"""Anything undeclared is not the customer's: an orchestrator must not infer customer fault
260+
from the status code, which also carries transport semantics."""
261+
from test.assets.exception_status_code import function_raises_provider_error as test_fn
262+
263+
client = TestClient(wrap_in_fastapi(func=test_fn, plugin_id="mock_plugin"))
264+
265+
resp = client.post("/invoke", json={"file_data": file_data.model_dump()})
266+
invoke_response = InvokeResponse.model_validate(resp.json())
267+
268+
assert invoke_response.plugin_error is None
269+
270+
271+
def test_streaming_user_error_declares_user_audience():
272+
"""The streaming error envelope carries the same audience as the non-streaming path."""
273+
from test.assets.exception_status_code import (
274+
async_gen_function_raises_user_error_mid_stream as test_fn,
275+
)
276+
277+
client = TestClient(wrap_in_fastapi(func=test_fn, plugin_id="mock_plugin"))
278+
279+
resp = client.post("/invoke", json={"file_data": mock_file_data[0].model_dump()})
280+
281+
assert resp.status_code == 200
282+
assert resp.headers["content-type"] == "application/x-ndjson"
283+
284+
import json
285+
286+
lines = resp.content.decode().strip().split("\n")
287+
assert len(lines) == 2 # One yielded item, then the error envelope
288+
289+
assert InvokeResponse.model_validate(json.loads(lines[0])).plugin_error is None
290+
error_response = InvokeResponse.model_validate(json.loads(lines[1]))
291+
assert error_response.status_code >= 400
292+
assert error_response.plugin_error is not None
293+
assert error_response.plugin_error.audience is ErrorAudience.USER
294+
295+
225296
@pytest.mark.parametrize(
226297
"file_data", mock_file_data, ids=[type(fd).__name__ for fd in mock_file_data]
227298
)
@@ -608,9 +679,26 @@ def test_precheck_reports_failure_category_from_raised_error():
608679
body = resp.json()
609680
assert body["status_code"] == 403
610681
assert body["failure_category"] == "AUTH_PERMISSION_DENIED"
682+
# A plain exception is not the customer's to fix.
683+
assert body["plugin_error"] is None
611684
assert "credential rejected" in body["status_code_text"]
612685

613686

687+
def test_precheck_declares_user_audience_like_invoke_does():
688+
from unstructured_ingest.error import UserError
689+
690+
def user_fault_precheck() -> None:
691+
raise UserError("bad credentials")
692+
693+
client = TestClient(
694+
wrap_in_fastapi(func=_no_params, plugin_id="mock_plugin", precheck_func=user_fault_precheck)
695+
)
696+
697+
body = client.get("/precheck").json()
698+
699+
assert body["plugin_error"]["audience"] == "user"
700+
701+
614702
def test_precheck_success_has_no_failure_category():
615703
client = TestClient(
616704
wrap_in_fastapi(func=_no_params, plugin_id="mock_plugin", precheck_func=_passing_precheck)

0 commit comments

Comments
 (0)