Boto3 wrappers.
Boto3 is a rather low-level library to interact with AWS services. I don't find it very fun to use and this feeling seems to be shared, given that I have seen so many projects that write their own wrappers on top. I haven't seen one yet that I like.
The main issue is usually session management, or the lack thereof. Boto3 has an intricate sequence of authentication methods it tries and in my opinion it's just not worth it to depend on this implicit behavior. You always end up wanting to influence it somehow, be it becaues you run some code locally or debugging some issues in the cloud. Thus, authenetication should be an explicit action in your program.
So, I want the wrappre to:
- Easy to use in a notebook and in production code.
- Possibility to:
- Set a default session, as a singleton, and any function you call uses it.
- Explicitly pass a session.
- Set a new session, any function you call uses that one.
I think I found an interface I like for the first two cases. It only does session caching; for production you'd probably also want to cache clients.
import dataclasses
import threading
import typing
import boto3
@dataclasses.dataclass(kw_only=True)
class AWSSession:
name: str
kwargs: dict[str, typing.Any]
session: boto3.Session
_session_lock = threading.Lock()
_sessions: dict[str, AWSSession] = dict()
def get_session(session_name: str="", raise_if_not_exists: bool=False, **kwargs) -> boto3.Session:
session_name = session_name or "default"
try:
aws_session = _sessions[session_name]
except KeyError as e:
if raise_if_not_exists:
e.add_note(f"AWS session `{session_name}` not found.")
raise
with _session_lock:
try:
aws_session = _sessions[session_name]
except KeyError:
aws_session = AWSSession(
name=session_name,
kwargs=kwargs,
session=boto3.Session(**kwargs),
)
_sessions[session_name] = aws_session
return aws_session.session
def get_existing_session(session_name: str="") -> boto3.Session:
return get_session(session_name=session_name, raise_if_not_exists=True)
def set_default_session(**kwargs) -> None:
# Calling any boto3 function will use this session.
boto3.setup_default_session(**kwargs)
# Set up our own default session.
get_session(**kwargs)
def get_secret(secret_name: str, session_name: str = "") -> str:
session = get_existing_session(session_name=session_name)
# Do stuff with the session.
return "foobar"
class S3Stuff:
def some_method(self, session_name: str="")
return 42
For the third use case, I haven't found a completely satisfactory solution, yet. My current idea is to use a wrapper class:
class AWSSessionWrapper:
"""
Wraps all functions that take a session_name and passes the configured session, instead.
"""
def __init__(self, session_name: str = "", **kwargs):
self.session_name = session_name
# Make sure the session exists
get_session(session_name=self.session_name, **kwargs)
that internally calls all functions/classes/modules with the session_name parameter set in the class.
My main issue here is that I don't want to manually create a wrapper method for each function.
Instead:
- Introduce a decorator and decorate all functions/classes with
@takes_session_nameor similar. - Add a
__getattr__dunder method to the wrapper class. When we callAWSSessionWrapper(session_name="foobar").get_secret(...)it checks if there is a decorated function of that name and then passes thesession_nameto it.
I need to make sure that this works for functions, classes and modules.
The downside will be: static checkers/code completion will not understand what methods are available on the AWSSessionWrapper obect.
Ideas welcome.