lazy_s3_client.py

import time
import keyring
import keyring.errors
from getpass import getpass
import json
import boto3


class LazyS3Client:
    def __init__(self):
        self._wrapped = None

    def __getattr__(self, name):        
        if not self._wrapped:
            infra_auth_output = json.loads(LazyS3Client.get_creds("aws").get("aws"))
            aws_session = boto3.session.Session(
                aws_access_key_id=infra_auth_output["AccessKeyId"],
                aws_secret_access_key=infra_auth_output["SecretAccessKey"],
                aws_session_token=infra_auth_output["SessionToken"],
            )
            self._wrapped = aws_session.client("s3")

        # Delegate attribute access to the wrapped object
        attr = getattr(self._wrapped, name)
        if callable(attr):
            # If the attribute is callable, return a method that calls the original method
            def method(*args, **kwargs):
                result = attr(*args, **kwargs)
                return result
            return method
        else:
            return attr
    
    def get_creds(*args, hours=4):
        """Caches you frequently used credentials in macOS keychain"""

        def generate_key(shift = 0):
            # Get the current time in seconds since the epoch
            current_time = int(time.time())    
            # Calculate the number of the hours intervals since the epoch
            interval = (current_time // (hours * 3600)) + shift
            return str(interval)
        
        creds = dict()

        key = generate_key()
        for name in args:
            secret = keyring.get_password(key, name)
            if secret is None:
                # clean up keys for the last 80 hours
                for i in range(1, 20):
                    prev_key = generate_key(-i)
                    try:
                        keyring.delete_password(prev_key, secret)
                    except keyring.errors.PasswordDeleteError:
                        pass

                secret = getpass(f"Provide secret for '{name}' to be cached in macOS keychain for 4 hours")
                keyring.set_password(key, name, secret)

            creds.update({ name: secret })
        return creds