File size: 3,953 Bytes
2eee82e | 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 104 105 106 107 | import os
from minio import Minio
from minio.error import S3Error
from dotenv import load_dotenv
def sign_in():
"""Initialize and return a MinIO client using environment variables."""
# Load environment variables from minio.env file
load_dotenv("minio.env")
# Retrieve MinIO credentials and host from environment variables
minio_host = os.getenv("MINIO_HOST")
minio_root_user = os.getenv("MINIO_ROOT_USER")
minio_root_password = os.getenv("MINIO_ROOT_PASSWORD")
# Check if environment variables are set
if not all([minio_host, minio_root_user, minio_root_password]):
raise ValueError("Missing required environment variables: MINIO_HOST, MINIO_ROOT_USER, or MINIO_ROOT_PASSWORD")
# Initialize MinIO client
try:
minio_client = Minio(
endpoint=minio_host,
access_key=minio_root_user,
secret_key=minio_root_password,
secure=False # Set to False if not using HTTPS
)
print("Successfully connected to MinIO server")
return minio_client
except S3Error as err:
print(f"Failed to connect to MinIO server: {err}")
raise
def create_bucket(minio_client, bucket_name):
"""Create a bucket if it doesn't exist."""
try:
found = minio_client.bucket_exists(bucket_name)
if not found:
minio_client.make_bucket(bucket_name)
print(f"Created bucket: {bucket_name}")
else:
print(f"Bucket {bucket_name} already exists")
return True
except S3Error as err:
print(f"Error creating bucket: {err}")
return False
def upload_file(minio_client, bucket_name, client_file, server_file):
"""Upload a file to the specified bucket."""
try:
minio_client.fput_object(bucket_name, server_file, client_file)
print(f"{client_file} successfully uploaded as {server_file} to bucket {bucket_name}")
return True
except S3Error as err:
print(f"Error uploading file: {err}")
return False
def download_file(minio_client, bucket_name, object_name, destination_path):
"""Download a file from the specified bucket."""
try:
minio_client.fget_object(bucket_name, object_name, destination_path)
print(f"{object_name} successfully downloaded to {destination_path}")
return True
except S3Error as err:
print(f"Error downloading file: {err}")
return False
def list_objects(minio_client, bucket_name):
"""List all objects in the specified bucket."""
try:
objects = minio_client.list_objects(bucket_name, recursive=True)
object_list = [obj.object_name for obj in objects]
if object_list:
print(f"Objects in bucket {bucket_name}: {object_list}")
else:
print(f"No objects found in bucket {bucket_name}")
return object_list
except S3Error as err:
print(f"Error listing objects: {err}")
return []
# Example usage
if __name__ == "__main__":
# Configuration
bucket_name = "minio-ngrok-bucket"
client_file = "/content/sample_data/california_housing_test.csv"
server_file = "california_housing_test.csv"
download_path = "/content/sample_data/donload_california_housing_test.csv"
# Initialize MinIO client
try:
minio_client = sign_in()
# Create bucket
create_bucket(minio_client, bucket_name)
print("Bucket creation checked/attempted.")
# Upload file
upload_file(minio_client, bucket_name, client_file, server_file)
print("File upload attempted.")
# List objects in bucket
list_objects(minio_client, bucket_name)
print("Listed objects in bucket.")
# Download file
download_file(minio_client, bucket_name, server_file, download_path)
print("File download attempted.")
except Exception as e:
print(f"An error occurred: {e}") |