230 lines
8.8 KiB
Python
230 lines
8.8 KiB
Python
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
|