MinIO Object Storage Client

MinIO Object Storage Client

Problem Statement

Managing backups and file archives across a fleet of servers usually means either shelling out to aws s3 cp (which requires AWS credentials and doesn't work with self-hosted S3-compatible stores) or writing one-off boto3 scripts that get copy-pasted between projects. MinIO is the de-facto S3-compatible object store for on-prem and hybrid deployments, but its Python client is underused compared to boto3 even though it's lighter and has a cleaner API for common operations.

This script gives you a single, self-contained CLI for the operations you actually do day-to-day against a MinIO (or any S3-compatible) endpoint: uploading a directory tree, listing objects, downloading, and applying lifecycle-style expiration to uploaded files. It's designed to be dropped into a cron job, a CI pipeline, or a Kubernetes CronJob and configured entirely through environment variables or flags.

How It Works

The script uses the official minio Python SDK, which speaks the S3 protocol and works against MinIO, AWS S3, Wasabi, Backblaze B2's S3 endpoint, and similar services. On startup it builds a Minio client from either CLI flags or environment variables, verifies the bucket exists (creating it if --create-bucket is passed), and then dispatches to one of four subcommands: upload, download, list, or delete.

For uploads of a directory, the script walks the tree with os.walk, computes the object key by joining the destination prefix with the file's relative path (using forward slashes regardless of OS), and calls fput_object. The --expire flag maps a human-friendly duration (30d, 12h, 90m) to a lifecycle rule applied to the target prefix via set_bucket_lifecycle. Errors from the SDK (S3Error, MinioException) are caught and turned into a non-zero exit with a readable message so cron jobs and CI runners can detect failure.

Full Script

#!/usr/bin/env python3
"""
MinIO Object Storage Client

A small CLI for common MinIO / S3-compatible object storage operations:
upload a directory tree, download an object, list a prefix, or delete
an object. Supports applying a lifecycle expiration rule to uploaded
content so backups self-expire.

Examples:
    python3 minio_client.py --bucket backups --upload ./data/ --expire 30d
    python3 minio_client.py --bucket backups --list --prefix nightly/
    python3 minio_client.py --bucket backups --download nightly/db.sql.gz -o ./db.sql.gz
"""

from __future__ import annotations

import argparse
import os
import re
import sys
from datetime import timedelta
from pathlib import Path
from typing import Iterable

from minio import Minio
from minio.error import S3Error
from minio.lifecycleconfig import Expiration, LifecycleConfig, Rule
from minio.commonconfig import ENABLED

# ---------------------------------------------------------------------------
# Configuration (env vars with sensible defaults)
# ---------------------------------------------------------------------------
DEFAULT_ENDPOINT = os.environ.get("MINIO_ENDPOINT", "localhost:9000")
DEFAULT_ACCESS_KEY = os.environ.get("MINIO_ACCESS_KEY", "minioadmin")
DEFAULT_SECRET_KEY = os.environ.get("MINIO_SECRET_KEY", "minioadmin")
DEFAULT_SECURE = os.environ.get("MINIO_SECURE", "false").lower() in ("1", "true", "yes")
DEFAULT_REGION = os.environ.get("MINIO_REGION", "us-east-1")

# Matches strings like "30d", "12h", "90m", "45s"
DURATION_RE = re.compile(r"^(?P<value>\d+)(?P<unit>[smhd])$")


def parse_duration(text: str) -> timedelta:
    """Convert '30d' / '12h' / '90m' / '45s' into a timedelta."""
    match = DURATION_RE.match(text.strip().lower())
    if not match:
        raise argparse.ArgumentTypeError(
            f"Invalid duration '{text}'. Use forms like 30d, 12h, 90m, 45s."
        )
    value = int(match.group("value"))
    unit = match.group("unit")
    return {
        "s": timedelta(seconds=value),
        "m": timedelta(minutes=value),
        "h": timedelta(hours=value),
        "d": timedelta(days=value),
    }[unit]


def build_client(args: argparse.Namespace) -> Minio:
    """Construct a Minio client from CLI args / environment."""
    return Minio(
        endpoint=args.endpoint,
        access_key=args.access_key,
        secret_key=args.secret_key,
        secure=args.secure,
        region=args.region,
    )


def ensure_bucket(client: Minio, bucket: str, create: bool) -> None:
    """Verify the bucket exists, optionally creating it."""
    if client.bucket_exists(bucket):
        return
    if not create:
        raise RuntimeError(
            f"Bucket '{bucket}' does not exist. Pass --create-bucket to create it."
        )
    client.make_bucket(bucket)
    print(f"[+] Created bucket '{bucket}'", file=sys.stderr)


def iter_upload_files(source: Path) -> Iterable[tuple[Path, str]]:
    """Yield (local_path, relative_key) pairs for every file under source."""
    if source.is_file():
        yield source, source.name
        return
    for root, _dirs, files in os.walk(source):
        for name in files:
            full = Path(root) / name
            rel = full.relative_to(source).as_posix()
            yield full, rel


def apply_expiration(client: Minio, bucket: str, prefix: str, expire: timedelta) -> None:
    """Attach a lifecycle rule that deletes objects under prefix after `expire`."""
    rule_id = f"expire-{prefix.strip('/').replace('/', '-') or 'root'}"
    config = LifecycleConfig(
        [
            Rule(
                ENABLED,
                rule_filter=None,
                rule_id=rule_id,
                expiration=Expiration(days=expire.days or 1),
            )
        ]
    )
    client.set_bucket_lifecycle(bucket, config)
    print(
        f"[+] Lifecycle rule '{rule_id}' set to expire after {expire.days} day(s)",
        file=sys.stderr,
    )


def cmd_upload(client: Minio, args: argparse.Namespace) -> int:
    source = Path(args.upload).expanduser().resolve()
    if not source.exists():
        raise FileNotFoundError(f"Upload source not found: {source}")

    prefix = (args.prefix or "").strip("/")
    count = 0
    for local_path, rel_key in iter_upload_files(source):
        object_name = f"{prefix}/{rel_key}" if prefix else rel_key
        client.fput_object(args.bucket, object_name, str(local_path))
        print(f"[+] Uploaded {local_path} -> s3://{args.bucket}/{object_name}")
        count += 1

    if args.expire:
        apply_expiration(client, args.bucket, prefix, args.expire)

    print(f"[=] Uploaded {count} object(s) to '{args.bucket}'", file=sys.stderr)
    return 0


def cmd_download(client: Minio, args: argparse.Namespace) -> int:
    if not args.object:
        raise ValueError("--download requires an object key (positional or via --prefix)")
    dest = Path(args.output or Path(args.object).name).expanduser()
    dest.parent.mkdir(parents=True, exist_ok=True)
    client.fget_object(args.bucket, args.object, str(dest))
    print(f"[+] Downloaded s3://{args.bucket}/{args.object} -> {dest}")
    return 0


def cmd_list(client: Minio, args: argparse.Namespace) -> int:
    prefix = (args.prefix or "").strip("/")
    total = 0
    for obj in client.list_objects(args.bucket, prefix=prefix, recursive=True):
        size_kb = obj.size / 1024 if obj.size else 0
        print(f"{obj.object_name}\t{size_kb:.1f} KiB\t{obj.last_modified}")
        total += 1
    print(f"[=] {total} object(s) under prefix '{prefix or '/'}'", file=sys.stderr)
    return 0


def cmd_delete(client: Minio, args: argparse.Namespace) -> int:
    if not args.object:
        raise ValueError("--delete requires an object key")
    client.remove_object(args.bucket, args.object)
    print(f"[+] Deleted s3://{args.bucket}/{args.object}")
    return 0


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="MinIO / S3-compatible object storage client",
        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
    )
    parser.add_argument("--endpoint", default=DEFAULT_ENDPOINT, help="host:port of MinIO")
    parser.add_argument("--access-key", default=DEFAULT_ACCESS_KEY)
    parser.add_argument("--secret-key", default=DEFAULT_SECRET_KEY)
    parser.add_argument(
        "--secure",
        action="store_true",
        default=DEFAULT_SECURE,
        help="Use HTTPS",
    )
    parser.add_argument("--region", default=DEFAULT_REGION)
    parser.add_argument("--bucket", required=True, help="Target bucket name")
    parser.add_argument(
        "--create-bucket",
        action="store_true",
        help="Create the bucket if it does not exist",
    )

    group = parser.add_mutually_exclusive_group(required=True)
    group.add_argument("--upload", metavar="PATH", help="Upload a file or directory")
    group.add_argument("--download", metavar="KEY", help="Download an object by key")
    group.add_argument("--list", action="store_true", help="List objects")
    group.add_argument("--delete", metavar="KEY", help="Delete an object by key")

    parser.add_argument("--prefix", default="", help="Object key prefix")
    parser.add_argument("--object", default="", help="Object key (used by download/delete)")
    parser.add_argument("-o", "--output", help="Local output path for --download")
    parser.add_argument(
        "--expire",
        type=parse_duration,
        help="Apply lifecycle expiration (e.g. 30d, 12h) to uploaded prefix",
    )
    return parser


def main(argv: list[str] | None = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)

    # --download and --delete accept the key either positionally via flag
    # or through --object; keep them in sync for convenience.
    if args.download and not args.object:
        args.object = args.download
    if args.delete and not args.object:
        args.object = args.delete

    try:
        client = build_client(args)
        ensure_bucket(client, args.bucket, args.create_bucket)

        if args.upload:
            return cmd_upload(client, args)
        if args.download:
            return cmd_download(client, args)
        if args.list:
            return cmd_list(client, args)
        if args.delete:
            return cmd_delete(client, args)

        parser.error("No operation specified")
        return 2
    except S3Error as exc:
        print(f"[!] S3 error: {exc.code} - {exc.message}", file=sys.stderr)
        return 1
    except (FileNotFoundError, ValueError, RuntimeError) as exc:
        print(f"[!] {exc}", file=sys.stderr)
        return 1
    except KeyboardInterrupt:
        print("[!] Interrupted", file=sys.stderr)
        return 130


if __name__ == "__main__":
    sys.exit(main())

Configuration

All settings can be provided via environment variables (useful for cron / Kubernetes) or overridden with CLI flags.

Usage Examples

1. Upload a directory and set a 30-day expiration on it:

$ python3 minio_client.py --endpoint minio.example.com --bucket backups \
    --upload ./data/ --prefix nightly/ --expire 30d
[+] Uploaded /home/ops/data/db.sql.gz -> s3://backups/nightly/db.sql.gz
[+] Uploaded /home/ops/data/manifest.json -> s3://backups/nightly/manifest.json
[+] Lifecycle rule 'expire-nightly' set to expire after 30 day(s)
[=] Uploaded 2 object(s) to 'backups'

2. List objects under a prefix:

$ python3 minio_client.py --bucket backups --list --prefix nightly/
nightly/db.sql.gz       8421.3 KiB  2024-05-12 03:14:22+00:00
nightly/manifest.json      0.4 KiB   2024-05-12 03:14:22+00:00
[=] 2 object(s) under prefix 'nightly'

3. Download an object to a local file:

$ python3 minio_client.py --bucket backups --download nightly/db.sql.gz -o ./restore/db.sql.gz
[+] Downloaded s3://backups/nightly/db.sql.gz -> restore/db.sql.gz

4. Delete an object:

$ python3 minio_client.py --bucket backups --delete nightly/db.sql.gz
[+] Deleted s3://backups/nightly/db.sql.gz

Error Handling

• S3Error — raised by the SDK for anything the server rejects: bad credentials (AccessDenied), missing bucket (NoSuchBucket), missing object (NoSuchKey), or invalid lifecycle XML. The script prints the S3 error code plus message and exits 1.
• FileNotFoundError — thrown when --upload points at a path that doesn't exist locally. Caught and reported before any network call.
• RuntimeError — raised by ensure_bucket when the bucket is missing and --create-bucket wasn't passed, so you don't accidentally create buckets in production.
• ValueError — raised by the download/delete handlers when no object key is supplied.
• argparse.ArgumentTypeError — parse_duration rejects malformed durations like 30days before any work starts.
• KeyboardInterrupt — mapped to exit code 130, the standard for SIGINT, so Ctrl-C from a cron wrapper is reported correctly.

All errors are written to stderr and the process exits non-zero, which means the script is safe to use as a health-check step in CI or as a Kubernetes CronJob where exit codes trigger alerting.

Extending Ideas

• Server-side encryption: pass sse=SseCustomerKey(...) or sse=SseS3Config() to fput_object/fget_object to encrypt objects at rest without changing the CLI surface.
• Parallel uploads: wrap fput_object calls in a concurrent.futures.ThreadPoolExecutor (Minio's client is thread-safe) to saturate bandwidth on large backup trees.
• Presigned URLs: add a --presign KEY --expires 1h subcommand that calls client.presigned_get_object so you can hand out temporary download links without exposing credentials.
• Structured output: add a --json flag that emits {"uploaded": [...], "errors": [...]} instead of human-readable lines, making the script easier to consume from Ansible, Terraform, or a monitoring pipeline.