julee.repositories.minio.client

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

T

Classes

MinioClient	Protocol defining the MinIO client interface used by the repository.
MinioRepositoryMixin	Mixin that provides common repository patterns for Minio implementations.

Module Contents

class julee.repositories.minio.client.MinioClient

Bases: 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.

bucket_exists(bucket_name)

Check if a bucket exists.

Parameters:

bucket_name (str) – Name of the bucket to check

Returns:

True if bucket exists, False otherwise

Return type:

bool

get_object(bucket_name, object_name)

Retrieve an object from the bucket.

Parameters:

• bucket_name (str) – Name of the bucket

• object_name (str) – Name of the object to retrieve

Returns:

HTTPResponse containing the object data

Raises:

S3Error – If object retrieval fails (e.g., NoSuchKey)

Return type:

urllib3.response.BaseHTTPResponse

list_objects(bucket_name, prefix='')

List objects in a bucket with optional prefix filter.

Parameters:

• bucket_name (str) – Name of the bucket

• prefix (str) – Optional prefix to filter objects

Returns:

Iterator or list of objects matching the prefix

Raises:

S3Error – If bucket doesn’t exist or other errors

Return type:

Any

make_bucket(bucket_name)

Create a bucket.

Parameters:

bucket_name (str) – Name of the bucket to create

Raises:

S3Error – If bucket creation fails

put_object(bucket_name, object_name, data, length, content_type='application/octet-stream', metadata=None)

Store an object in the bucket.

Parameters:

• bucket_name (str) – Name of the bucket

• object_name (str) – Name of the object to store

• data (BinaryIO) – Object data (stream or bytes)

• length (int) – Size of the object in bytes

• content_type (str) – MIME type of the object

• metadata (dict[str, str | list[str] | tuple[str]] | None) – Optional metadata dict

Returns:

Object upload result

Raises:

S3Error – If object storage fails

Return type:

minio.api.ObjectWriteResult

stat_object(bucket_name, object_name)

Get object metadata without retrieving the object data.

Parameters:

• bucket_name (str) – Name of the bucket

• object_name (str) – Name of the object

Returns:

Object metadata

Raises:

S3Error – If object doesn’t exist (NoSuchKey) or other errors

Return type:

minio.datatypes.Object

class julee.repositories.minio.client.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__)

ensure_buckets_exist(bucket_names)

Ensure one or more buckets exist, creating them if necessary.

Parameters:

bucket_names (str | list[str]) – Single bucket name or list of bucket names

Raises:

S3Error – If bucket creation fails

generate_id_with_prefix(prefix)

Generate a unique ID with the given prefix and log the generation.

Parameters:

prefix (str) – Prefix for the generated ID (e.g., “ks”, “doc”)

Returns:

Unique ID string in format “{prefix}-{uuid}”

Return type:

str

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.

Parameters:

• bucket_name (str) – Name of the bucket

• object_name (str) – Name of the object

• model_class (type[T]) – Pydantic model class to deserialize to

• not_found_log_message (str) – Message to log when object is not found

• error_log_message (str) – Message to log on other errors

• extra_log_data (dict[str, Any] | None) – Additional data to include in log entries

Returns:

Deserialized Pydantic model instance, or None if not found

Raises:

S3Error – For non-NoSuchKey errors

Return type:

T | None

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.

Parameters:

• bucket_name (str) – Name of the bucket

• object_names (list[str]) – List of object names to retrieve

• not_found_log_message (str) – Message to log when objects are not found

• error_log_message (str) – Message to log on other errors

• extra_log_data (dict[str, Any] | None) – 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

Return type:

dict[str, julee.domain.models.custom_fields.content_stream.ContentStream | None]

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.

Parameters:

• bucket_name (str) – Name of the bucket

• object_names (list[str]) – List of object names to retrieve

• model_class (type[T]) – Pydantic model class to deserialize to

• not_found_log_message (str) – Message to log when objects are not found

• error_log_message (str) – Message to log on other errors

• extra_log_data (dict[str, Any] | None) – 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

Return type:

dict[str, T | None]

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.

Parameters:

• bucket_name (str) – Name of the bucket to list objects from

• prefix (str) – Object name prefix to filter by (e.g., “spec/”, “query/”)

• entity_type_name (str) – Name for logging (e.g., “specs”, “queries”)

Returns:

List of entity IDs extracted from object names

Raises:

Exception – If listing objects fails

Return type:

list[str]

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.

Parameters:

• bucket_name (str) – Name of the bucket

• object_name (str) – Name of the object

• model (pydantic.BaseModel) – Pydantic model instance to serialize

• success_log_message (str) – Message to log on successful storage

• error_log_message (str) – Message to log on error

• extra_log_data (dict[str, Any] | None) – Additional data to include in log entries

Raises:

S3Error – If object storage fails

update_timestamps(model)

Update timestamps on a model (created_at if None, always updated_at).

Parameters:

model (Any) – Pydantic model with created_at and updated_at fields

client: MinioClient

logger: Any

julee.repositories.minio.client.T