AWS - Boto3 waiters style for when InternetGateway, Subnet, Vpc and Dhcp Options are deleted (Eventual Consistency)

vpc.py

#!/usr/bin/env python

import boto3
import logging
import time

from botocore.exceptions import ClientError


class Waiter:
    def __init__(self, client, id, max_attempts=40, delay=15, name=None, error_code=None, logger=logging.getLogger()):
        self.client = client
        self.id = id

        self.max_attempts = max_attempts
        self.delay = delay

        self.name = name
        self.error_code = error_code

        self.logger = logger

    def wait_method(self):
        pass

    def wait(self):
        attempts = 0
        while True:
            if attempts == self.max_attempts:
                raise Exception('Error after waiting for %s(%s) to be deleted.' % (self.name, self.id))
            try:
                self.wait_method()
                self.logger.debug('%s(%s) still exists.' % (self.name, self.id))
            except ClientError as e:
                if e.response.get('Error').get('Code') == self.error_code:
                    self.logger.debug('%s(%s) deleted or not found' % (self.name, self.id))
                    return
                else:
                    raise
            time.sleep(self.delay)
            attempts = attempts + 1


class InternetGatewayDeleted(Waiter):
    def __init__(self, client, id, logger):
        super(InternetGatewayDeleted, self).__init__(client, id, name='Internet Gateway', error_code='InvalidInternetGatewayID.NotFound', logger=logger)

    def wait_method(self):
        self.client.describe_internet_gateways(InternetGatewayIds=[self.id])


class SubnetDeleted(Waiter):
    def __init__(self, client, id, logger):
        super(SubnetDeleted, self).__init__(client, id, name='Subnet', error_code='InvalidSubnetID.NotFound', logger=logger)

    def wait_method(self):
        self.client.describe_subnets(SubnetIds=[self.id])


class VpcDeleted(Waiter):
    def __init__(self, client, id, logger):
        super(VpcDeleted, self).__init__(client, id, name='Vpc', error_code='InvalidVpcID.NotFound', logger=logger)

    def wait_method(self):
        self.client.describe_vpcs(VpcIds=[self.id])


class DhcpOptionsDeleted(Waiter):
    def __init__(self, client, id, logger):
        super(DhcpOptionsDeleted, self).__init__(client, id, name='Dhcp Options', error_code='InvalidDhcpOptionID.NotFound', logger=logger)

    def wait_method(self):
        self.client.describe_dhcp_options(DhcpOptionsIds=[self.id])


ec2 = boto3.resource('ec2')
client = ec2.meta.client

format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
logging.basicConfig(format=format)
logger = logging.getLogger('test')
logger.setLevel(logging.DEBUG)

InternetGatewayDeleted(client, 'igw-ad514dcg', logger).wait()
SubnetDeleted(client, 'subnet-fec1d28b', logger).wait()
VpcDeleted(client, 'vpc-aea345ca', logger).wait()
DhcpOptionsDeleted(client, 'dopt-11d2c674', logger).wait()