MinIO list_objects_v2 性能优化：大规模对象列表的策略与实践

from sqlalchemy import create_engine, Column, Integer, String, BigInteger, DateTime
from sqlalchemy.orm import sessionmaker, declarative_base
from datetime import datetime
import boto3 # 假设使用boto3与MinIO交互

# 1. 定义数据库模型
Base = declarative_base()

class ObjectMetadata(Base):
    __tablename__ = 'object_keys'
    id = Column(Integer, primary_key=True)
    bucket_name = Column(String, nullable=False, index=True)
    object_key = Column(String, nullable=False, unique=True, index=True)
    size = Column(BigInteger)
    last_modified = Column(DateTime)
    # 可以添加更多自定义元数据字段，例如 'content_type', 'user_id' 等

    def __repr__(self):
        return f"<ObjectMetadata(bucket='{self.bucket_name}', key='{self.object_key}')>"

# 2. 数据库初始化
DATABASE_URL = "sqlite:///./minio_metadata.db" # 示例使用SQLite
engine = create_engine(DATABASE_URL)
Base.metadata.create_all(engine) # 创建表
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 3. MinIO客户端初始化
s3_client = boto3.client(
    's3',
    endpoint_url='http://localhost:9000', # 替换为你的MinIO地址
    aws_access_key_id='minioadmin',       # 替换为你的Access Key
    aws_secret_access_key='minioadmin',   # 替换为你的Secret Key
    config=boto3.session.Config(signature_version='s3v4')
)

# 4. 封装对象操作函数
def upload_object_and_update_db(bucket: str, key: str, data: bytes):
    """
    上传对象到MinIO并同步更新数据库。
    """
    db = SessionLocal()
    try:
        # 1. 上传到MinIO
        s3_client.put_object(Bucket=bucket, Key=key, Body=data)

        # 2. 更新数据库
        new_obj = ObjectMetadata(
            bucket_name=bucket,
            object_key=key,
            size=len(data),
            last_modified=datetime.now()
        )
        db.add(new_obj)
        db.commit()
        print(f"Object '{key}' uploaded to MinIO and metadata updated in DB.")
        return True
    except Exception as e:
        db.rollback() # 确保事务回滚
        print(f"Error uploading object or updating DB for '{key}': {e}")
        return False
    finally:
        db.close()

def delete_object_and_update_db(bucket: str, key: str):
    """
    从MinIO删除对象并同步更新数据库。
    """
    db = SessionLocal()
    try:
        # 1. 从MinIO删除
        s3_client.delete_object(Bucket=bucket, Key=key)

        # 2. 从数据库删除
        db.query(ObjectMetadata).filter_by(bucket_name=bucket, object_key=key).delete()
        db.commit()
        print(f"Object '{key}' deleted from MinIO and metadata removed from DB.")
        return True
    except Exception as e:
        db.rollback()
        print(f"Error deleting object or updating DB for '{key}': {e}")
        return False
    finally:
        db.close()

def get_object_keys_from_db(bucket: str, prefix: str = None, limit: int = 100, offset: int = 0):
    """
    从数据库获取对象键列表，支持前缀过滤和分页。
    """
    db = SessionLocal()
    try:
        query = db.query(ObjectMetadata).filter_by(bucket_name=bucket)
        if prefix:
            query = query.filter(ObjectMetadata.object_key.like(f"{prefix}%"))

        results = query.offset(offset).limit(limit).all()
        return [obj.object_key for obj in results]
    except Exception as e:
        print(f"Error querying DB for object keys: {e}")
        return []
    finally:
        db.close()

# 示例使用
if __name__ == "__main__":
    test_bucket = "my-test-bucket"

    # 确保MinIO中存在该桶 (如果不存在，boto3.put_object会自动创建)
    # s3_client.create_bucket(Bucket=test_bucket) 

    # 上传一些对象
    upload_object_and_update_db(test_bucket, "data/file1.txt", b"Hello MinIO")
    upload_object_and_update_db(test_bucket, "data/sub/file2.txt", b"Another file")
    upload_object_and_update_db(test_bucket, "images/pic1.jpg", b"Image data")

    # 从数据库获取对象列表
    print("\nAll object keys from DB:")
    keys = get_object_keys_from_db(test_bucket)
    for key in keys:
        print(key)

    print("\nObject keys with prefix 'data/' from DB:")
    data_keys = get_object_keys_from_db(test_bucket, prefix="data/")
    for key in data_keys:
        print(key)

    # 删除一个对象
    delete_object_and_update_db(test_bucket, "data/file1.txt")

    print("\nObject keys after deletion:")
    keys_after_delete = get_object_keys_from_db(test_bucket)
    for key in keys_after_delete:
        print(key)