Extensible providers

Author: LucasAlvesSoares

src/sap_cloud_sdk/core/bootstrap.py

@@ -1,16 +1,17 @@
 """Top-level bootstrap() entry point for the SAP Cloud SDK."""
 
-from typing import Any
+from typing import Any, List, Optional
 
 from sap_cloud_sdk.core.runtime_context._protocol import ContextProvider
 
 
-def bootstrap(app: Any, provider: ContextProvider) -> None:
+def bootstrap(app: Any, providers: Optional[List[ContextProvider]] = None) -> None:
     """Wire the SDK runtime context into your application framework.
 
     Call this once at application startup. The SDK will attach a middleware
     to *app* that populates :class:`~sap_cloud_sdk.core.runtime_context.RequestContext`
-    on every inbound request using *provider*.
+    on every inbound request by running all *providers* against it and merging
+    the results.
 
     After bootstrapping, any SDK module (auditlog, telemetry, etc.) can call
     :func:`~sap_cloud_sdk.core.runtime_context.get_context` to read
@@ -20,36 +21,36 @@ def bootstrap(app: Any, provider: ContextProvider) -> None:
       - Starlette / FastAPI
 
     Args:
-        app:      The application instance to attach the middleware to.
-        provider: A :class:`~sap_cloud_sdk.core.runtime_context.ContextProvider`
-                  that knows how to extract context from a request in this framework.
+        app:       The application instance to attach the middleware to.
+        providers: One or more :class:`~sap_cloud_sdk.core.runtime_context.ContextProvider`
+                   instances. Defaults to ``[IASContextProvider()]``.
 
     Raises:
         TypeError: If *app* is not a recognised framework application type.
 
     Example::
 
-        from starlette.applications import Starlette
         from sap_cloud_sdk import bootstrap
-        from sap_cloud_sdk.core.runtime_context import IASContextProvider
 
-        app = Starlette(...)
-        bootstrap(app, provider=IASContextProvider())
+        bootstrap(app)  # IASContextProvider by default
+
+        # multiple providers:
+        bootstrap(app, providers=[IASContextProvider(), MyCustomProvider()])
     """
-    _attach(app, provider)
+    if not providers:
+        from sap_cloud_sdk.core.runtime_context import IASContextProvider
+        providers = [IASContextProvider()]
+    _attach(app, providers)
 
 
-def _attach(app: Any, provider: ContextProvider) -> None:
+def _attach(app: Any, providers: List[ContextProvider]) -> None:
     """Detect framework and register the appropriate context middleware."""
-    # Starlette / FastAPI — both expose add_middleware and share the same base
     try:
         from starlette.applications import Starlette
-        from sap_cloud_sdk.core.runtime_context.starlette import (
-            StarletteContextMiddleware,
-        )
+        from sap_cloud_sdk.core.runtime_context.starlette import StarletteContextMiddleware
 
         if isinstance(app, Starlette):
-            app.add_middleware(StarletteContextMiddleware, provider=provider)
+            app.add_middleware(StarletteContextMiddleware, providers=providers)
             return
     except ImportError:
         pass

src/sap_cloud_sdk/core/runtime_context/__init__.py

@@ -34,13 +34,15 @@
     sdk_context,
     set_context,
 )
+from sap_cloud_sdk.core.runtime_context._envelope import RequestEnvelope
 from sap_cloud_sdk.core.runtime_context._protocol import ContextProvider
 from sap_cloud_sdk.core.runtime_context._providers import IASContextProvider
 
 __all__ = [
     "ContextProvider",
     "IASContextProvider",
     "RequestContext",
+    "RequestEnvelope",
     "async_sdk_context",
     "get_context",
     "sdk_context",

src/sap_cloud_sdk/core/runtime_context/_envelope.py

@@ -0,0 +1,24 @@
+"""Framework-agnostic request envelope passed to ContextProviders."""
+
+from dataclasses import dataclass, field
+from typing import Any, Dict, Optional
+
+
+@dataclass
+class RequestEnvelope:
+    """Normalized view of an inbound request, independent of framework.
+
+    Framework middlewares build this from their native request object.
+    :class:`~sap_cloud_sdk.core.runtime_context.ContextProvider` implementations
+    read from it — they never touch framework-specific types.
+
+    Attributes:
+        headers:  Case-insensitive HTTP headers (or equivalent for gRPC/etc.).
+        body:     Raw request body bytes. ``None`` if not extracted.
+        metadata: Catch-all for framework extras (query params, gRPC metadata,
+                  connection info, etc.). Reserved for future providers.
+    """
+
+    headers: Dict[str, str] = field(default_factory=dict)
+    body: Optional[bytes] = field(default=None)
+    metadata: Dict[str, Any] = field(default_factory=dict)

src/sap_cloud_sdk/core/runtime_context/_protocol.py

@@ -1,23 +1,29 @@
 """ContextProvider protocol — the pluggable extraction interface."""
 
-from typing import Any, Protocol, runtime_checkable
+from typing import Protocol, runtime_checkable
 
 from sap_cloud_sdk.core.runtime_context._context import RequestContext
+from sap_cloud_sdk.core.runtime_context._envelope import RequestEnvelope
 
 
 @runtime_checkable
 class ContextProvider(Protocol):
-    """Extract a :class:`RequestContext` from a framework-specific request object.
+    """Extract a :class:`RequestContext` from a :class:`RequestEnvelope`.
 
     Implement this protocol to teach the SDK how to read caller-identity
-    information from a new framework (Flask, gRPC, raw WSGI, etc.).
+    information from a specific auth provider (IAS, XSUAA, etc.).
+
+    The envelope is framework-agnostic — providers never touch Starlette,
+    Flask, or gRPC types directly. The framework middleware is responsible
+    for building the envelope.
 
     Example::
 
         class MyProvider(ContextProvider):
-            def extract(self, request: MyRequest) -> RequestContext:
-                return RequestContext(tenant_id=request.tenant, user_id=request.user)
+            def extract(self, envelope: RequestEnvelope) -> RequestContext:
+                token = envelope.headers.get("x-my-token", "")
+                return RequestContext(tenant_id=decode(token).tenant)
     """
 
-    def extract(self, request: Any) -> RequestContext:  # pragma: no cover
+    def extract(self, envelope: RequestEnvelope) -> RequestContext:  # pragma: no cover
         ...

src/sap_cloud_sdk/core/runtime_context/_providers.py

@@ -1,31 +1,29 @@
 """Built-in ContextProvider implementations."""
 
-from typing import Any
-
 from sap_cloud_sdk.core.runtime_context._context import RequestContext
+from sap_cloud_sdk.core.runtime_context._envelope import RequestEnvelope
 from sap_cloud_sdk.core.runtime_context._protocol import ContextProvider
 from sap_cloud_sdk.ias import parse_token
 
 
 class IASContextProvider(ContextProvider):
     """Extracts tenant/user context from an IAS JWT ``Authorization`` header.
 
-    Works with any request object that exposes headers via ``request.headers``
-    (Starlette, FastAPI, etc.).
+    Reads from a :class:`RequestEnvelope` — works with any framework.
 
     The following claims are mapped:
-      - ``app_tid``   → :attr:`RequestContext.tenant_id`
-      - ``user_uuid`` → :attr:`RequestContext.user_id`
-      - ``x-sap-origin`` header → :attr:`RequestContext.trigger_type`
+
+      - ``app_tid``        → :attr:`~.RequestContext.tenant_id`
+      - ``user_uuid``      → :attr:`~.RequestContext.user_id`
+      - ``x-sap-origin``  → :attr:`~.RequestContext.trigger_type`
 
     Full :class:`~sap_cloud_sdk.ias.IASClaims` are stored under
     ``RequestContext.extras["ias.claims"]`` for modules that need them.
     """
 
-    def extract(self, request: Any) -> RequestContext:
-        headers = getattr(request, "headers", {})
-        auth = headers.get("authorization", "") or headers.get("Authorization", "")
-        origin = headers.get("x-sap-origin") or headers.get("X-Sap-Origin")
+    def extract(self, envelope: RequestEnvelope) -> RequestContext:
+        auth = envelope.headers.get("authorization", "")
+        origin = envelope.headers.get("x-sap-origin")
 
         claims = None
         if auth:

src/sap_cloud_sdk/core/runtime_context/starlette.py

@@ -1,8 +1,9 @@
 """Starlette/FastAPI context middleware."""
 
-from typing import Any
+from typing import Any, List, Optional
 
-from sap_cloud_sdk.core.runtime_context._context import async_sdk_context
+from sap_cloud_sdk.core.runtime_context._context import RequestContext, async_sdk_context
+from sap_cloud_sdk.core.runtime_context._envelope import RequestEnvelope
 from sap_cloud_sdk.core.runtime_context._protocol import ContextProvider
 
 try:
@@ -16,34 +17,48 @@
     ) from exc
 
 
+def _merge(contexts: List[RequestContext]) -> RequestContext:
+    """Merge multiple RequestContexts — first non-None value wins per field."""
+    merged = RequestContext()
+    for ctx in contexts:
+        if merged.tenant_id is None:
+            merged.tenant_id = ctx.tenant_id
+        if merged.user_id is None:
+            merged.user_id = ctx.user_id
+        if merged.trigger_type is None:
+            merged.trigger_type = ctx.trigger_type
+        merged.extras.update(ctx.extras)
+    return merged
+
+
 class StarletteContextMiddleware(BaseHTTPMiddleware):
     """Starlette/FastAPI middleware that populates the SDK runtime context.
 
-    Runs *provider*.extract() on every inbound request and makes the result
-    available via :func:`~sap_cloud_sdk.core.runtime_context.get_context`
-    for the duration of that request.
+    Builds a :class:`~sap_cloud_sdk.core.runtime_context.RequestEnvelope` from
+    each inbound request, runs all *providers* against it, and merges the results
+    into a single :class:`~sap_cloud_sdk.core.runtime_context.RequestContext`
+    available via :func:`~sap_cloud_sdk.core.runtime_context.get_context` for
+    the duration of that request.
+
+    First non-None value wins per field when merging. Extras are union-merged
+    (later providers can add keys, but not overwrite earlier ones).
 
     Usage::
 
-        from starlette.applications import Starlette
         from sap_cloud_sdk import bootstrap
 
-        app = Starlette(...)
-        bootstrap(app)
-
-    Or manually::
-
-        from sap_cloud_sdk.core.runtime_context.starlette import StarletteContextMiddleware
-        from sap_cloud_sdk.core.runtime_context import IASContextProvider
+        bootstrap(app)  # IASContextProvider by default
 
-        app.add_middleware(StarletteContextMiddleware, provider=IASContextProvider())
+        # or with multiple providers:
+        bootstrap(app, providers=[IASContextProvider(), MyCustomProvider()])
     """
 
-    def __init__(self, app: Any, provider: ContextProvider) -> None:
+    def __init__(self, app: Any, providers: List[ContextProvider]) -> None:
         super().__init__(app)
-        self._provider = provider
+        self._providers = providers
 
     async def dispatch(self, request: Request, call_next: Any) -> Response:
-        ctx = self._provider.extract(request)
+        envelope = RequestEnvelope(headers=dict(request.headers))
+        ctx = _merge([p.extract(envelope) for p in self._providers])
         async with async_sdk_context(ctx):
             return await call_next(request)