Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def aiobotocore_client(service, tracer):
"""Helper function that creates a new aiobotocore client so that
it is closed at the end of the context manager.
"""
session = aiobotocore.session.get_session()
endpoint = LOCALSTACK_ENDPOINT_URL[service]
client = session.create_client(
service,
region_name='us-west-2',
endpoint_url=endpoint,
aws_access_key_id='aws',
aws_secret_access_key='aws',
aws_session_token='aws',
)
Pin.override(client, tracer=tracer)
try:
yield client
finally:
client.close()
def __init__(self, aws_access_key_id=None, aws_secret_access_key=None,
aws_session_token=None, region_name=None,
botocore_session=None, profile_name=None, loop=None):
if botocore_session is not None:
self._session = botocore_session
else:
# Create a new default session
self._session = aiobotocore.session.get_session(loop=loop)
# Setup custom user-agent string if it isn't already customized
if self._session.user_agent_name == 'Botocore':
botocore_info = 'Botocore/{0}'.format(
self._session.user_agent_version)
if self._session.user_agent_extra:
self._session.user_agent_extra += ' ' + botocore_info
else:
self._session.user_agent_extra = botocore_info
self._session.user_agent_name = 'Boto3'
self._session.user_agent_version = boto3.__version__
if profile_name is not None:
self._session.set_config_variable('profile', profile_name)
if aws_access_key_id or aws_secret_access_key or aws_session_token: