Let's say I am running k8s cluster on AWS (not EKS). I want applications on the k8s pods to assume various Roles, obtain temporary credentials, and access AWS Resources. Here is the sample code of my test application.

import boto3, os

def get_web_token():
    path = os.getenv("AWS_WEB_IDENTITY_TOKEN_FILE")
    with open(path) as f:
        lines = f.readlines()
    return lines[0]

def get_role():
    return os.getenv("AWS_ROLE_ARN")

if __name__ == "__main__":
    client = boto3.client('sts')
    web_token = get_web_token()
    role_arn = get_role()
    response = client.assume_role_with_web_identity(
        DurationSeconds=3600,
        RoleArn=role_arn,
        RoleSessionName='app1',
        WebIdentityToken=web_token
    )

    credentials = response['Credentials']

    s3_resource = boto3.resource(
        's3',
        aws_access_key_id=credentials['AccessKeyId'],
        aws_secret_access_key=credentials['SecretAccessKey'],
        aws_session_token=credentials['SessionToken'],
    )

    for bucket in s3_resource.buckets.all():
        print(bucket.name)

However, this is a toy example and I want to move all production Python services to IRSA.

In real life, one of the production services would initialize a connection in the class constructor so it is practically impossible to tell how long this connection has to be valid and credentials will expire if the application is running without any restarts.

class S3StorageClient:

    def __init__(self):
        self.bucket = get_bucket_name()
        self.s3_client = boto3.client(
            "s3",
            aws_access_key_id=SECRETS_S3_ACCESS_KEY_ID,
            aws_secret_access_key=SECRETS_S3_SECRET_ACCESS_KEY,
        )

What is the best practice to handle credentials expiry?

Should I catch AccessDenied exception on each boto3 library call and reinitialize connection and retry? But how I would tell the issue is credentials expiry, not some other problem that blocks access to a resource?

Is there any other approach that I am missing?