get_aws_objects.py

import boto3.resources.base
import botocore.client
from boto3 import Session

from get_datafile_credentials import get_datafile_credentials


def get_aws_session(
    username: str, password: str, environment: str = "stage"
) -> boto3.Session | None:
    """Returns a Boto3 Session object configured with AWS credentials for accessing the DataCite Monthly Data File.

    :param username: Your DataCite username
    :param password: Your DataCite password
    :param environment: The target environment, either "stage" or None (for production)
    :rtype: boto3.Session (if successful) | None (if an error)
    """
    aws_credentials = get_datafile_credentials(username, password, environment)

    if not aws_credentials:
        return None

    session = Session(
        aws_access_key_id=aws_credentials[0],
        aws_secret_access_key=aws_credentials[1],
        aws_session_token=aws_credentials[2],
        region_name="eu-west-1",
    )

    return session if session else None


def get_client(
    username: str, password: str, environment: str = "stage"
) -> botocore.client.BaseClient | None:
    """Returns a Boto3 client object for S3, configured with AWS credentials for accessing the DataCite Monthly Data File.

    :param username: Your DataCite username
    :param password: Your DataCite password
    :param environment: The target environment, either "stage" or None (for production)
    :rtype: client.S3 (if successful) | None (if an error)
    """
    session = get_aws_session(username, password, environment)

    if not session:
        return None

    aws_client = session.client("s3")

    return aws_client if aws_client else None


def get_bucket(
    username: str, password: str, environment: str = "stage"
) -> boto3.resources.base.ServiceResource | None:
    """Returns a Boto3 S3 Bucket resource, configured with AWS credentials for accessing the DataCite Monthly Data File and the appropriate S3 bucket.

    :param username: Your DataCite username
    :param password: Your DataCite password
    :param environment: The target environment, either "stage" or None (for production)
    :rtype: s3.Bucket (if successful) | None (if an error)
    """
    session = get_aws_session(username, password, environment)

    if not session:
        return None

    bucket = (
        f"monthly-datafile{'.' + environment + '.' if environment else '.'}datacite.org"
    )

    s3_resource = session.resource("s3")
    bucket_resource = s3_resource.Bucket(bucket)

    return bucket_resource if bucket_resource else None