pip install minio

from minio import Minio

# 创建具有匿名访问权限的客户端。
client = Minio("play.min.io")

# 创建具有访问权限和密钥的客户端。
client = Minio("s3.amazonaws.com", "ACCESS-KEY", "SECRET-KEY")

# 创建具有特定区域的访问密钥和密钥的客户端。
client = Minio(
    "play.minio.io:9000",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    region="my-region",
)

# 使用代理服务器创建带有自定义HTTP客户端的客户端。
import urllib3
client = Minio(
    "SERVER:PORT",
    access_key="ACCESS_KEY",
    secret_key="SECRET_KEY",
    secure=True,
    http_client=urllib3.ProxyManager(
        "https://PROXYSERVER:PROXYPORT/",
        timeout=urllib3.Timeout.DEFAULT_TIMEOUT,
        cert_reqs="CERT_REQUIRED",
        retries=urllib3.Retry(
            total=5,
            backoff_factor=0.2,
            status_forcelist=[500, 502, 503, 504],
        ),
    ),
)

# 创建bucket。
client.make_bucket("my-bucket")

# 在特定区域创建bucket。
client.make_bucket("my-bucket", "us-west-1")

# 在特定区域创建具有对象锁定功能的桶。
client.make_bucket("my-bucket", "eu-west-2", object_lock=True)

# 查询所有桶
buckets = client.list_buckets()
for bucket in buckets:
    print(bucket.name, bucket.creation_date)
# 判断桶是否存在
if client.bucket_exists("my-bucket"):
    print("my-bucket exists")
else:
    print("my-bucket does not exist")

# 列出对象信息。
objects = client.list_objects("my-bucket")
for obj in objects:
    print(obj)

# 列出名称以“my/prefix/”开头的对象信息。
objects = client.list_objects("my-bucket", prefix="my/prefix/")
for obj in objects:
    print(obj)

# 递归地列出对象信息。
objects = client.list_objects("my-bucket", recursive=True)
for obj in objects:
    print(obj)

# 递归列出名称以开头的对象信息
# "my/prefix/".
objects = client.list_objects(
    "my-bucket", prefix="my/prefix/", recursive=True,
)
for obj in objects:
    print(obj)

# 在对象名称后递归列出对象信息
objects = client.list_objects(
    "my-bucket", recursive=True, start_after="my/prefix/world/1",
)
for obj in objects:
    print(obj)

client.remove_bucket("my-bucket")

policy = client.get_bucket_policy("my-bucket")

# 匿名只读存储桶策略。
policy = {
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Principal": {"AWS": "*"},
            "Action": ["s3:GetBucketLocation", "s3:ListBucket"],
            "Resource": "arn:aws:s3:::my-bucket",
        },
        {
            "Effect": "Allow",
            "Principal": {"AWS": "*"},
            "Action": "s3:GetObject",
            "Resource": "arn:aws:s3:::my-bucket/*",
        },
    ],
}
client.set_bucket_policy("my-bucket", json.dumps(policy))

# 匿名读写存储桶策略。
policy = {
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Principal": {"AWS": "*"},
            "Action": [
                "s3:GetBucketLocation",
                "s3:ListBucket",
                "s3:ListBucketMultipartUploads",
            ],
            "Resource": "arn:aws:s3:::my-bucket",
        },
        {
            "Effect": "Allow",
            "Principal": {"AWS": "*"},
            "Action": [
                "s3:GetObject",
                "s3:PutObject",
                "s3:DeleteObject",
                "s3:ListMultipartUploadParts",
                "s3:AbortMultipartUpload",
            ],
            "Resource": "arn:aws:s3:::my-bucket/images/*",
        },
    ],
}
client.set_bucket_policy("my-bucket", json.dumps(policy))

client.delete_bucket_policy("my-bucket")

config = client.get_bucket_notification("my-bucket")

config = NotificationConfig(
    queue_config_list=[
        QueueConfig(
            "QUEUE-ARN-OF-THIS-BUCKET",
            ["s3:ObjectCreated:*"],
            config_id="1",
            prefix_filter_rule=PrefixFilterRule("abc"),
        ),
    ],
)
client.set_bucket_notification("my-bucket", config)

client.delete_bucket_notification("my-bucket")

with client.listen_bucket_notification(
    "my-bucket",
    prefix="my-prefix/",
    events=["s3:ObjectCreated:*", "s3:ObjectRemoved:*"],
) as events:
    for event in events:
        print(event)

config = client.get_bucket_encryption("my-bucket")

client.set_bucket_encryption(
    "my-bucket", SSEConfig(Rule.new_sse_s3_rule()),
)

client.delete_bucket_encryption("my-bucket")

config = client.get_bucket_versioning("my-bucket")
print(config.status)

client.set_bucket_versioning("my-bucket", VersioningConfig(ENABLED))

config = client.get_bucket_replication("my-bucket")

config = ReplicationConfig(
    "REPLACE-WITH-ACTUAL-ROLE",
    [
        Rule(
            Destination(
                "REPLACE-WITH-ACTUAL-DESTINATION-BUCKET-ARN",
            ),
            ENABLED,
            delete_marker_replication=DeleteMarkerReplication(
                DISABLED,
            ),
            rule_filter=Filter(
                AndOperator(
                    "TaxDocs",
                    {"key1": "value1", "key2": "value2"},
                ),
            ),
            rule_id="rule1",
            priority=1,
        ),
    ],
)
client.set_bucket_replication("my-bucket", config)

client.delete_bucket_replication("my-bucket")

config = client.get_bucket_lifecycle("my-bucket")

config = LifecycleConfig(
    [
        Rule(
            ENABLED,
            rule_filter=Filter(prefix="documents/"),
            rule_id="rule1",
            transition=Transition(days=30, storage_class="GLACIER"),
        ),
        Rule(
            ENABLED,
            rule_filter=Filter(prefix="logs/"),
            rule_id="rule2",
            expiration=Expiration(days=365),
        ),
    ],
)
client.set_bucket_lifecycle("my-bucket", config)

client.delete_bucket_lifecycle("my-bucket")

tags = client.get_bucket_tags("my-bucket")

tags = Tags.new_bucket_tags()
tags["Project"] = "Project One"
tags["User"] = "jsmith"
client.set_bucket_tags("my-bucket", tags)

client.delete_bucket_tags("my-bucket")

config = client.get_object_lock_config("my-bucket")

config = ObjectLockConfig(GOVERNANCE, 15, DAYS)
client.set_object_lock_condig("my-bucket", config)

client.delete_object_lock_config("my-bucket")