Collections# Overview#. A collection provides an iterable interface to a group of resources. Collections behave similarly to Django QuerySets and expose a similar API. A collection seamlessly handles pagination for you, making it possible to easily iterate over all items from all pages of data.
Paginators are created via the get_paginator() method of a boto3 client. The get_paginator() method accepts an operation name and returns a reusable Paginator object. You then call the paginate method of the Paginator, passing in any relevant operation parameters to apply to the underlying API operation.
You can create a session:
import boto3
session = boto3.Session(
aws_access_key_id=settings.AWS_SERVER_PUBLIC_KEY,
aws_secret_access_key=settings.AWS_SERVER_SECRET_KEY,
)
Then use that session to get an S3 resource:
s3 = session.resource('s3')
I have found a good example to refresh the credentials within this link: https://pritul95.github.io/blogs/boto3/2020/08/01/refreshable-boto3-session/
but there this a little bug inside. Be careful about that.
Here is the corrected code:
from uuid import uuid4
from datetime import datetime
from time import time
from boto3 import Session
from botocore.credentials import RefreshableCredentials
from botocore.session import get_session
class RefreshableBotoSession:
"""
Boto Helper class which lets us create refreshable session, so that we can cache the client or resource.
Usage
-----
session = RefreshableBotoSession().refreshable_session()
client = session.client("s3") # we now can cache this client object without worrying about expiring credentials
"""
def __init__(
self,
region_name: str = None,
profile_name: str = None,
sts_arn: str = None,
session_name: str = None,
session_ttl: int = 3000
):
"""
Initialize `RefreshableBotoSession`
Parameters
----------
region_name : str (optional)
Default region when creating new connection.
profile_name : str (optional)
The name of a profile to use.
sts_arn : str (optional)
The role arn to sts before creating session.
session_name : str (optional)
An identifier for the assumed role session. (required when `sts_arn` is given)
session_ttl : int (optional)
An integer number to set the TTL for each session. Beyond this session, it will renew the token.
50 minutes by default which is before the default role expiration of 1 hour
"""
self.region_name = region_name
self.profile_name = profile_name
self.sts_arn = sts_arn
self.session_name = session_name or uuid4().hex
self.session_ttl = session_ttl
def __get_session_credentials(self):
"""
Get session credentials
"""
session = Session(region_name=self.region_name, profile_name=self.profile_name)
# if sts_arn is given, get credential by assuming given role
if self.sts_arn:
sts_client = session.client(service_name="sts", region_name=self.region_name)
response = sts_client.assume_role(
RoleArn=self.sts_arn,
RoleSessionName=self.session_name,
DurationSeconds=self.session_ttl,
).get("Credentials")
credentials = {
"access_key": response.get("AccessKeyId"),
"secret_key": response.get("SecretAccessKey"),
"token": response.get("SessionToken"),
"expiry_time": response.get("Expiration").isoformat(),
}
else:
session_credentials = session.get_credentials().__dict__
credentials = {
"access_key": session_credentials.get("access_key"),
"secret_key": session_credentials.get("secret_key"),
"token": session_credentials.get("token"),
"expiry_time": datetime.fromtimestamp(time() + self.session_ttl).isoformat(),
}
return credentials
def refreshable_session(self) -> Session:
"""
Get refreshable boto3 session.
"""
# get refreshable credentials
refreshable_credentials = RefreshableCredentials.create_from_metadata(
metadata=self.__get_session_credentials(),
refresh_using=self.__get_session_credentials,
method="sts-assume-role",
)
# attach refreshable credentials current session
session = get_session()
session._credentials = refreshable_credentials
session.set_config_variable("region", self.region_name)
autorefresh_session = Session(botocore_session=session)
return autorefresh_session
Such a credential-validating method does exist; it's the STS GetCallerIdentity API call (boto3 method docs).
With expired temporary credentials:
>>> import boto3
>>> sts = boto3.client('sts')
>>> sts.get_caller_identity()
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "/home/jantman/venv/lib/python3.8/site-packages/botocore/client.py", line 276, in _api_call
return self._make_api_call(operation_name, kwargs)
File "/home/jantman/venv/lib/python3.8/site-packages/botocore/client.py", line 586, in _make_api_call
raise error_class(parsed_response, operation_name)
botocore.exceptions.ClientError: An error occurred (ExpiredToken) when calling the GetCallerIdentity operation: The security token included in the request is expired
With invalid credentials:
>>> import boto3
>>> sts = boto3.client('sts')
>>> sts.get_caller_identity()
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "/home/jantman/venvs/current/lib/python3.8/site-packages/botocore/client.py", line 316, in _api_call
return self._make_api_call(operation_name, kwargs)
File "/home/jantman/venvs/current/lib/python3.8/site-packages/botocore/client.py", line 626, in _make_api_call
raise error_class(parsed_response, operation_name)
botocore.exceptions.ClientError: An error occurred (InvalidClientTokenId) when calling the GetCallerIdentity operation: The security token included in the request is invalid
With valid credentials (IDs replaced with X's):
>>> import boto3
>>> sts = boto3.client('sts')
>>> sts.get_caller_identity()
{'UserId': 'AROAXXXXXXXXXXXXX:XXXXXXX', 'Account': 'XXXXXXXXXXXX', 'Arn': 'arn:aws:sts::XXXXXXXXXXXX:assumed-role/Admin/JANTMAN', 'ResponseMetadata': {'RequestId': 'f44ec1d9-XXXX-XXXX-XXXX-a26c85be1c60', 'HTTPStatusCode': 200, 'HTTPHeaders': {'x-amzn-requestid': 'f44ec1d9-XXXX-XXXX-XXXX-a26c85be1c60', 'content-type': 'text/xml', 'content-length': '426', 'date': 'Thu, 28 May 2020 10:45:36 GMT'}, 'RetryAttempts': 0}}
Invalid credentials will raise an exception and valid credentials won't, so you can do something such as:
import boto3
sts = boto3.client('sts')
try:
sts.get_caller_identity()
print("Credentials are valid.")
except boto3.exceptions.ClientError:
print("Credentials are NOT valid.")
Yes. Your credentials are used to sign all the requests you send out, so what you have to do is configure the client to not perform the signing step at all. You can do that as follows:
import boto3
from botocore import UNSIGNED
from botocore.client import Config
s3 = boto3.client('s3', config=Config(signature_version=UNSIGNED))
# Use the client
An alternative to hardcoding your Access Key ID and Secret Access Key would be to use Amazon Cognito Identity Pools, which generate roles with a set of permissions. I would recommend you to look into the GetId and GetCredentialsForIdentity API calls in Boto3.