julee.repositories.minio.client =============================== .. py:module:: julee.repositories.minio.client .. autoapi-nested-parse:: MinioClient protocol definition and repository utilities. This module defines the protocol interface that both the real Minio client and our fake test client must implement. This follows Clean Architecture dependency inversion principles by depending on abstractions rather than concrete implementations. It also provides MinioRepositoryMixin, a mixin that encapsulates common patterns used across all Minio repository implementations to reduce code duplication and ensure consistent error handling and logging. Attributes ---------- .. autoapisummary:: julee.repositories.minio.client.T Classes ------- .. autoapisummary:: julee.repositories.minio.client.MinioClient julee.repositories.minio.client.MinioRepositoryMixin Module Contents --------------- .. py:class:: MinioClient Bases: :py:obj:`Protocol` Protocol defining the MinIO client interface used by the repository. This protocol captures only the methods we actually use, making our dependency explicit and testable. Both the real minio.Minio client and our FakeMinioClient implement this protocol. .. py:method:: bucket_exists(bucket_name) Check if a bucket exists. :param bucket_name: Name of the bucket to check :returns: True if bucket exists, False otherwise .. py:method:: get_object(bucket_name, object_name) Retrieve an object from the bucket. :param bucket_name: Name of the bucket :param object_name: Name of the object to retrieve :returns: HTTPResponse containing the object data :raises S3Error: If object retrieval fails (e.g., NoSuchKey) .. py:method:: list_objects(bucket_name, prefix = '') List objects in a bucket with optional prefix filter. :param bucket_name: Name of the bucket :param prefix: Optional prefix to filter objects :returns: Iterator or list of objects matching the prefix :raises S3Error: If bucket doesn't exist or other errors .. py:method:: make_bucket(bucket_name) Create a bucket. :param bucket_name: Name of the bucket to create :raises S3Error: If bucket creation fails .. py:method:: put_object(bucket_name, object_name, data, length, content_type = 'application/octet-stream', metadata = None) Store an object in the bucket. :param bucket_name: Name of the bucket :param object_name: Name of the object to store :param data: Object data (stream or bytes) :param length: Size of the object in bytes :param content_type: MIME type of the object :param metadata: Optional metadata dict :returns: Object upload result :raises S3Error: If object storage fails .. py:method:: stat_object(bucket_name, object_name) Get object metadata without retrieving the object data. :param bucket_name: Name of the bucket :param object_name: Name of the object :returns: Object metadata :raises S3Error: If object doesn't exist (NoSuchKey) or other errors .. py:class:: MinioRepositoryMixin Mixin that provides common repository patterns for Minio implementations. This mixin encapsulates common functionality used across all Minio repository implementations, including: - Bucket creation and management - JSON serialization/deserialization with proper error handling - Standardized S3Error handling for NoSuchKey cases - Consistent logging patterns - Response cleanup - ID generation with logging Classes using this mixin must provide: - self.client: MinioClient instance - self.logger: logging.Logger instance (typically set in __init__) .. py:method:: ensure_buckets_exist(bucket_names) Ensure one or more buckets exist, creating them if necessary. :param bucket_names: Single bucket name or list of bucket names :raises S3Error: If bucket creation fails .. py:method:: generate_id_with_prefix(prefix) Generate a unique ID with the given prefix and log the generation. :param prefix: Prefix for the generated ID (e.g., "ks", "doc") :returns: Unique ID string in format "{prefix}-{uuid}" .. py:method:: get_json_object(bucket_name, object_name, model_class, not_found_log_message, error_log_message, extra_log_data = None) Get a JSON object from Minio and deserialize it to a Pydantic model. :param bucket_name: Name of the bucket :param object_name: Name of the object :param model_class: Pydantic model class to deserialize to :param not_found_log_message: Message to log when object is not found :param error_log_message: Message to log on other errors :param extra_log_data: Additional data to include in log entries :returns: Deserialized Pydantic model instance, or None if not found :raises S3Error: For non-NoSuchKey errors .. py:method:: get_many_binary_objects(bucket_name, object_names, not_found_log_message, error_log_message, extra_log_data = None) Get multiple binary objects from Minio as ContentStreams. Note: S3/MinIO does not have native batch retrieval operations. This method makes individual GetObject calls for each object but provides consolidated error handling, logging, and connection reuse. :param bucket_name: Name of the bucket :param object_names: List of object names to retrieve :param not_found_log_message: Message to log when objects are not found :param error_log_message: Message to log on other errors :param extra_log_data: Additional data to include in log entries :returns: Dict mapping object_name to ContentStream (or None if not found) :raises S3Error: For non-NoSuchKey errors .. py:method:: get_many_json_objects(bucket_name, object_names, model_class, not_found_log_message, error_log_message, extra_log_data = None) Get multiple JSON objects from Minio and deserialize them. Note: S3/MinIO does not have native batch retrieval operations. This method makes individual GetObject calls for each object but provides consolidated error handling, logging, and connection reuse. The real benefit comes with other backends (PostgreSQL, Redis, etc.) that support true batch operations. :param bucket_name: Name of the bucket :param object_names: List of object names to retrieve :param model_class: Pydantic model class to deserialize to :param not_found_log_message: Message to log when objects are not found :param error_log_message: Message to log on other errors :param extra_log_data: Additional data to include in log entries :returns: Dict mapping object_name to deserialized model (or None if not found) :raises S3Error: For non-NoSuchKey errors .. py:method:: list_objects_with_prefix_extract_ids(bucket_name, prefix, entity_type_name) Extract entity IDs from objects with a given prefix. This method provides a common implementation for listing objects and extracting IDs, eliminating code duplication in list_all methods. :param bucket_name: Name of the bucket to list objects from :param prefix: Object name prefix to filter by (e.g., "spec/", "query/") :param entity_type_name: Name for logging (e.g., "specs", "queries") :returns: List of entity IDs extracted from object names :raises Exception: If listing objects fails .. py:method:: put_json_object(bucket_name, object_name, model, success_log_message, error_log_message, extra_log_data = None) Store a Pydantic model as a JSON object in Minio. :param bucket_name: Name of the bucket :param object_name: Name of the object :param model: Pydantic model instance to serialize :param success_log_message: Message to log on successful storage :param error_log_message: Message to log on error :param extra_log_data: Additional data to include in log entries :raises S3Error: If object storage fails .. py:method:: update_timestamps(model) Update timestamps on a model (created_at if None, always updated_at). :param model: Pydantic model with created_at and updated_at fields .. py:attribute:: client :type: MinioClient .. py:attribute:: logger :type: Any .. py:data:: T