Coverage for dataexcept/serverless_context.py: 92%
80 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-27 14:44 +0000
1"""Framework-neutral serverless invocation context for observability.
3The adapter accepts plain invocation metadata so AWS Lambda, Azure Functions,
4Google Cloud Functions, OpenFaaS and custom function runtimes can share one
5dependency-free boundary model. Event payloads are intentionally excluded.
6"""
8from __future__ import annotations
10from collections.abc import Mapping, Sequence
12from .observability import OperationContext
13from .trace_context import W3CTraceContext, trace_context_from_mapping
15__all__ = ["ServerlessContext", "serverless_context_from_invocation"]
17_DEFAULT_INVOCATION_ID_KEYS = ("invocation_id", "request_id", "execution_id")
18_DEFAULT_CORRELATION_ID_KEYS = (
19 "correlation_id",
20 "x-correlation-id",
21 "correlation-id",
22)
25def _safe_text(value: object) -> str | None:
26 if not isinstance(value, str):
27 return None
28 normalized = value.strip()
29 if not normalized or "\r" in value or "\n" in value:
30 return None
31 return normalized
34def _validated_optional_text(value: str | None, field: str) -> str | None:
35 if value is None:
36 return None
37 normalized = _safe_text(value)
38 if normalized is None:
39 raise ValueError(f"{field} must be non-empty single-line text or None")
40 return normalized
43def _normalized_metadata(metadata: Mapping[str, object]) -> dict[str, object]:
44 normalized: dict[str, object] = {}
45 for key, value in metadata.items():
46 if isinstance(key, str): 46 ↛ 45line 46 didn't jump to line 45 because the condition on line 46 was always true
47 normalized[key.lower()] = value
48 return normalized
51def _validated_keys(keys: Sequence[str], field: str) -> tuple[str, ...]:
52 if isinstance(keys, (str, bytes)): 52 ↛ 53line 52 didn't jump to line 53 because the condition on line 52 was never true
53 raise TypeError(f"{field} must be a sequence of metadata keys")
55 result: list[str] = []
56 for key in keys:
57 if not isinstance(key, str): 57 ↛ 58line 57 didn't jump to line 58 because the condition on line 57 was never true
58 raise TypeError(f"{field} entries must be strings")
59 normalized = key.strip().lower()
60 if not normalized: 60 ↛ 61line 60 didn't jump to line 61 because the condition on line 60 was never true
61 raise ValueError(f"{field} entries must not be empty")
62 result.append(normalized)
63 return tuple(result)
66def _first_metadata_value(
67 metadata: Mapping[str, object],
68 keys: Sequence[str],
69) -> str | None:
70 for key in keys:
71 value = _safe_text(metadata.get(key.lower()))
72 if value is not None:
73 return value
74 return None
77def _validated_cold_start(value: bool | None) -> bool | None:
78 if value is None:
79 return None
80 if not isinstance(value, bool):
81 raise TypeError("cold_start must be a boolean or None")
82 return value
85class ServerlessContext:
86 """Invocation operation context plus optional runtime metadata."""
88 __slots__ = ("cold_start", "operation_context", "runtime", "trace_context")
90 def __init__(
91 self,
92 operation_context: OperationContext,
93 trace_context: W3CTraceContext | None,
94 *,
95 cold_start: bool | None,
96 runtime: str | None,
97 ) -> None:
98 self.operation_context = operation_context
99 self.trace_context = trace_context
100 self.cold_start = cold_start
101 self.runtime = runtime
104def serverless_context_from_invocation(
105 function_name: str,
106 *,
107 invocation_id: str | None = None,
108 metadata: Mapping[str, object] | None = None,
109 system: str | None = "serverless",
110 component: str | None = None,
111 correlation_id: str | None = None,
112 cold_start: bool | None = None,
113 runtime: str | None = None,
114 invocation_id_keys: Sequence[str] = _DEFAULT_INVOCATION_ID_KEYS,
115 correlation_id_keys: Sequence[str] = _DEFAULT_CORRELATION_ID_KEYS,
116) -> ServerlessContext:
117 """Build observability context for one serverless function invocation.
119 function_name is the stable deployed function identity. Per-invocation
120 request IDs stay correlation metadata and event payloads are never captured.
121 Explicit invocation and correlation IDs take precedence over metadata
122 fallbacks. Incoming W3C trace context is preserved when present.
123 """
124 if not isinstance(function_name, str): 124 ↛ 125line 124 didn't jump to line 125 because the condition on line 124 was never true
125 raise TypeError("function_name must be a string")
126 operation = function_name.strip()
127 if not operation:
128 raise ValueError("function_name must not be empty")
129 if "\r" in function_name or "\n" in function_name:
130 raise ValueError("function_name must be a single line")
132 if metadata is None:
133 metadata = {}
134 if not isinstance(metadata, Mapping):
135 raise TypeError("metadata must be a mapping or None")
137 normalized = _normalized_metadata(metadata)
138 invocation_keys = _validated_keys(invocation_id_keys, "invocation_id_keys")
139 correlation_keys = _validated_keys(
140 correlation_id_keys,
141 "correlation_id_keys",
142 )
143 explicit_invocation_id = _validated_optional_text(
144 invocation_id,
145 "invocation_id",
146 )
147 explicit_correlation_id = _validated_optional_text(
148 correlation_id,
149 "correlation_id",
150 )
152 operation_context = OperationContext(
153 system=system,
154 component=component,
155 operation=operation,
156 request_id=(
157 explicit_invocation_id or _first_metadata_value(normalized, invocation_keys)
158 ),
159 correlation_id=(
160 explicit_correlation_id
161 or _first_metadata_value(normalized, correlation_keys)
162 ),
163 )
165 trace_context = trace_context_from_mapping(normalized)
166 if trace_context is not None:
167 operation_context = trace_context.to_operation_context(operation_context)
169 return ServerlessContext(
170 operation_context,
171 trace_context,
172 cold_start=_validated_cold_start(cold_start),
173 runtime=_validated_optional_text(runtime, "runtime"),
174 )