MCPcopy Create free account
hub / github.com/agenticsorg/edge-agents / wait_for

Method wait_for

scripts/agents/devops/src/aws/base.py:370–400  ·  view source on GitHub ↗

Wait for a specific state using a boto3 waiter. Args: waiter_name: Name of the waiter to use (e.g., 'instance_running'). waiter_args: Arguments for the waiter. max_attempts: Maximum number of attempts. delay: Delay between att

(
        self,
        waiter_name: str,
        waiter_args: Dict[str, Any],
        max_attempts: int = 40,
        delay: int = 15
    )

Source from the content-addressed store, hash-verified

368 ]
369
370 def wait_for(
371 self,
372 waiter_name: str,
373 waiter_args: Dict[str, Any],
374 max_attempts: int = 40,
375 delay: int = 15
376 ) -> None:
377 """
378 Wait for a specific state using a boto3 waiter.
379
380 Args:
381 waiter_name: Name of the waiter to use (e.g., 'instance_running').
382 waiter_args: Arguments for the waiter.
383 max_attempts: Maximum number of attempts.
384 delay: Delay between attempts in seconds.
385
386 Raises:
387 AWSServiceError: If the wait operation times out or fails.
388 """
389 try:
390 waiter = self.client.get_waiter(waiter_name)
391 waiter.wait(
392 WaiterConfig={
393 'Delay': delay,
394 'MaxAttempts': max_attempts
395 },
396 **waiter_args
397 )
398 except Exception as e:
399 operation = f"wait_for_{waiter_name}"
400 self.handle_error(e, operation)
401
402 def format_resource_name(self, name_format: str, **kwargs) -> str:
403 """

Callers 7

create_instanceMethod · 0.80
start_instanceMethod · 0.80
stop_instanceMethod · 0.80
terminate_instanceMethod · 0.80
create_amiMethod · 0.80
test_wait_forMethod · 0.80
test_wait_for_errorMethod · 0.80

Calls 1

handle_errorMethod · 0.95

Tested by 2

test_wait_forMethod · 0.64
test_wait_for_errorMethod · 0.64