File size: 2,798 Bytes
e47481b df6c855 7bfc491 e47481b 98e87f9 e47481b |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 |
import os
from io import BytesIO
import boto3
import requests
DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
BEDROCK_MODEL_ID = "anthropic.claude-3-5-sonnet-20240620-v1:0"
OPENAI_MODEL_ID = "o4-mini"
bedrock_runtime = boto3.client("bedrock-runtime", region_name=os.getenv("AWS_REGION"))
def get_file(task_id: str) -> BytesIO:
"""Fetches a file associated with a task ID from the default API URL.
Parameters
----------
task_id : str
The ID of the task for which the file is to be fetched.
Returns
-------
BytesIO
A BytesIO object containing the file content.
Raises
------
requests.exceptions.RequestException
If there is an error during the HTTP request.
Exception
For any other unexpected errors that may occur.
"""
url = f"{DEFAULT_API_URL}/files/{task_id}"
try:
response = requests.get(url, timeout=15)
response.raise_for_status()
return BytesIO(response.content)
except requests.exceptions.RequestException as e:
print(f"Error fetching file for task {task_id}: {e}")
raise
except Exception as e:
print(f"An unexpected error occurred fetching file for task {task_id}: {e}")
raise
def s3_upload_file(file_content: BytesIO, bucket_name: str, object_name: str) -> None:
"""Uploads a file to an S3 bucket.
Parameters
----------
file_content : BytesIO
The content of the file to upload.
bucket_name : str
The name of the S3 bucket.
object_name : str
The name of the object in the S3 bucket.
Raises
------
Exception
If there is an error during the upload process.
"""
try:
s3_client = boto3.client("s3")
s3_client.put_object(
Bucket=bucket_name,
Key=object_name,
Body=file_content.getvalue(),
ContentType="application/octet-stream",
)
except Exception as e:
print(f"Error uploading file to S3: {e}")
raise
def s3_download_file(bucket_name: str, object_name: str) -> BytesIO:
"""Downloads a file from an S3 bucket.
Parameters
----------
bucket_name : str
The name of the S3 bucket.
object_name : str
The name of the object in the S3 bucket.
Returns
-------
BytesIO
A BytesIO object containing the downloaded file content.
Raises
------
Exception
If there is an error during the download process.
"""
try:
s3_client = boto3.client("s3")
response = s3_client.get_object(Bucket=bucket_name, Key=object_name)
return BytesIO(response["Body"].read())
except Exception as e:
print(f"Error downloading file from S3: {e}")
raise
|