使用awscli时,~/.aws/cli/cache中有一个凭据缓存,可以让我缓存凭据一段时间。这在使用MFA时非常有帮助。boto3是否具有类似的功能,还是我必须显式地缓存从 session = boto3.session.Session(profile_name='CTO:Admin')
返回的凭据?
使用awscli时,~/.aws/cli/cache中有一个凭据缓存,可以让我缓存凭据一段时间。这在使用MFA时非常有帮助。boto3是否具有类似的功能,还是我必须显式地缓存从 session = boto3.session.Session(profile_name='CTO:Admin')
返回的凭据?
已经存在。
http://boto3.readthedocs.org/en/latest/guide/configuration.html#assume-role-provider
当你指定一个包含IAM角色配置的配置文件时,boto3会发出AssumeRole调用来检索临时凭证。后续的boto3 API调用将使用缓存的临时凭证,直到它们过期,此时boto3将自动刷新凭证。boto3不会将这些临时凭证写入磁盘。这意味着AssumeRole调用中的临时凭证仅在单个Session内在内存中缓存。从该会话创建的所有客户端将共享相同的临时凭证。
import boto3_session_cache
# This returns a regular boto3 client object with the underlying session configured with local credential cache
client = boto3_session_cache.client('ecs')
ecs_clusters = client.list_clusters()
from os import path
import os
import sys
import json
import datetime
from distutils.spawn import find_executable
from botocore.exceptions import ProfileNotFound
import boto3
import botocore
def json_encoder(obj):
"""JSON encoder that formats datetimes as ISO8601 format."""
if isinstance(obj, datetime.datetime):
return obj.isoformat()
else:
return obj
class JSONFileCache(object):
"""JSON file cache.
This provides a dict like interface that stores JSON serializable
objects.
The objects are serialized to JSON and stored in a file. These
values can be retrieved at a later time.
"""
CACHE_DIR = path.expanduser(path.join('~', '.aws', 'ansible-ec2', 'cache'))
def __init__(self, working_dir=CACHE_DIR):
self._working_dir = working_dir
def __contains__(self, cache_key):
actual_key = self._convert_cache_key(cache_key)
return path.isfile(actual_key)
def __getitem__(self, cache_key):
"""Retrieve value from a cache key."""
actual_key = self._convert_cache_key(cache_key)
try:
with open(actual_key) as f:
return json.load(f)
except (OSError, ValueError, IOError):
raise KeyError(cache_key)
def __setitem__(self, cache_key, value):
full_key = self._convert_cache_key(cache_key)
try:
file_content = json.dumps(value, default=json_encoder)
except (TypeError, ValueError):
raise ValueError("Value cannot be cached, must be "
"JSON serializable: %s" % value)
if not path.isdir(self._working_dir):
os.makedirs(self._working_dir)
with os.fdopen(os.open(full_key,
os.O_WRONLY | os.O_CREAT, 0o600), 'w') as f:
f.truncate()
f.write(file_content)
def _convert_cache_key(self, cache_key):
full_path = path.join(self._working_dir, cache_key + '.json')
return full_path
session = boto3.session.Session()
try:
cred_chain = session._session.get_component('credential_provider')
except ProfileNotFound:
print "Invalid Profile"
sys.exit(1)
provider = cred_chain.get_provider('assume-role')
provider.cache = JSONFileCache()
# Do something with the session...
ec2 = session.resource('ec2')
datetime.datetime
的自定义 JSON 编码。从 awscli/utils.py 中添加 json_encode
函数,并将其添加为默认编码器 json.dumps(value, default=json_encoder)
。 - Rickard von Essen