How to use the botocore.session.get_session function in botocore

To help you get started, we’ve selected a few botocore examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github boto / boto3 / tests / unit / docs / __init__.py View on Github external
def setup_client_and_resource(self):
        self._write_models()
        self.loader = Loader(extra_search_paths=[self.root_dir])
        self.botocore_session = botocore.session.get_session()
        self.botocore_session.register_component('data_loader', self.loader)
        self.session = Session(
            botocore_session=self.botocore_session, region_name='us-east-1')
        self.client = self.session.client('myservice', 'us-east-1')
        self.resource = self.session.resource('myservice', 'us-east-1')
github aws / chalice / tests / unit / deploy / test_newdeployer.py View on Github external
def test_can_create_deletion_deployer():
    session = botocore.session.get_session()
    deployer = create_deletion_deployer(TypedAWSClient(session), UI())
    assert isinstance(deployer, Deployer)
github mintel / pytest-localstack / tests / integration / test_contrib / test_botocore.py View on Github external
def test_patch(region_name, service_name, make_test_session):
    """Test patching."""
    localstack = make_test_session(region_name=region_name)

    with localstack:
        # Haven't patched yet.
        # A regular botocore client should point to AWS right now.
        original_bc_session = botocore.session.get_session()
        original_bc_client = original_bc_session.create_client(
            service_name, localstack.region_name
        )
        assert "127.0.0.1" not in original_bc_client._endpoint.host

        with localstack.botocore.patch_botocore():
            # Original client should now point to Localstack
            assert "127.0.0.1" in original_bc_client._endpoint.host

            # Original session should create Localstack clients
            ls_client = original_bc_session.create_client(
                service_name, localstack.region_name
            )
            assert "127.0.0.1" in ls_client._endpoint.host

        # Original client back to AWS
github canonical / cloud-init / tests / cloud_tests / platforms / ec2 / platform.py View on Github external
def get_session():
    mysess = session.get_session()
    mysess.unregister('after-call.ec2.GetConsoleOutput',
                      handlers.decode_console_output)
    mysess.register('after-call.ec2.GetConsoleOutput',
                    _decode_console_output_as_bytes)
    return boto3.Session(botocore_session=mysess)
github piontas / python-aada / aada / cli.py View on Github external
def main(self):
        parser = self._create_parser()
        self._parsed_args = parser.parse_args(self.args)

        if self._parsed_args.profile:
            self._session = Session(profile=self._parsed_args.profile)
        else:
            self._session = get_session()

        if self._parsed_args.debug:
            self._debug = True

        if self._parsed_args.no_headless:
            self._headless = False

        if self._parsed_args.role:
            self._role = self._parsed_args.role

        if self._parsed_args.account:
            self._account = self._parsed_args.account

        return self.__getattribute__('_{}'.format(self._parsed_args.command))()
github ansible / ansible / lib / ansible / plugins / inventory / aws_ec2.py View on Github external
def _set_credentials(self):
        '''
            :param config_data: contents of the inventory config file
        '''

        self.boto_profile = self.get_option('aws_profile')
        self.aws_access_key_id = self.get_option('aws_access_key')
        self.aws_secret_access_key = self.get_option('aws_secret_key')
        self.aws_security_token = self.get_option('aws_security_token')
        self.iam_role_arn = self.get_option('iam_role_arn')

        if not self.boto_profile and not (self.aws_access_key_id and self.aws_secret_access_key):
            session = botocore.session.get_session()
            try:
                credentials = session.get_credentials().get_frozen_credentials()
            except AttributeError:
                pass
            else:
                self.aws_access_key_id = credentials.access_key
                self.aws_secret_access_key = credentials.secret_key
                self.aws_security_token = credentials.token

        if not self.boto_profile and not (self.aws_access_key_id and self.aws_secret_access_key):
            raise AnsibleError("Insufficient boto credentials found. Please provide them in your "
                               "inventory configuration file or set them as environment variables.")
github opendatacube / datacube-core / datacube / utils / aws / __init__.py View on Github external
def botocore_default_region(session: Optional[Session] = None) -> Optional[str]:
    """ Returns default region name as configured on the system.
    """
    if session is None:
        session = botocore.session.get_session()
    return session.get_config_variable('region')
github nccgroup / aws-inventory / aws_inventory / invoker.py View on Github external
def _probe_services(self):
        try:
            # create a config for all clients to share
            client_config = botocore.config.Config(
                connect_timeout=config.CLIENT_CONNECT_TIMEOUT,
                read_timeout=config.CLIENT_READ_TIMEOUT
            )

            session = botocore.session.get_session()
            for svc_name in self.svc_descriptors:
                operations = self.svc_descriptors[svc_name]['ops']
                regions = self.svc_descriptors[svc_name]['regions']

                # call each API across each region
                api_version = session.get_config_variable('api_versions').get(svc_name, None)
                params = {'svc_name': svc_name,
                          'dry_run': self.script_args.dry_run,
                          'store': self.store}
                for region in regions:
                    self.progress_bar.update_svc_text(svc_name, region)
                    try:
                        client = session.create_client(
                            svc_name,
                            region_name=region,
                            api_version=api_version,