yt-dlp-stream-to-s3/yt_dlp_stream_to_s3/client.py

import boto3
from botocore.exceptions import ClientError
from yt_dlp_stream_to_s3.config import (
    AWS_ENDPOINT_URL,
    AWS_SECRET_ACCESS_KEY,
    AWS_ACCESS_KEY_ID,
)
from yt_dlp_stream_to_s3.errors import YtDlpStreamToS3ConnectionError


try:
    s3 = boto3.client(
        "s3",
        endpoint_url=AWS_ENDPOINT_URL,
        aws_secret_access_key=AWS_SECRET_ACCESS_KEY,
        aws_access_key_id=AWS_ACCESS_KEY_ID,
    )

except Exception:
    raise YtDlpStreamToS3ConnectionError(
        f"Could not connect to S3 service {AWS_ENDPOINT_URL}"
    )


def create_presigned_url(bucket_name, object_name, expiration=3600):
    return s3.generate_presigned_url(
        "get_object",
        Params={"Bucket": bucket_name, "Key": object_name},
        ExpiresIn=expiration,
    )