Python boto3 模块，setup_default_session() 实例源码

def __init__(self, role_to_assume):
        self.role_to_assume = role_to_assume
        boto3.setup_default_session(profile_name='default')
        if role_to_assume:
            self.client = boto3.client('sts')
            response = self.client.assume_role(
                RoleArn=role_to_assume, RoleSessionName="ecs_explorer")

            creds = response['Credentials']
            self.ecs = boto3.client('ecs',
                                    aws_access_key_id=creds['AccessKeyId'],
                                    aws_secret_access_key=creds['SecretAccessKey'],
                                    aws_session_token=creds['SessionToken'])

            self.ec2 = boto3.client('ec2',
                                    aws_access_key_id=creds['AccessKeyId'],
                                    aws_secret_access_key=creds['SecretAccessKey'],
                                    aws_session_token=creds['SessionToken'])
        else: 
            self.ecs = boto3.client('ecs')
            self.ec2 = boto3.client('ec2')

def run(self, command: str) -> None:
        """Execute one of the available commands.

        Args:
            command: Command in self.commands.
        """
        boto3.setup_default_session(region_name=self._config.aws_region)

        # Validate the configuration.
        try:
            if command not in {'compile_rules', 'configure', 'unit_test'}:
                self._config.validate()
            getattr(self, command)()  # Command validation already happened in the ArgumentParser.
        except InvalidConfigError as error:
            sys.exit('ERROR: {}\nPlease run "python3 manage.py configure"'.format(error))
        except TestFailureError as error:
            sys.exit('TEST FAILED: {}'.format(error))

def get_service(hass, config):
    """Get the AWS SQS notification service."""
    # pylint: disable=import-error
    import boto3

    aws_config = config.copy()

    del aws_config[CONF_PLATFORM]
    del aws_config[CONF_NAME]

    profile = aws_config.get(CONF_PROFILE_NAME)

    if profile is not None:
        boto3.setup_default_session(profile_name=profile)
        del aws_config[CONF_PROFILE_NAME]

    sqs_client = boto3.client("sqs", **aws_config)

    return AWSSQS(sqs_client)

def get_service(hass, config):
    """Get the AWS SNS notification service."""
    # pylint: disable=import-error
    import boto3

    aws_config = config.copy()

    del aws_config[CONF_PLATFORM]
    del aws_config[CONF_NAME]

    profile = aws_config.get(CONF_PROFILE_NAME)

    if profile is not None:
        boto3.setup_default_session(profile_name=profile)
        del aws_config[CONF_PROFILE_NAME]

    sns_client = boto3.client("sns", **aws_config)

    return AWSSNS(sns_client)

def get_service(hass, config):
    """Get the AWS Lambda notification service."""
    context_str = json.dumps({'hass': hass.config.as_dict(),
                              'custom': config[CONF_CONTEXT]})
    context_b64 = base64.b64encode(context_str.encode("utf-8"))
    context = context_b64.decode("utf-8")

    # pylint: disable=import-error
    import boto3

    aws_config = config.copy()

    del aws_config[CONF_PLATFORM]
    del aws_config[CONF_NAME]
    del aws_config[CONF_CONTEXT]

    profile = aws_config.get(CONF_PROFILE_NAME)

    if profile is not None:
        boto3.setup_default_session(profile_name=profile)
        del aws_config[CONF_PROFILE_NAME]

    lambda_client = boto3.client("lambda", **aws_config)

    return AWSLambda(lambda_client, context)

def clean_boto_configuration(monkeypatch):
    monkeypatch.setenv('AWS_ACCESS_KEY_ID', 'AK000000000000000000')
    monkeypatch.setenv('AWS_SECRET_ACCESS_KEY',
                       '0000000000000000000000000000000000000000')
    monkeypatch.setenv('AWS_DEFAULT_REGION', 'us-east-1')
    # Reset previously created default session (for credentials)
    boto3.setup_default_session()

def __init__(self, debug=False, profile=None):
        """
        Init
        :param debug:
        """
        self.debug = debug
        self.profile = profile
        if self.profile:
            boto3.setup_default_session(profile_name=self.profile)
        PrintMsg.debug = self.debug
        self.client = boto3.client('iam')

def main():
    artifacts_dir = "artifacts"
    suite = unittest.TestSuite()

    parser = argparse.ArgumentParser()
    parser.add_argument("-c", "--config", type=str, required=True, help="Selects the config file")
    parser.add_argument("-p", "--profile", type=str, help="Specifies the boto profile to choose from  ~/.aws/config")
    parser.add_argument("--report", default='html', help="Prints test execution on the console rather than generating a HTML report", choices=['text', 'html'])
    args = parser.parse_args()

    if args.profile :
        boto3.setup_default_session(profile_name=args.profile)

    testConfig = yaml.load(open(args.config, 'r'))

    for testCategory, tests in testConfig.iteritems():
        for test, enabled in tests.iteritems():
            if enabled:
                suite.addTest(eval(testCategory+"Audit")(test))
    runner = ''

    if args.report == 'text':
        runner = unittest.TextTestRunner(verbosity=2)
    elif args.report == 'html':
        reportFile = open("test_results.html", "w")
        runner = HTMLTestRunner.HTMLTestRunner(
            stream=reportFile,
            title='aws-security-test - Report',
            verbosity=2
        )
    else:
        print 'Invalid report type'
        exit(1)

    if not os.path.exists(artifacts_dir):
        os.makedirs(artifacts_dir)

    testExecution = runner.run(suite)
    return len(testExecution.failures) + len(testExecution.errors)

def __init__(self, subject, email_from, email_to, credentials):
        Email.__init__(self, subject, email_from, email_to)
        boto3.setup_default_session(
            aws_access_key_id=credentials['AWS_ID'],
            aws_secret_access_key=credentials['AWS_PASS'],
            region_name=credentials['region']
        )
        self.client = boto3.client('ses')

def __init__(self, type, service=None, region='us-west-2', profile='default'):
        self.region = region
        self.connection_type = type
        self.service = service
        self.client = None
        self.resource = None
        self.profile = profile
        try:
            boto3.setup_default_session(profile_name=self.profile)
        except Exception as e:
            logger.info("Problem setting default boto3 session: {}".format(e))

def pytest_configure():
    boto3.setup_default_session(region_name='eu-west-1')

def main():
    argparser = argparse.ArgumentParser()
    argparser.add_argument('--site', help='Name of the Site VPC', required=True)
    argparser.add_argument('--cidr', help='CIDR to add route for', required=True)
    argparser.add_argument('--peeringid', help='Peering connection we are going to use', required=True)
    argparser.add_argument('--profile', help='AWS Profile to use', required=True)
    args = argparser.parse_args()
    cidr = args.cidr
    site = args.site
    peeringid = args.peeringid
    profile = args.profile

    if profile != None: boto3.setup_default_session(profile_name=profile)
    ec2resource = boto3.resource('ec2', region_name='eu-central-1')
    ec2Client = boto3.client('ec2', region_name='eu-central-1')


    vpcid = get_vpcid(site, client = ec2Client)
    print('Looking up route tables for %s' % vpcid)
    route_tables = get_route_tables(vpcid, ec2resource)
    print ('Retrieved %s tables, proceeding to add tables' % len(route_tables))
    for table_id in route_tables:
        print('Adding route for %s to %s' % (cidr, peeringid))
        try:
            response = add_peering_route(table_id, cidr, peeringid, ec2resource)
            if response:
                continue
            else:
                print('Error occurred adding route to %s' % table_id)
        except botocore.exceptions.ClientError as e:
            if e.response['Error']['Code'] == 'RouteAlreadyExists':
                print('Route already exists on %s, continuing' % table_id)
                continue
            else:
                print('Unexpected error: %s' % e)



    print('Routes have been added!')

def old(region, profile='default'):
    project_details = json.load(open('project.json'))
    boto3.setup_default_session(
        profile_name=profile,
        region_name=region
    )
    client = boto3.client('apigateway', region_name=region)
    raml = ramlfications.parse('api_schema.raml')
    api_name = raml.title
    api_gateway = get_api_by_name(client, api_name)
    if api_gateway is None:
        api_gateway = client.create_rest_api(name=api_name)
    aws_resources = client.get_resources(restApiId=api_gateway['id'])['items']
    root = grab_root_resource(aws_resources)
    resources = api.transform_resources(raml, raml.resources)
    resources = associate_resources(aws_resources, resources)
    aws_authorizers = client.get_authorizers(restApiId=api_gateway['id'])['items']  # NOQA
    authorizers = associate_authorizers(aws_authorizers, raml.security_schemes or [])  # NOQA
    create_authorizer = functools.partial(
        create_security_scheme,
        client,
        api_gateway['id'],
        project_details['name']
    )
    authorizers = map(create_authorizer, authorizers)

    for resource in resources:
        print 'Creating Resource'
        create_resource(
            client,
            api_gateway['id'],
            root['id'],
            resource,
            project_details['name'],
            authorizers
        )
    deployment = client.create_deployment(
        restApiId=api_gateway['id'],
        stageName=raml.base_uri
    )
    data = {
        'deployment': deployment['id'],
        'api': api_gateway['id'],
        'uri': 'https://{}.execute-api.{}.amazonaws.com/{}/'.format(
            api_gateway['id'],
            region,
            raml.base_uri
        )
    }
    print data