from temporalio import activity, workflow with workflow.unsafe.imports_passed_through(): import traceback from io import BytesIO from typing import Any import boto3 # type: ignore[import-untyped] from botocore.config import Config # type: ignore[import-untyped] from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler from sientia_do.notifications.models import NotificationLevel from sientia_do.observability.logger import Logger from sientia_do.temporal.activities.base import BaseActivity class MinIO(BaseActivity): """ MinIO (S3-compatible) storage activities for file operations. This class provides activities for interacting with MinIO object storage, including file download and deletion operations. It handles authentication, connection management, and comprehensive error handling. The class implements best practices for S3/MinIO operations: - Connection reuse (boto3 client is thread-safe) - Automatic retry with exponential backoff - Comprehensive error handling and logging - Notification integration for critical errors Attributes: endpoint_url (str): MinIO server endpoint URL access_key (str): MinIO access key ID secret_key (str): MinIO secret access key region (str): MinIO region name use_ssl (bool): Whether to use SSL/TLS for connections minio_client: Boto3 S3 client configured for MinIO """ def __init__( self, endpoint_url: str, access_key: str, secret_key: str, region: str, use_ssl: bool, max_retry_attempts: int, retry_mode: str, connect_timeout: int, read_timeout: int, logger: Logger, notification_handler: NotificationHandler, ): """ Initialize MinIO activities with server configuration. This constructor creates a persistent boto3 S3 client that will be reused across all activity calls. The client is thread-safe and includes automatic retry configuration. Args: endpoint_url: MinIO server endpoint URL (e.g., http://localhost:9000) access_key: MinIO access key ID for authentication secret_key: MinIO secret access key for authentication region: MinIO region name (e.g., us-east-1) use_ssl: Whether to use SSL/TLS for connections max_retry_attempts: Maximum number of retry attempts (e.g., 3) retry_mode: Retry mode - standard, legacy, or adaptive (e.g., adaptive) connect_timeout: Connection timeout in seconds (e.g., 10) read_timeout: Read timeout in seconds (e.g., 60) logger: Logger instance for observability and debugging notification_handler: Notification handler for alerts and monitoring Raises: ConnectionError: If boto3 client initialization fails """ BaseActivity.__init__(self, logger, notification_handler, set_error_counter=True) self.endpoint_url = endpoint_url self.access_key = access_key self.secret_key = secret_key self.region = region self.use_ssl = use_ssl self.max_retry_attempts = max_retry_attempts self.retry_mode = retry_mode self.connect_timeout = connect_timeout self.read_timeout = read_timeout # Configure boto3 with retry strategy # This handles transient network errors and connection issues automatically boto_config = Config( region_name=region, retries={ 'max_attempts': max_retry_attempts, 'mode': retry_mode, }, connect_timeout=connect_timeout, read_timeout=read_timeout, ) try: self.minio_client = boto3.client( 's3', endpoint_url=endpoint_url, aws_access_key_id=access_key, aws_secret_access_key=secret_key, config=boto_config, use_ssl=use_ssl, ) self.info(f'MinIO client initialized successfully: {endpoint_url}') except Exception as e: error_msg = f'Failed to initialize MinIO client: {str(e)}' self.error(error_msg) raise ConnectionError(error_msg) from e @activity.defn(name='fetch_file_from_minio') async def fetch_file_from_minio(self, input_data: dict[str, Any]) -> BytesIO: """ Fetch a file from MinIO and return its content as a BytesIO object. This activity downloads a file from a MinIO bucket and returns the content as a BytesIO object, which is a file-like object that can be used directly with many Python libraries (pandas, PIL, etc.). The operation includes: 1. Input validation 2. File download from MinIO 3. Content reading and wrapping in BytesIO 4. Comprehensive error handling and logging Args: input_data: Configuration for file fetch operation Required keys: - metadata (dict): Workflow execution metadata - bucket_name (str): MinIO bucket name - file_name (str): File path/key in the bucket Returns: BytesIO: File content as a file-like object Raises: OSError: If file fetch fails due to network, permission, or other errors """ metadata = input_data.get('metadata', {}) bucket_name = input_data['bucket_name'] file_name = input_data['file_name'] self.info(f'Fetching file from MinIO: {bucket_name}/{file_name}', metadata) try: # Download file from MinIO response = self.minio_client.get_object(Bucket=bucket_name, Key=file_name) # Read file content with response['Body'] as body: file_content = body.read() file_size = len(file_content) self.info( f'File fetched successfully: {bucket_name}/{file_name} ({file_size} bytes)', metadata, ) return BytesIO(file_content) except Exception as e: # noqa: BLE001 error_msg = f'Error fetching file from MinIO - Bucket: {bucket_name}, File: {file_name}, Error: {str(e)}' trace = traceback.format_exc() self.send_notification( metadata=metadata, notification_id='FETCH_FILE_FROM_MINIO_ERROR', message=error_msg, block='fetch_file_from_minio', level=NotificationLevel.ERROR, attachment_content=trace, ) self.error(trace, metadata=metadata) raise OSError(error_msg) from e @activity.defn(name='delete_file_from_minio') async def delete_file_from_minio(self, input_data: dict[str, Any]) -> None: """ Delete a file from MinIO storage. This activity removes a file from a MinIO bucket. The operation is idempotent - deleting a non-existent file is considered successful. The operation includes: 1. Input validation 2. File deletion from MinIO 3. Comprehensive error handling and logging Args: input_data: Configuration for file deletion operation Required keys: - metadata (dict): Workflow execution metadata - bucket_name (str): MinIO bucket name - file_name (str): File path/key to delete Returns: None Raises: OSError: If file deletion fails due to permission or other errors """ metadata = input_data.get('metadata', {}) bucket_name = input_data['bucket_name'] file_name = input_data['file_name'] self.info(f'Deleting file from MinIO: {bucket_name}/{file_name}', metadata) try: # Delete file from MinIO # Note: delete_object is idempotent - no error if file doesn't exist self.minio_client.delete_object(Bucket=bucket_name, Key=file_name) self.info(f'File deleted successfully: {bucket_name}/{file_name}', metadata) except Exception as e: # noqa: BLE001 error_msg = f'Error deleting file from MinIO - Bucket: {bucket_name}, File: {file_name}, Error: {str(e)}' trace = traceback.format_exc() self.send_notification( metadata=metadata, notification_id='DELETE_FILE_FROM_MINIO_ERROR', message=error_msg, block='delete_file_from_minio', level=NotificationLevel.ERROR, attachment_content=trace, ) self.error(trace, metadata=metadata) raise OSError(error_msg) from e