python3操作对象存储minio

目录[-]

搭建minio服务器

mkdir -p /usr/local/minio/{bin,etc,data}

创建minio用户

groupadd -g 2021 minio
useradd -r -u 2021 -g 2021 -c "Minio User" -s /sbin/nologin minio
# 查看用户
id minio
# uid=2021(minio) gid=2021(minio) 组=2021(minio)

下载minio二进制包

curl -O https://dl.minio.io/server/minio/release/linux-amd64/minio

chmod  750   minio
cp   minio  /usr/local/minio/bin

创建MinIO配置文件

cat >/usr/local/minio/etc/minio.conf<<EOF
MINIO_VOLUMES="/usr/local/minio/data"
MINIO_OPTS="-C /usr/local/minio/etc --address :9000"
MINIO_ACCESS_KEY="minioadmin"
MINIO_SECRET_KEY="minioadmin"
EOF

配置自启动文件

vim /etc/systemd/system/minio.service  
[Unit]
Description=MinIO
Documentation=https://docs.min.io
Wants=network-online.target
After=network-online.target
AssertFileIsExecutable=/usr/local/minio/bin/minio
[Service]
User=minio
Group=minio
EnvironmentFile=/usr/local/minio/etc/minio.conf
ExecStart=/usr/local/minio/bin/minio server $MINIO_OPTS $MINIO_VOLUMES
Restart=always
LimitNOFILE=65536
TimeoutStopSec=infinity
SendSIGKILL=no
[Install]
WantedBy=multi-user.target

更改属组，属主

chown -R minio:minio /usr/local/minio

启动minio,并设置开机自启动

systemctl daemon-reload
systemctl enable minio
systemctl start minio
systemctl status minio

防火墙设置开放9000端口

firewall-cmd --zone=public --add-port=9000/tcp --permanent
firewall-cmd --reload

进入Minio web管理页面 ip地址:9000

输入帐号密码。帐号为设置的MINIO_ACCESS_KEY，密码为设置的MINIO_SECRET_KEY

`python操作minio`

版本要求

python 3.6+

下载

pip install minio

连接

from minio import Minio
client = Minio('0.0.0.0:9000',access_key='xujunkai',secret_key='12345678',secure=False)

查询桶是否存在

from minio import Minio
from minio.error import S3Error
try:
    client = Minio('0.0.0.0:9000',access_key='xujunkai',secret_key='12345678',secure=False)
    found = client.bucket_exists("buk2")
except S3Error as e:
    print("error:", e)
print(found)# 返回布尔值 True or False

上传文件,若文件已存在，会直接覆盖

client.fput_object("buk2", "1.jpg", "/opt/img/1.jpg")
# 文件不存在会报错FileNotFoundError

获取文件数据写入指定文件

data = client.get_object("buk2", "1.jpg")
with open("/opt/1demo.jpg","wb") as fp:
    for d in data.stream(1024):
        fp.write(d)

直接将文件下载到本地

client.fget_object("buk2", "1.jpg", "/opt/static/new.jpg")

删除文件

client.remove_object("buk2", file_name)

删除多个文件

client.remove_objects("buk2", ["baidu.html", "taobao.html"])

`封装python api`

import os
from minio import Minio
from minio.error import S3Error
from datetime import timedelta
from minio.deleteobjects import DeleteObject
class Bucket(object):
    client = None
    policy = '{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":{"AWS":["*"]},"Action":["s3:GetBucketLocation","s3:ListBucket"],"Resource":["arn:aws:s3:::%s"]},{"Effect":"Allow","Principal":{"AWS":["*"]},"Action":["s3:GetObject"],"Resource":["arn:aws:s3:::%s/*"]}]}'
    def __new__(cls, *args, **kwargs):
        if not cls.client:
            cls.client = object.__new__(cls)
        return cls.client
    def __init__(self, service, access_key, secret_key, secure=False):
        self.service = service
        self.client = Minio(service, access_key=access_key, secret_key=secret_key, secure=secure)
    def exists_bucket(self, bucket_name):
        """
        判断桶是否存在
        :param bucket_name: 桶名称
        :return:
        """
        return self.client.bucket_exists(bucket_name=bucket_name)
    def create_bucket(self, bucket_name:str, is_policy:bool=True):
        """
        创建桶 + 赋予策略
        :param bucket_name: 桶名
        :param is_policy: 策略
        :return:
        """
        if self.exists_bucket(bucket_name=bucket_name):
            return False
        else:
            self.client.make_bucket(bucket_name = bucket_name)
        if is_policy:
            policy = self.policy % (bucket_name, bucket_name)
            self.client.set_bucket_policy(bucket_name=bucket_name, policy=policy)
        return True

    def get_bucket_list(self):
        """
        列出存储桶
        :return:
        """
        buckets = self.client.list_buckets()
        bucket_list = []
        for bucket in buckets:
            bucket_list.append(
                {"bucket_name": bucket.name, "create_time": bucket.creation_date}
            )
        return bucket_list

    def remove_bucket(self, bucket_name):
        """
        删除桶
        :param bucket_name:
        :return:
        """
        try:
            self.client.remove_bucket(bucket_name=bucket_name)
        except S3Error as e:
            print("[error]:", e)
            return False
        return True
    def bucket_list_files(self, bucket_name, prefix):
        """
        列出存储桶中所有对象
        :param bucket_name: 同名
        :param prefix: 前缀
        :return:
        """
        try:
            files_list = self.client.list_objects(bucket_name=bucket_name, prefix=prefix, recursive=True)
            for obj in files_list:
                print(obj.bucket_name, obj.object_name.encode('utf-8'), obj.last_modified,
                      obj.etag, obj.size, obj.content_type)
        except S3Error as e:
            print("[error]:", e)
    def bucket_policy(self, bucket_name):
        """
        列出桶存储策略
        :param bucket_name:
        :return:
        """
        try:
            policy = self.client.get_bucket_policy(bucket_name)
        except S3Error as e:
            print("[error]:", e)
            return None
        return policy

    def download_file(self, bucket_name, file, file_path, stream=1024*32):
        """
        从bucket 下载文件 + 写入指定文件
        :return:
        """
        try:
            data = self.client.get_object(bucket_name, file)
            with open(file_path, "wb") as fp:
                for d in data.stream(stream):
                    fp.write(d)
        except S3Error as e:
            print("[error]:", e)
    def fget_file(self, bucket_name, file, file_path):
        """
        下载保存文件保存本地
        :param bucket_name:
        :param file:
        :param file_path:
        :return:
        """
        self.client.fget_object(bucket_name, file, file_path)
    def copy_file(self, bucket_name, file, file_path):
        """
        拷贝文件（最大支持5GB）
        :param bucket_name:
        :param file:
        :param file_path:
        :return:
        """
        self.client.copy_object(bucket_name, file, file_path)
    def upload_file(self,bucket_name, file, file_path, content_type):
        """
        上传文件 + 写入
        :param bucket_name: 桶名
        :param file: 文件名
        :param file_path: 本地文件路径
        :param content_type: 文件类型
        :return:
        """
        try:
            with open(file_path, "rb") as file_data:
                file_stat = os.stat(file_path)
                self.client.put_object(bucket_name, file, file_data, file_stat.st_size, content_type=content_type)
        except S3Error as e:
            print("[error]:", e)
    def fput_file(self, bucket_name, file, file_path):
        """
        上传文件
        :param bucket_name: 桶名
        :param file: 文件名
        :param file_path: 本地文件路径
        :return:
        """
        try:
            self.client.fput_object(bucket_name, file, file_path)
        except S3Error as e:
            print("[error]:", e)

    def stat_object(self, bucket_name, file):
        """
        获取文件元数据
        :param bucket_name:
        :param file:
        :return:
        """
        try:
            data = self.client.stat_object(bucket_name, file)
            print(data.bucket_name)
            print(data.object_name)
            print(data.last_modified)
            print(data.etag)
            print(data.size)
            print(data.metadata)
            print(data.content_type)
        except S3Error as e:
            print("[error]:", e)
    def remove_file(self, bucket_name, file):
        """
        移除单个文件
        :return:
        """
        self.client.remove_object(bucket_name, file)
    def remove_files(self, bucket_name, file_list):
        """
        删除多个文件
        :return:
        """
        delete_object_list = [DeleteObject(file) for file in file_list]
        for del_err in self.client.remove_objects(bucket_name, delete_object_list):
            print("del_err", del_err)
    def presigned_get_file(self, bucket_name, file, days=7):
        """
        生成一个http GET操作 签证URL
        :return:
        """
        return self.client.presigned_get_object(bucket_name, file, expires=timedelta(days=days))


if __name__ == '__main__':
    minio_obj = Bucket(service="10.0.0.70:9000", access_key="xujunkai", secret_key="12345678")

docker方式快速构建minio

docker run -p 9000:9000 --name minio \
    -d --restart=always \
    -e "MINIO_ACCESS_KEY=admin" \
    -e "MINIO_SECRET_KEY=xjk214365" \
    -v /home/data:/data \
    -v /home/config:/root/.minio \
    minio/minio server /data

参照：

https://docs.min.io/docs/python-client-quickstart-guide

https://blog.csdn.net/weixin_40547993/article/details/110682587?utm_medium=distribute.pc_relevant.none-task-blog-2~default~BlogCommendFromMachineLearnPai2~default-1.vipsorttest&depth_1-utm_source=distribute.pc_relevant.none-task-blog-2~default~BlogCommendFromMachineLearnPai2~default-1.vipsorttest

选择风格关闭

python3操作对象存储minio

搭建minio服务器

`python操作minio`

`封装python api`

我要留言 / 展开表单