feat(etl-uvicorn): install invocation-settings handling (0.1.0)

Author: CyMule

CHANGELOG.md

@@ -1,3 +1,27 @@
+## 0.1.0
+
+* **The wrapper now installs the invocation-settings envelope handling itself.** Every wrapped app
+  gets the `utic-invocation-settings` ASGI middleware and a `/metadata` route at construction: the
+  reserved `invocation_settings` / `invocation_context` fields are handled outside the generated
+  handler schema, a sealed `dag_node_settings` member is decrypted with the configured private
+  key, and the resolved values are exposed request-scoped through
+  `current_invocation_settings()` / `current_invocation_context()`. Missing fields preserve the
+  existing fallback behavior; when `FF_REQUIRE_INVOKE_WITH_SEALED_DAG_NODE_SETTINGS` is enabled,
+  missing or plaintext settings fail closed. Repeated installation is safe: the middleware
+  installs once and the last `/metadata` registration wins.
+* **New opt-in `invoke_with_sealed_dag_node_settings` advertisement.** Pass
+  `invoke_with_sealed_dag_node_settings=True` to `wrap_in_fastapi` / `generate_fast_api` (or
+  `--sealed-dag-node-settings` on the CLI) only for a plugin that consumes per-invoke settings;
+  it advertises that the application accepts and consumes sealed per-invocation settings.
+  A plugin that serves a custom `/metadata` payload must register it via `add_metadata_route`
+  (which replaces the wrapper's route) — a plain `@app.get("/metadata")` added after construction
+  is shadowed by the wrapper's earlier registration.
+* **Sync plugin functions now observe request-scoped context.** `invoke_func` copies the current
+  context into the executor thread; previously `run_in_executor` dropped contextvars, so a sync
+  function reading a request-scoped binding (such as `current_invocation_settings()`) would see
+  it as absent and could take an unintended fallback path.
+* **Python floor is now 3.11** (required by `utic-invocation-settings`).
+
 ## 0.0.45
 
 * **`/invoke` no longer demands a body from a plugin whose parameters are all optional.** A pydantic

pyproject.toml

@@ -1,7 +1,7 @@
 [project]
 name = "unstructured_platform_plugins"
 description = "Wrapper to convert arbitrary code into a uvicorn/fastapi implementation for Unstructured Platform"
-requires-python = ">=3.10"
+requires-python = ">=3.11"
 classifiers = [
     "Development Status :: 4 - Beta",
     "Intended Audience :: Developers",
@@ -10,7 +10,6 @@ classifiers = [
     "License :: OSI Approved :: Apache Software License",
     "Operating System :: OS Independent",
     "Programming Language :: Python :: 3",
-    "Programming Language :: Python :: 3.10",
     "Programming Language :: Python :: 3.11",
     "Programming Language :: Python :: 3.12",
     "Programming Language :: Python :: 3.13",
@@ -24,6 +23,7 @@ dependencies = [
     "fastapi",
     "click",
     "unstructured-ingest",
+    "utic-invocation-settings>=0.3.0,<1.0.0",
     "opentelemetry-instrumentation-fastapi",
     "opentelemetry-exporter-otlp-proto-grpc",
     "dataclasses-json"

test/api/test_invocation_settings.py

@@ -0,0 +1,103 @@
+"""The wrapper-installed invocation-settings surface: /metadata and reserved-field binding."""
+
+from typing import Optional
+
+from fastapi.testclient import TestClient
+from pydantic import BaseModel
+from utic_invocation_settings import current_invocation_settings
+
+from unstructured_platform_plugins.etl_uvicorn.api_generator import wrap_in_fastapi
+
+
+class _Echo(BaseModel):
+    content: str
+    settings: Optional[dict]
+
+
+def _echo_settings(content: str) -> _Echo:
+    return _Echo(content=content, settings=current_invocation_settings())
+
+
+def test_metadata_route_is_registered_with_default_capabilities():
+    client = TestClient(wrap_in_fastapi(func=_echo_settings, plugin_id="mock_plugin"))
+
+    resp = client.get("/metadata")
+
+    assert resp.status_code == 200
+    payload = resp.json()
+    assert payload["identifier"] == "mock_plugin"
+    assert payload["capabilities"] == ["invocation_settings", "invocation_context"]
+
+
+def test_sealed_capability_is_opt_in():
+    client = TestClient(
+        wrap_in_fastapi(
+            func=_echo_settings,
+            plugin_id="mock_plugin",
+            invoke_with_sealed_dag_node_settings=True,
+        )
+    )
+
+    payload = client.get("/metadata").json()
+
+    assert "invoke_with_sealed_dag_node_settings" in payload["capabilities"]
+
+
+def test_reserved_settings_field_binds_without_appearing_in_schema():
+    app = wrap_in_fastapi(func=_echo_settings, plugin_id="mock_plugin")
+    client = TestClient(app)
+
+    resp = client.post(
+        "/invoke",
+        json={"content": "hello", "invocation_settings": {"model": "m"}},
+    )
+
+    assert resp.status_code == 200
+    output = resp.json()["output"]
+    assert output == {"content": "hello", "settings": {"model": "m"}}
+    # The wrapper does not add the reserved field to the generated handler input model.
+    openapi = app.openapi()
+    request_schema = openapi["paths"]["/invoke"]["post"]["requestBody"]["content"][
+        "application/json"
+    ]["schema"]
+    schema_name = request_schema["$ref"].rsplit("/", 1)[-1]
+    properties = openapi["components"]["schemas"][schema_name]["properties"]
+    assert "invocation_settings" not in properties
+
+
+def test_sync_function_sees_bound_settings_across_the_executor():
+    # Sync functions run in an executor thread; the context must be copied there or the
+    # request-scoped binding would read as absent.
+    def sync_echo(content: str) -> _Echo:
+        return _Echo(content=content, settings=current_invocation_settings())
+
+    client = TestClient(wrap_in_fastapi(func=sync_echo, plugin_id="mock_plugin"))
+
+    resp = client.post(
+        "/invoke", json={"content": "hello", "invocation_settings": {"model": "m"}}
+    )
+
+    assert resp.json()["output"]["settings"] == {"model": "m"}
+
+
+async def _async_echo(content: str) -> _Echo:
+    return _Echo(content=content, settings=current_invocation_settings())
+
+
+def test_async_function_sees_bound_settings():
+    client = TestClient(wrap_in_fastapi(func=_async_echo, plugin_id="mock_plugin"))
+
+    resp = client.post(
+        "/invoke", json={"content": "hello", "invocation_settings": {"model": "m"}}
+    )
+
+    assert resp.json()["output"]["settings"] == {"model": "m"}
+
+
+def test_absent_reserved_fields_bind_none():
+    client = TestClient(wrap_in_fastapi(func=_echo_settings, plugin_id="mock_plugin"))
+
+    resp = client.post("/invoke", json={"content": "hello"})
+
+    assert resp.status_code == 200
+    assert resp.json()["output"] == {"content": "hello", "settings": None}

unstructured_platform_plugins/__version__.py

@@ -1 +1 @@
-__version__ = "0.0.45"  # pragma: no cover
+__version__ = "0.1.0"  # pragma: no cover

unstructured_platform_plugins/etl_uvicorn/api_generator.py

@@ -1,4 +1,5 @@
 import asyncio
+import contextvars
 import hashlib
 import inspect
 import json
@@ -14,6 +15,7 @@
 from typing_extensions import deprecated
 from unstructured_ingest.data_types.file_data import BatchFileData, FileData, file_data_from_dict
 from unstructured_ingest.error import UnstructuredIngestError
+from utic_invocation_settings import add_metadata_route, install_invocation_envelope
 from uvicorn.config import LOG_LEVELS
 from uvicorn.importer import import_from_string
 
@@ -67,7 +69,13 @@ async def invoke_func(func: Callable, kwargs: Optional[dict[str, Any]] = None) -
     if inspect.iscoroutinefunction(func):
         return await func(**kwargs)
     else:
-        return await asyncio.get_event_loop().run_in_executor(None, partial(func, **kwargs))
+        # run_in_executor does not propagate contextvars, so without copying the context a sync
+        # plugin would observe request-scoped bindings (current_invocation_settings and friends)
+        # as absent and could take an unintended fallback path.
+        ctx = contextvars.copy_context()
+        return await asyncio.get_event_loop().run_in_executor(
+            None, ctx.run, partial(func, **kwargs)
+        )
 
 
 def check_precheck_func(precheck_func: Callable):
@@ -117,9 +125,15 @@ def wrap_in_fastapi(
     func: Callable,
     plugin_id: str,
     precheck_func: Optional[Callable] = None,
+    invoke_with_sealed_dag_node_settings: bool = False,
 ) -> FastAPI:
     try:
-        return _wrap_in_fastapi(func=func, plugin_id=plugin_id, precheck_func=precheck_func)
+        return _wrap_in_fastapi(
+            func=func,
+            plugin_id=plugin_id,
+            precheck_func=precheck_func,
+            invoke_with_sealed_dag_node_settings=invoke_with_sealed_dag_node_settings,
+        )
     except Exception as e:
         logger.error(f"failed to wrap function in FastAPI: {e}", exc_info=True)
         raise EtlApiException(e) from e
@@ -129,6 +143,7 @@ def _wrap_in_fastapi(
     func: Callable,
     plugin_id: str,
     precheck_func: Optional[Callable] = None,
+    invoke_with_sealed_dag_node_settings: bool = False,
 ) -> FastAPI:
     if precheck_func is not None:
         check_precheck_func(precheck_func=precheck_func)
@@ -347,6 +362,19 @@ async def get_id() -> str:
     except TypeError as e:
         raise TypeError(f"failed to validate function schema: {e}") from e
 
+    # The middleware handles the reserved /invoke fields (invocation_settings and
+    # invocation_context) outside the generated handler schema. It resolves sealed settings with
+    # the configured private key and exposes both values through request-scoped accessors. The
+    # sealed-settings capability remains opt-in because it asserts that the wrapped function
+    # consumes current_invocation_settings(), not merely that the host can resolve it. Repeated
+    # installation is safe: the middleware installs once and the last /metadata registration wins.
+    add_metadata_route(
+        fastapi_app,
+        identifier=plugin_id,
+        invoke_with_sealed_dag_node_settings=invoke_with_sealed_dag_node_settings,
+    )
+    install_invocation_envelope(fastapi_app)
+
     FastAPIInstrumentor.instrument_app(
         fastapi_app, tracer_provider=get_trace_provider(), meter_provider=get_metric_provider()
     )
@@ -361,6 +389,7 @@ def generate_fast_api(
     id_method: Optional[str] = None,
     precheck_str: Optional[str] = None,
     precheck_method: Optional[str] = None,
+    invoke_with_sealed_dag_node_settings: bool = False,
 ) -> FastAPI:
     instance = import_from_string(app)
     func = get_func(instance, method_name)
@@ -379,4 +408,9 @@ def generate_fast_api(
     elif precheck_method:
         precheck_func = get_func(instance, precheck_method)
 
-    return wrap_in_fastapi(func=func, plugin_id=plugin_id, precheck_func=precheck_func)
+    return wrap_in_fastapi(
+        func=func,
+        plugin_id=plugin_id,
+        precheck_func=precheck_func,
+        invoke_with_sealed_dag_node_settings=invoke_with_sealed_dag_node_settings,
+    )

unstructured_platform_plugins/etl_uvicorn/main.py

@@ -56,6 +56,7 @@ def api_wrapper(
         plugin_id_method: Optional[str] = None,
         precheck_app: Optional[str] = None,
         precheck_app_method: Optional[str] = None,
+        sealed_dag_node_settings: bool = False,
         **kwargs,
     ):
         # Make sure logging is configured before the call to run() so any setup has the same format
@@ -73,6 +74,7 @@ def api_wrapper(
             id_method=plugin_id_method,
             precheck_str=precheck_app,
             precheck_method=precheck_app_method,
+            invoke_with_sealed_dag_node_settings=sealed_dag_node_settings,
         )
         # Explicitly map values that are manipulated in the original
         # call to run(), preventing **kwargs reference
@@ -130,6 +132,14 @@ def api_wrapper(
                 "If precheck-app not provided, assumes method "
                 "lives on main class passes in.",
             ),
+            click.Option(
+                ["--sealed-dag-node-settings"],
+                is_flag=True,
+                default=False,
+                help="Advertise the invoke_with_sealed_dag_node_settings capability on "
+                "/metadata. Set only for a plugin that consumes per-invoke settings "
+                "through current_invocation_settings().",
+            ),
         ]
     )
     return cmd