From 7ddb90eb2764b6282b967720ab7cf4bf0d920d66 Mon Sep 17 00:00:00 2001 From: Nick Franck <46548427+CyMule@users.noreply.github.com> Date: Wed, 15 Jul 2026 15:49:14 -0400 Subject: [PATCH] feat: reserved invocation_settings envelope field on generated /invoke Every wrap_in_fastapi-generated /invoke now accepts an optional invocation_settings object as a reserved envelope field. It is kept out of the function's /schema (no namespace collision with data inputs) and delivered either as a kwarg when the wrapped function declares an invocation_settings parameter, or via the get_invocation_settings() ContextVar accessor otherwise (context propagated across the executor thread). The ContextVar is set even on the kwarg path so helper code deep in a plugin's call stack can always use the accessor. GET /capabilities advertises support so callers (e.g. the ETL controller) can gate injection without schema sniffing. The signature check for the kwarg path is hoisted to wrap time; requests pay a boolean, not reflection. This lets one standing plugin process serve work items with differing node settings (warm pools, worker fleets, per-invoke reconfiguration) instead of freezing settings from a boot-time file. 75 tests passed (5 new in test/test_invocation_settings.py). --- test/test_invocation_settings.py | 63 +++++++++++++++++++ .../etl_uvicorn/api_generator.py | 40 +++++++++++- .../etl_uvicorn/invocation_context.py | 20 ++++++ 3 files changed, 120 insertions(+), 3 deletions(-) create mode 100644 test/test_invocation_settings.py create mode 100644 unstructured_platform_plugins/etl_uvicorn/invocation_context.py diff --git a/test/test_invocation_settings.py b/test/test_invocation_settings.py new file mode 100644 index 0000000..6045cb3 --- /dev/null +++ b/test/test_invocation_settings.py @@ -0,0 +1,63 @@ +from typing import Any, Optional + +from fastapi.testclient import TestClient + +from unstructured_platform_plugins.etl_uvicorn.api_generator import wrap_in_fastapi +from unstructured_platform_plugins.etl_uvicorn.invocation_context import ( + get_invocation_settings, +) +from pydantic import BaseModel + + +class Out(BaseModel): + content: str + settings_seen: Optional[dict[str, Any]] = None + + +def fn_declares(content: str, invocation_settings: Optional[dict[str, Any]] = None) -> Out: + return Out(content=content, settings_seen=invocation_settings) + + +def fn_uses_context(content: str) -> Out: + return Out(content=content, settings_seen=get_invocation_settings()) + + +def test_declared_param_receives_settings(): + client = TestClient(wrap_in_fastapi(func=fn_declares, plugin_id="t1")) + resp = client.post( + "/invoke", + json={"content": "x", "invocation_settings": {"strategy": "fast"}}, + ) + assert resp.status_code == 200 + assert resp.json()["output"]["settings_seen"] == {"strategy": "fast"} + + +def test_context_accessor_receives_settings(): + client = TestClient(wrap_in_fastapi(func=fn_uses_context, plugin_id="t2")) + resp = client.post( + "/invoke", + json={"content": "x", "invocation_settings": {"k": 1}}, + ) + assert resp.status_code == 200 + assert resp.json()["output"]["settings_seen"] == {"k": 1} + + +def test_omitted_settings_is_none_and_context_resets(): + client = TestClient(wrap_in_fastapi(func=fn_uses_context, plugin_id="t3")) + with_settings = client.post("/invoke", json={"content": "x", "invocation_settings": {"k": 1}}) + without = client.post("/invoke", json={"content": "x"}) + assert with_settings.json()["output"]["settings_seen"] == {"k": 1} + assert without.json()["output"]["settings_seen"] is None + + +def test_capabilities_endpoint(): + client = TestClient(wrap_in_fastapi(func=fn_uses_context, plugin_id="t4")) + resp = client.get("/capabilities") + assert resp.status_code == 200 + assert resp.json()["invocation_settings"] is True + + +def test_schema_unpolluted_by_envelope_field(): + client = TestClient(wrap_in_fastapi(func=fn_uses_context, plugin_id="t5")) + schema = client.get("/schema").json() + assert "invocation_settings" not in schema["inputs"].get("properties", {}) diff --git a/unstructured_platform_plugins/etl_uvicorn/api_generator.py b/unstructured_platform_plugins/etl_uvicorn/api_generator.py index 06e0074..af01f79 100644 --- a/unstructured_platform_plugins/etl_uvicorn/api_generator.py +++ b/unstructured_platform_plugins/etl_uvicorn/api_generator.py @@ -1,4 +1,5 @@ import asyncio +import contextvars import hashlib import inspect import json @@ -18,6 +19,9 @@ from uvicorn.importer import import_from_string from unstructured_platform_plugins.etl_uvicorn.otel import get_metric_provider, get_trace_provider +from unstructured_platform_plugins.etl_uvicorn.invocation_context import ( + _invocation_settings_var, +) from unstructured_platform_plugins.etl_uvicorn.utils import ( get_func, get_input_schema, @@ -67,7 +71,12 @@ 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)) + # copy_context() so ContextVars (e.g. invocation_settings) reach the + # worker thread; run_in_executor does not propagate them by itself. + ctx = contextvars.copy_context() + return await asyncio.get_event_loop().run_in_executor( + None, partial(ctx.run, partial(func, **kwargs)) + ) def check_precheck_func(precheck_func: Callable): @@ -149,8 +158,22 @@ class InvokeResponse(BaseModel): output: Optional[response_type] = None message_channels: MessageChannels = Field(default_factory=MessageChannels) - input_schema = get_input_schema(func, omit=["usage", "filedata_meta", "message_channels"]) + input_schema = get_input_schema( + func, omit=["usage", "filedata_meta", "message_channels", "invocation_settings"] + ) input_schema_model = schema_to_base_model(input_schema) + # First-class per-invoke settings: every generated /invoke accepts an optional + # invocation_settings object, independent of the wrapped function's signature. + # Delivery: passed as a kwarg when the function declares the parameter, + # otherwise exposed through get_invocation_settings() for the call's duration. + input_schema_model = create_model( + "InvokeEnvelope", + __base__=input_schema_model, + invocation_settings=(Optional[dict[str, Any]], None), + ) + func_declares_invocation_settings = ( + "invocation_settings" in inspect.signature(func).parameters + ) logging.getLogger("etl_uvicorn.fastapi") @@ -261,6 +284,10 @@ async def run_job(request: input_schema_model) -> ResponseType: log_func_and_body(func=func, body=request.json()) # Create dictionary from pydantic model while preserving underlying types request_dict = {f: getattr(request, f) for f in request.model_fields} + invocation_settings = request_dict.pop("invocation_settings", None) + if func_declares_invocation_settings: + request_dict["invocation_settings"] = invocation_settings + settings_token = _invocation_settings_var.set(invocation_settings) # Make sure nested classes get instantiated correctly if "file_data" in request_dict: request_dict["file_data"] = file_data_from_dict( @@ -269,7 +296,10 @@ async def run_job(request: input_schema_model) -> ResponseType: map_inputs(func=func, raw_inputs=request_dict) if logger.level == LOG_LEVELS.get("trace", logging.NOTSET): logger.log(level=logger.level, msg=f"passing inputs to function: {request_dict}") - return await wrap_fn(func=func, kwargs=request_dict) + try: + return await wrap_fn(func=func, kwargs=request_dict) + finally: + _invocation_settings_var.reset(settings_token) else: @@ -315,6 +345,10 @@ async def run_precheck() -> InvokePrecheckResponse: async def get_id() -> str: return plugin_id + @fastapi_app.get("/capabilities") + async def get_capabilities() -> dict[str, Any]: + return {"invocation_settings": True} + # Run initial schema validation try: asyncio.run(get_schema()) diff --git a/unstructured_platform_plugins/etl_uvicorn/invocation_context.py b/unstructured_platform_plugins/etl_uvicorn/invocation_context.py new file mode 100644 index 0000000..b2a9ceb --- /dev/null +++ b/unstructured_platform_plugins/etl_uvicorn/invocation_context.py @@ -0,0 +1,20 @@ +"""Per-invocation context for first-class /invoke envelope fields. + +``invocation_settings`` is a reserved, always-accepted field of every generated +``/invoke`` request. Functions that declare an ``invocation_settings`` parameter +receive it as a kwarg; all other code (including plugins with hand-rolled apps +that adopt the same contract) can read it for the duration of the call via +:func:`get_invocation_settings`. +""" + +from contextvars import ContextVar +from typing import Any, Optional + +_invocation_settings_var: ContextVar[Optional[dict[str, Any]]] = ContextVar( + "invocation_settings", default=None +) + + +def get_invocation_settings() -> Optional[dict[str, Any]]: + """The invocation_settings of the in-flight /invoke call, if any.""" + return _invocation_settings_var.get()