Setting the file. One moment.
Di Capture · AWS Observability · aws/agent-toolkit-for-aws · Skills Docs
Back to the top of the page 69
Creating Amazon Aurora Db Cluster With Instances
92
Routing Traffic With Route53 And CloudFront
Resilience Program Design
Debugging Lambda Timeouts
scripts/cloudwatch/ di_capture.py
Python · 210 lines · 9 KB
absent from the API payload) and a *present-but-empty* list (``()`` — key
16 emitted as ``[]``) so an API response round-trips parse → payload → parse
17 without losing information. It does *not* assert what the backend means by
18 either shape. The *create* operation deliberately does not expose both shapes:
19 it rejects empty lists and the ``*`` wildcard and treats an omitted list as
20 "capture nothing for that field" (see ``create_instrumentation``). Renderers
21 report the raw shape rather than labeling it "all" or "none".
22 """
23
24 from dataclasses import dataclass, field
25 from types import MappingProxyType
26 from typing import Any, Dict, Mapping, Optional, Sequence, Union
27
28
29 @dataclass ( frozen = True )
30 class CaptureLimits :
31 """Optional size caps applied to a code-capture payload."""
32
33 max_hits: Optional[ int ] = None
34 max_string_length: Optional[ int ] = None
35 max_collection_width: Optional[ int ] = None
36 max_collection_depth: Optional[ int ] = None
37 max_stack_frames: Optional[ int ] = None
38 max_stack_trace_size: Optional[ int ] = None
39 max_object_depth: Optional[ int ] = None
40 max_fields_per_object: Optional[ int ] = None
41
42 def is_empty (self) -> bool :
43 """Return True when no capture limit is set."""
44 return all (
45 v is None
46 for v in (
47 self .max_hits,
48 self .max_string_length,
49 self .max_collection_width,
50 self .max_collection_depth,
51 self .max_stack_frames,
52 self .max_stack_trace_size,
53 self .max_object_depth,
54 self .max_fields_per_object,
55 )
56 )
57
58 def to_api_payload (self) -> Dict[ str , int ]:
59 """Render the set capture limits as the CaptureLimits payload."""
60 payload: Dict[ str , int ] = {}
61 if self .max_hits is not None :
62 payload[ "MaxHits" ] = self .max_hits
63 if self .max_string_length is not None :
64 payload[ "MaxStringLength" ] = self .max_string_length
65 if self .max_collection_width is not None :
66 payload[ "MaxCollectionWidth" ] = self .max_collection_width
67 if self .max_collection_depth is not None :
68 payload[ "MaxCollectionDepth" ] = self .max_collection_depth
69 if self .max_stack_frames is not None :
70 payload[ "MaxStackFrames" ] = self .max_stack_frames
71 if self .max_stack_trace_size is not None :
72 payload[ "MaxStackTraceSize" ] = self .max_stack_trace_size
73 if self .max_object_depth is not None :
74 payload[ "MaxObjectDepth" ] = self .max_object_depth
75 if self .max_fields_per_object is not None :
76 payload[ "MaxFieldsPerObject" ] = self .max_fields_per_object
77 return payload
78
79
80 @dataclass ( frozen = True )
81 class CodeCapture :
82 """Capture configuration for BREAKPOINT and PROBE.
83
84 ``capture_arguments`` and ``capture_locals`` are stored as tuples so the
85 ``frozen=True`` immutability contract holds against mutation through the
86 container (a caller's reference to the source list cannot mutate this
87 instance). Constructors still accept any iterable of strings — including
88 a list — and ``__post_init__`` converts to ``tuple``. This is the same
89 discipline ``Location`` applies to ``extra_fields`` via
90 ``MappingProxyType``.
91
92 The ``Optional`` distinction is preserved across the round-trip: ``None``
93 omits the key from the API payload, while an empty tuple ``()`` emits the
94 key as ``[]``. This ADT does not assign semantics to either shape; the
95 create tool restricts which shapes it will send (see
96 ``create_instrumentation``).
97 """
98
99 capture_return: bool
100 capture_stack_trace: bool
101 # Declared as ``Sequence[str]`` because the constructor accepts any string
102 # sequence (commonly a list); ``__post_init__`` coerces to ``tuple`` so the
103 # stored value is always an immutable tuple despite the broader input type.
104 capture_arguments: Optional[Sequence[ str ]] = None
105 capture_locals: Optional[Sequence[ str ]] = None
106 limits: CaptureLimits = field( default_factory = CaptureLimits)
107
108 def __post_init__ (self) -> None :
109 """Coerce argument/local name lists to tuples for the frozen contract."""
110 if self .capture_arguments is not None and not isinstance ( self .capture_arguments, tuple ):
111 object . __setattr__ ( self , "capture_arguments" , tuple ( self .capture_arguments))
112 if self .capture_locals is not None and not isinstance ( self .capture_locals, tuple ):
113 object . __setattr__ ( self , "capture_locals" , tuple ( self .capture_locals))
114
115 def to_api_payload (self) -> Dict[ str , Any]:
116 """Render the CodeCapture create-request payload."""
117 config: Dict[ str , Any] = {
118 "CaptureReturn" : self .capture_return,
119 "CaptureStackTrace" : self .capture_stack_trace,
120 "CaptureLimits" : self .limits.to_api_payload(),
121 }
122 if self .capture_arguments is not None :
123 config[ "CaptureArguments" ] = list ( self .capture_arguments)
124 if self .capture_locals is not None :
125 config[ "CaptureLocals" ] = list ( self .capture_locals)
126 return { "CodeCapture" : config}
127
128
129 @dataclass ( frozen = True )
130 class UnknownCapture :
131 """A CaptureConfiguration union that did not match a known variant.
132
133 ``raw`` is wrapped in ``MappingProxyType`` to keep the ``frozen=True``
134 contract intact against mutation through the source dict — the same
135 discipline applied to ``Location.extra_fields`` and
136 ``CodeCapture.capture_arguments``.
137 """
138
139 raw: Mapping[ str , Any]
140
141 def __post_init__ (self) -> None :
142 """Wrap ``raw`` in a read-only proxy to honor the frozen contract."""
143 if not isinstance ( self .raw, MappingProxyType):
144 object . __setattr__ ( self , "raw" , MappingProxyType( dict ( self .raw)))
145
146
147 Capture = Union[CodeCapture, UnknownCapture]
148
149
150 _CODE_CAPTURE_HINT_KEYS = (
151 "CaptureReturn" ,
152 "CaptureLimits" ,
153 "CaptureArguments" ,
154 "CaptureStackTrace" ,
155 )
156
157
158 def capture_from_response (union_dict: Optional[Dict[ str , Any]]) -> Capture:
159 """Parse a ``CaptureConfiguration`` union returned by the API into the ADT.
160
161 Falls back to inferring a ``CodeCapture`` if a CodeCapture-shaped dict
162 is passed without the ``CodeCapture`` wrapper key — this matches the
163 legacy ``extract_capture_variant`` fallback that some response shapes
164 relied on.
165 """
166 if not isinstance (union_dict, dict ):
167 return UnknownCapture( raw = {})
168
169 code = union_dict.get( "CodeCapture" )
170 if isinstance (code, dict ):
171 return _code_capture_from_dict(code)
172
173 if any (key in union_dict for key in _CODE_CAPTURE_HINT_KEYS ):
174 return _code_capture_from_dict(union_dict)
175
176 return UnknownCapture( raw = dict (union_dict))
177
178
179 def _code_capture_from_dict (payload: Dict[ str , Any]) -> CodeCapture:
180 raw_limits = payload.get( "CaptureLimits" ) or {}
181 if not isinstance (raw_limits, dict ):
182 raw_limits = {}
183 limits = CaptureLimits(
184 max_hits = raw_limits.get( "MaxHits" ),
185 max_string_length = raw_limits.get( "MaxStringLength" ),
186 max_collection_width = raw_limits.get( "MaxCollectionWidth" ),
187 max_collection_depth = raw_limits.get( "MaxCollectionDepth" ),
188 max_stack_frames = raw_limits.get( "MaxStackFrames" ),
189 max_stack_trace_size = raw_limits.get( "MaxStackTraceSize" ),
190 max_object_depth = raw_limits.get( "MaxObjectDepth" ),
191 max_fields_per_object = raw_limits.get( "MaxFieldsPerObject" ),
192 )
193 return CodeCapture(
194 capture_return = bool (payload.get( "CaptureReturn" )),
195 capture_stack_trace = bool (payload.get( "CaptureStackTrace" )),
196 capture_arguments = (
197 payload.get( "CaptureArguments" ) if "CaptureArguments" in payload else None
198 ),
199 capture_locals = payload.get( "CaptureLocals" ) if "CaptureLocals" in payload else None ,
200 limits = limits,
201 )
202
203
204 __all__ = [
205 "CaptureLimits" ,
206 "CodeCapture" ,
207 "UnknownCapture" ,
208 "Capture" ,
209 "capture_from_response" ,
210 ]