role_name = os.getenv('KINESIS_ROLE_ARN')
session_name = 'kinesis_session'
sts_client = boto3.client('sts')
auto_refresh_session = _create_refreshable_session()
kinesis_client = auto_refresh_session.client('kinesis')
kinesis_stream_name = "Log"
def _refresh_kinesis_role():
params = {
'RoleArn': role_name,
'RoleSessionName': session_name,
'DurationSeconds': 3600,
}
response = sts_client.assume_role(**params).get('Credentials')
credentials = {
'access_key': response.get('AccessKeyId'),
'secret_key': response.get('SecretAccessKey'),
'token': response.get('SessionToken'),
'expiry_time': response.get('Expiration').isoformat(),
}
return credentials
def _create_refreshable_session():
session_credentials = RefreshableCredentials.create_from_metadata(
metadata=_refresh_kinesis_role(),
refresh_using=_refresh_kinesis_role,
method='sts-assume-role',
)
session = get_session()
session._credentials = session_credentials
autorefresh_session = boto3.Session(botocore_session=session)
return autorefresh_session
Я пытаюсь создать сеанс автоматического обновления, чтобы взять на себя роль AWS, следуя приведенному здесь руководству: https://dev.to/li_chastina/auto-refresh-aws-tokens-using-iam-role-and-boto3-2cjf.
Я получаю ошибку неразрешенной ссылки на _create_refreshable_session () в строке 4Я вижу аналогичный вопрос об этом на stackoverflow, но я не совсем понимаю ответ в этом. Кажется, это проблема масштаба. Этот ответ в основном говорит о том, что вам нужно определить эту функцию, прежде чем использовать ее на верхнем уровне? Есть ли способ решить эту проблему, следуя порядку кода?
В руководстве автор использовал self для этих двух функций, но, похоже, не включил их в какой-либо класс, который меня смущает, поэтомуЯ удалил собственную часть, я не знаю, вызывает ли это какую-либо ошибку.