Setting the file. One moment.
Di Gateway · 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
This file
Number 37.50
Position 50 of 67
Type Python
Size 7 KB
Lines 153 scripts/cloudwatch/ di_gateway.py
Python · 153 lines · 7 KB
get_application_signals_client
14 from di_error_translation import render_client_error, translate_aws_error
15
16
17 class GatewayError ( Exception ):
18 """Wraps any exception raised by an application-signals call.
19
20 The original exception is preserved on ``original_exc`` so callers can
21 pass the gateway error through ``render_error`` without losing the
22 botocore-specific data ``render_client_error`` needs.
23 """
24
25 def __init__ (self, original_exc: BaseException ):
26 """Wrap ``original_exc``, preserving it for later rendering."""
27 super (). __init__ ( str (original_exc))
28 self .original_exc = original_exc
29
30
31 # The full set of application-signals client methods these tools are allowed to call. The
32 # wrapper functions below pass only these hardcoded names, but ``_call`` validates against
33 # this frozen set before dispatch so the seam cannot be turned into an arbitrary-method
34 # dispatcher by any (future) caller. ``_bind_method`` then selects the bound method by LITERAL
35 # attribute access (one ``if`` per op) — never ``getattr(client, name)`` — so there is no
36 # string-driven dispatch. The two are kept in lockstep by the gateway sync-guard test.
37 _ALLOWED_OPERATIONS = frozenset (
38 {
39 "create_instrumentation_configuration" ,
40 "list_instrumentation_configurations" ,
41 "get_instrumentation_configuration" ,
42 "delete_instrumentation_configuration" ,
43 "batch_delete_instrumentation_configurations" ,
44 "get_instrumentation_configuration_status" ,
45 }
46 )
47
48
49 def _bind_method (client: Any, method_name: str ):
50 """Return the bound boto3 client method for ``method_name`` via literal attribute access.
51
52 boto3 client methods are generated dynamically per client instance, so they cannot be
53 bound as a static dict at import time the way our own module functions can. Instead each
54 allowlisted op is reached by a hardcoded attribute name (``client.create_instrumentation_
55 configuration`` etc.), never ``getattr(client, method_name)``. ``method_name`` has already
56 been checked against ``_ALLOWED_OPERATIONS`` by ``_call``, so the final ``raise`` is
57 unreachable; it keeps the allowlist and these branches in sync (asserted by the tests).
58 """
59 if method_name == "create_instrumentation_configuration" :
60 return client.create_instrumentation_configuration
61 if method_name == "list_instrumentation_configurations" :
62 return client.list_instrumentation_configurations
63 if method_name == "get_instrumentation_configuration" :
64 return client.get_instrumentation_configuration
65 if method_name == "delete_instrumentation_configuration" :
66 return client.delete_instrumentation_configuration
67 if method_name == "batch_delete_instrumentation_configurations" :
68 return client.batch_delete_instrumentation_configurations
69 if method_name == "get_instrumentation_configuration_status" :
70 return client.get_instrumentation_configuration_status
71 raise ValueError ( f "Disallowed application-signals operation: { method_name !r} " )
72
73
74 def _call (method_name: str , ** kwargs: Any) -> Dict[ str , Any]:
75 if method_name not in _ALLOWED_OPERATIONS :
76 raise ValueError ( f "Disallowed application-signals operation: { method_name !r} " )
77 client = get_application_signals_client()
78 method = _bind_method(client, method_name)
79 try :
80 return method( ** kwargs)
81 except (BotoCoreError, ClientError) as exc:
82 # Narrow on purpose: these two cover the full botocore exception
83 # surface (``ClientError`` for service-side errors, ``BotoCoreError``
84 # for credentials/connection/timeout failures). Programming errors
85 # (``AttributeError`` from a typo, ``TypeError`` from a bad kwarg)
86 # propagate unwrapped so they surface as themselves in tracebacks
87 # instead of masquerading as AWS failures.
88 raise GatewayError(exc) from exc
89
90
91 def create_instrumentation_configuration ( ** kwargs: Any) -> Dict[ str , Any]:
92 """Call ``CreateInstrumentationConfiguration`` through the gateway."""
93 return _call( "create_instrumentation_configuration" , ** kwargs)
94
95
96 def list_instrumentation_configurations ( ** kwargs: Any) -> Dict[ str , Any]:
97 """Call ``ListInstrumentationConfigurations`` through the gateway."""
98 return _call( "list_instrumentation_configurations" , ** kwargs)
99
100
101 def get_instrumentation_configuration ( ** kwargs: Any) -> Dict[ str , Any]:
102 """Call ``GetInstrumentationConfiguration`` through the gateway."""
103 return _call( "get_instrumentation_configuration" , ** kwargs)
104
105
106 def delete_instrumentation_configuration ( ** kwargs: Any) -> Dict[ str , Any]:
107 """Call ``DeleteInstrumentationConfiguration`` through the gateway."""
108 return _call( "delete_instrumentation_configuration" , ** kwargs)
109
110
111 def batch_delete_instrumentation_configurations ( ** kwargs: Any) -> Dict[ str , Any]:
112 """Call ``BatchDeleteInstrumentationConfigurations`` through the gateway."""
113 return _call( "batch_delete_instrumentation_configurations" , ** kwargs)
114
115
116 def get_instrumentation_configuration_status ( ** kwargs: Any) -> Dict[ str , Any]:
117 """Call ``GetInstrumentationConfigurationStatus`` through the gateway."""
118 return _call( "get_instrumentation_configuration_status" , ** kwargs)
119
120
121 def render_error (
122 err: GatewayError,
123 * ,
124 action: str ,
125 attempted_label: str = "ATTEMPTED PARAMETERS:" ,
126 attempted: Optional[Mapping[ str , object ]] = None ,
127 possible_causes: Optional[Sequence[ str ]] = None ,
128 troubleshooting: Optional[Sequence[ str ]] = None ,
129 trailer: Optional[ str ] = None ,
130 ) -> str :
131 """Render a ``GatewayError`` using the appropriate error template.
132
133 Callers that want tailored prose for a ``ClientError`` pass
134 ``possible_causes`` / ``troubleshooting`` / ``trailer``; those flow
135 through ``render_client_error``. Callers that pass none of those — and
136 every non-``ClientError`` exception regardless — fall through to
137 ``translate_aws_error``, which carries its own canned bullets per
138 exception type. This preserves the per-tool rendering contract that
139 existed before tools were routed through the gateway.
140 """
141 exc = err.original_exc
142 has_tailored_prose = bool (possible_causes or troubleshooting or trailer)
143 if isinstance (exc, ClientError) and has_tailored_prose:
144 return render_client_error(
145 exc,
146 action = action,
147 attempted_label = attempted_label,
148 attempted = attempted,
149 possible_causes = possible_causes,
150 troubleshooting = troubleshooting,
151 trailer = trailer,
152 )
153 return translate_aws_error(exc, action = action, context = attempted)