Setting the file. One moment.
Di Location · 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
def level
— line 166
This file
Number 37.52
Position 52 of 67
Type Python
Size 15 KB
Lines 399 scripts/cloudwatch/ di_location.py
Python · 399 lines · 15 KB
16
17 Tools never construct API dicts directly; they call ``parse_*_inputs`` and
18 then ``loc.to_api_payload()`` / ``loc.to_identifier()``. Renderers never
19 inspect raw union dicts; they call ``location_from_response`` and use the
20 type's instance methods.
21 """
22
23 from dataclasses import dataclass, field
24 from types import MappingProxyType
25 from typing import Any, Dict, List, Mapping, Optional, Tuple, Union
26
27 from di_validation import _validate_location_inputs, canonical_language
28
29 _EMPTY_EXTRA_FIELDS : Mapping[ str , Any] = MappingProxyType({})
30
31
32 def _freeze_mapping (value: Mapping[ str , Any]) -> Mapping[ str , Any]:
33 """Wrap an extra-fields dict in a read-only proxy.
34
35 Prevents frozen dataclasses from being mutated through their containers.
36 Idempotent: an existing ``MappingProxyType`` is returned unchanged.
37 """
38 if isinstance (value, MappingProxyType):
39 return value
40 return MappingProxyType( dict (value))
41
42
43 @dataclass ( frozen = True )
44 class CodeLocation :
45 """A code-based instrumentation target (BREAKPOINT or PROBE).
46
47 Which fields identify the target vs. which are metadata depends on language:
48
49 * Java — ``code_unit`` (package), ``class_name`` (simple name), and
50 ``method_name`` together identify the target; all required.
51 * Python — ``code_unit`` (dotted module path) and ``method_name``
52 identify the target; ``class_name`` is optional (qualifies a
53 method defined in a class).
54 * JavaScript — ``file_path`` + ``line_number`` identify the target;
55 ``code_unit``/``class_name``/``method_name`` are not used.
56
57 ``line_number`` makes any target line-level (fires at that line rather than
58 on method entry/exit).
59 """
60
61 language: str
62 file_path: str
63 code_unit: Optional[ str ] = None
64 class_name: Optional[ str ] = None
65 method_name: Optional[ str ] = None
66 line_number: Optional[ int ] = None
67 extra_fields: Mapping[ str , Any] = field( default_factory =lambda : _EMPTY_EXTRA_FIELDS )
68
69 def __post_init__ (self) -> None :
70 """Freeze ``extra_fields`` past the dataclass frozen guard."""
71 # frozen=True only blocks reassignment; the dict itself stays
72 # mutable unless we wrap it. Use object.__setattr__ to assign past
73 # the frozen guard.
74 object . __setattr__ ( self , "extra_fields" , _freeze_mapping( self .extra_fields))
75
76 def describe (self) -> str :
77 """Return a one-line human description of the code target."""
78 target = self .file_path or "N/A"
79 if self .class_name:
80 target += f " :: { self .class_name } "
81 if self .method_name:
82 target += f ". { self .method_name } "
83 if self .line_number is not None :
84 target += f ":L { self .line_number } "
85 return target
86
87 def level (self) -> str :
88 """Return the breakpoint granularity (line-level or function-level)."""
89 if self .line_number is not None :
90 return f "LINE-LEVEL (L { self .line_number } )"
91 return "FUNCTION/METHOD-LEVEL"
92
93 def format_details (self, location_hash: Optional[ str ] = None ) -> str :
94 """Render the code location as labeled detail lines."""
95 lines = [ "- LocationKind: CODE" ]
96 if location_hash:
97 lines.append( f "- LocationHash: { location_hash } " )
98 ordered = [
99 ( "Language" , self .language),
100 ( "File Path" , self .file_path),
101 ( "Code Unit" , self .code_unit),
102 ( "Class Name" , self .class_name),
103 ( "Method Name" , self .method_name),
104 ( "Line Number" , self .line_number),
105 ]
106 for label, value in ordered:
107 # Mirror legacy ``format_location_details`` semantics: present-but-empty
108 # fields (``Language=""``) still render as ``- Language: `` so missing
109 # API fields produce a visible blank instead of being silently dropped.
110 if value is not None :
111 lines.append( f "- { label } : { value } " )
112 for key in sorted ( self .extra_fields.keys()):
113 lines.append( f "- { key } : { self .extra_fields[key] } " )
114 return " \n " .join(lines) + " \n "
115
116 def to_api_payload (self) -> Dict[ str , Any]:
117 """Return the CodeLocation create-request payload."""
118 return { "CodeLocation" : self ._to_code_location_dict()}
119
120 def to_identifier (self) -> Dict[ str , Any]:
121 """Return the CodeLocation lookup identifier payload."""
122 return { "CodeLocation" : self ._to_code_location_dict()}
123
124 def _to_code_location_dict (self) -> Dict[ str , Any]:
125 payload: Dict[ str , Any] = {
126 "Language" : self .language,
127 "FilePath" : self .file_path,
128 }
129 if self .code_unit:
130 payload[ "CodeUnit" ] = self .code_unit
131 if self .class_name:
132 payload[ "ClassName" ] = self .class_name
133 if self .method_name:
134 payload[ "MethodName" ] = self .method_name
135 if self .line_number is not None :
136 payload[ "LineNumber" ] = self .line_number
137 return payload
138
139
140 @dataclass ( frozen = True )
141 class HashLocation :
142 """An existing configuration referenced by its 16-char location hash.
143
144 Carries a deliberately narrower interface than its sibling variants:
145
146 * ``to_api_payload`` is *unsupported*: a hash cannot describe a *new*
147 configuration — ``create_instrumentation_configuration`` requires
148 a real CodeLocation.
149 * ``format_details`` is *unsupported*: a hash has no fields beyond
150 itself; ``render_location_block`` prints the hash via its
151 ``HashLocation`` special case instead.
152
153 Both methods exist as stubs that raise ``NotImplementedError`` with a
154 descriptive message rather than being absent. The asymmetry is still
155 the design — these are lookup-only — but the explicit raise turns
156 a confusing ``AttributeError`` into a clear "use ``to_identifier()``
157 instead" message when a future caller forgets the discipline.
158 """
159
160 location_hash: str
161
162 def describe (self) -> str :
163 """Return a one-line description naming the location hash."""
164 return f "LocationHash { self .location_hash } "
165
166 def level (self) -> Optional[ str ]:
167 """Return None — a hash carries no breakpoint granularity."""
168 return None
169
170 def to_identifier (self) -> Dict[ str , Any]:
171 """Return the LocationHash lookup identifier payload."""
172 return { "LocationHash" : self .location_hash}
173
174 def to_api_payload (self) -> Dict[ str , Any]:
175 """Unsupported — a hash cannot describe a new configuration."""
176 raise NotImplementedError (
177 "HashLocation cannot be used in create requests — use to_identifier() instead. "
178 "create_instrumentation_configuration requires a CodeLocation."
179 )
180
181 def format_details (self, location_hash: Optional[ str ] = None ) -> str :
182 """Unsupported — a hash has no fields; describe() gives a one-liner."""
183 raise NotImplementedError (
184 "HashLocation has no fields to format — render_location_block handles it directly. "
185 "Use describe() for a one-line target string."
186 )
187
188
189 @dataclass ( frozen = True )
190 class UnknownLocation :
191 """A location union returned by the API that does not match any known variant.
192
193 A forward-compat fallback: ``location_from_response`` produces this so
194 renderers don't crash on future API additions. Input parsers never
195 produce it. Mirrors ``UnknownCapture`` in shape and naming — both are
196 public so callers doing exhaustive ``isinstance`` matching don't need
197 to reach into a private name.
198
199 ``raw`` is wrapped in ``MappingProxyType`` so the ``frozen=True``
200 contract holds against mutation through the source dict.
201 """
202
203 raw: Mapping[ str , Any]
204
205 def __post_init__ (self) -> None :
206 """Wrap ``raw`` in a read-only proxy to honor the frozen contract."""
207 if not isinstance ( self .raw, MappingProxyType):
208 object . __setattr__ ( self , "raw" , MappingProxyType( dict ( self .raw)))
209
210 def describe (self) -> str :
211 """Return 'N/A' — an unknown location has no describable target."""
212 return "N/A"
213
214 def level (self) -> Optional[ str ]:
215 """Return None — an unknown location has no granularity."""
216 return None
217
218 def format_details (self, location_hash: Optional[ str ] = None ) -> str :
219 """Render the unknown location's raw fields as detail lines."""
220 lines = [ "- LocationKind: UNKNOWN" ]
221 if location_hash:
222 lines.append( f "- LocationHash: { location_hash } " )
223 if self .raw:
224 for key in sorted ( self .raw.keys()):
225 lines.append( f "- { key } : { self .raw[key] } " )
226 else :
227 lines.append( "- Location payload could not be parsed." )
228 return " \n " .join(lines) + " \n "
229
230
231 Location = Union[CodeLocation, HashLocation, UnknownLocation]
232
233 # A location resolved from *caller* inputs (create/lookup). Unlike ``Location``,
234 # this never includes ``UnknownLocation`` — that variant only arises when parsing
235 # an API *response* (see ``location_from_response``). Narrowing the parser return
236 # types to this union lets callers use ``to_identifier``/``to_api_payload``
237 # without a cast, since both members implement them.
238 ResolvedLocation = Union[CodeLocation, HashLocation]
239
240
241 # ──────────────────────────── input parsers ────────────────────────────
242
243
244 def parse_create_inputs (
245 * ,
246 normalized_type: str ,
247 language: Optional[ str ] = None ,
248 file_path: Optional[ str ] = None ,
249 code_unit: Optional[ str ] = None ,
250 class_name: Optional[ str ] = None ,
251 method_name: Optional[ str ] = None ,
252 line_number: Optional[ int ] = None ,
253 ) -> Tuple[Optional[ResolvedLocation], Optional[ str ]]:
254 """Parse tool kwargs into a ``Location`` for a create-instrumentation call.
255
256 HashLocation is not accepted: callers cannot create an instrumentation from
257 an existing hash. Returns ``(location, None)`` on success or
258 ``(None, error_text)`` when inputs are invalid; ``error_text`` is rendered
259 verbatim back to the caller.
260 """
261 if not language or not file_path:
262 return None , (
263 "ERROR: BREAKPOINT/PROBE require language and file_path. \n "
264 'Example: language="Python", file_path="/app/handler.py"'
265 )
266
267 location_validation_error = _validate_location_inputs(
268 language = language,
269 file_path = file_path,
270 code_unit = code_unit,
271 class_name = class_name,
272 method_name = method_name,
273 line_number = line_number,
274 )
275 if location_validation_error:
276 return None , location_validation_error
277
278 return (
279 CodeLocation(
280 language = canonical_language(language) or language,
281 file_path = file_path,
282 code_unit = code_unit,
283 class_name = class_name,
284 method_name = method_name,
285 line_number = line_number,
286 ),
287 None ,
288 )
289
290
291 def parse_lookup_inputs (
292 * ,
293 normalized_type: str ,
294 location_hash: Optional[ str ] = None ,
295 language: Optional[ str ] = None ,
296 file_path: Optional[ str ] = None ,
297 code_unit: Optional[ str ] = None ,
298 class_name: Optional[ str ] = None ,
299 method_name: Optional[ str ] = None ,
300 line_number: Optional[ int ] = None ,
301 allow_code_location_lookup: bool = True ,
302 ) -> Tuple[Optional[ResolvedLocation], Optional[ str ]]:
303 """Parse tool kwargs into a ``Location`` for a lookup operation.
304
305 Lookup accepts a location_hash or a code location. Resolution order:
306 hash > code location. Returns ``(location, None)`` or ``(None, error_text)``.
307 """
308 if location_hash:
309 return HashLocation( location_hash = location_hash), None
310
311 if language and file_path:
312 if not allow_code_location_lookup:
313 return None , "code location lookup is not supported for this operation."
314 return (
315 CodeLocation(
316 language = canonical_language(language) or language,
317 file_path = file_path,
318 code_unit = code_unit,
319 class_name = class_name,
320 method_name = method_name,
321 line_number = line_number,
322 ),
323 None ,
324 )
325
326 return None , (
327 "missing location identifier input. Provide location_hash "
328 "OR language+file_path (code location)."
329 )
330
331
332 # ──────────────────────────── response parser ────────────────────────────
333
334
335 _KNOWN_CODE_FIELDS = { "Language" , "FilePath" , "CodeUnit" , "ClassName" , "MethodName" , "LineNumber" }
336
337
338 def location_from_response (union_dict: Optional[Dict[ str , Any]]) -> Location:
339 """Parse a ``Location`` union returned by the API into the ADT.
340
341 Returns ``UnknownLocation`` if the dict has no recognized variant — this
342 keeps response rendering forward-compatible with future API additions.
343 """
344 if not isinstance (union_dict, dict ):
345 return UnknownLocation( raw = {})
346
347 code = union_dict.get( "CodeLocation" )
348 if isinstance (code, dict ):
349 return _code_location_from_dict(code)
350
351 if "Language" in union_dict or "FilePath" in union_dict:
352 return _code_location_from_dict(union_dict)
353
354 return UnknownLocation( raw = dict (union_dict))
355
356
357 def _code_location_from_dict (payload: Dict[ str , Any]) -> CodeLocation:
358 extras = {k: v for k, v in payload.items() if k not in _KNOWN_CODE_FIELDS }
359 return CodeLocation(
360 language = payload.get( "Language" , "" ),
361 file_path = payload.get( "FilePath" , "" ),
362 code_unit = payload.get( "CodeUnit" ),
363 class_name = payload.get( "ClassName" ),
364 method_name = payload.get( "MethodName" ),
365 line_number = payload.get( "LineNumber" ),
366 extra_fields = extras,
367 )
368
369
370 # ──────────────────────────── shared helpers ────────────────────────────
371
372
373 def render_location_block (location: Location, location_hash: Optional[ str ] = None ) -> str :
374 """Render the standard LOCATION block plus the optional INSTRUMENTATION level line."""
375 if isinstance (location, HashLocation):
376 # HashLocation has no API-side dict to format; should not normally
377 # reach a renderer, but keep the path safe.
378 block = f "- LocationKind: HASH \n - LocationHash: { location.location_hash }\n "
379 return block
380
381 output = location.format_details( location_hash = location_hash)
382 level = location.level()
383 if level:
384 output += " \n INSTRUMENTATION: \n "
385 output += f "- Level: { level }\n "
386 return output
387
388
389 __all__ : List[ str ] = [
390 "CodeLocation" ,
391 "HashLocation" ,
392 "UnknownLocation" ,
393 "Location" ,
394 "ResolvedLocation" ,
395 "parse_create_inputs" ,
396 "parse_lookup_inputs" ,
397 "location_from_response" ,
398 "render_location_block" ,
399 ]