chainlit s3 minio 存储集成配置

chainlit s3 默认对于minio 的支持没有明确说明，但是我们可以通过配置解决（环境变量以及~/.aws/config 都可以）

使用

• 代码配置

import chainlit as cl
import chainlit.data as cl_data
from chainlit.data.sql_alchemy import SQLAlchemyDataLayer
from chainlit.types import ThreadDict
import os
 
from chainlit.data.storage_clients import S3StorageClient
 
cl_data._data_layer = SQLAlchemyDataLayer(conninfo=os.environ["PG_CONNECTION_STRING"],storage_provider=S3StorageClient(bucket="chainlit"))
启动
环境变量配置
export AWS_ENDPOINT_URL=http://localhost:9000
export AWS_ACCESS_KEY_ID=minio
export AWS_SECRET_ACCESS_KEY=minio123

import chainlit.data as cl_data

• 效果

说明

因为storage_provider是一个抽象类，实际上我们可以自己实现一个minio的S3StorageClient，目前社区也有实现，只是目前还没merge，参考代码

class MinioStorageClient(BaseStorageClient):
    """
    Class to enable MinIO storage provider
 
    params:
        bucket: Bucket name, should be set with public access
        endpoint_url: MinIO server endpoint, defaults to "http://localhost:9000"
        aws_access_key_id: Default is "minioadmin"
        aws_secret_access_key: Default is "minioadmin"
        verify_ssl: Set to True only if not using HTTP or HTTPS with self-signed SSL certificates
    """
    def __init__(self, bucket: str, endpoint_url: str = 'http://localhost:9000', aws_access_key_id: str = 'minioadmin', aws_secret_access_key: str = 'minioadmin', verify_ssl: bool = False):
        try:
            self.bucket = bucket
            self.endpoint_url = endpoint_url
            self.client = boto3.client("s3", endpoint_url=self.endpoint_url, aws_access_key_id=aws_access_key_id, aws_secret_access_key=aws_secret_access_key, verify=verify_ssl)
            logger.info("MinioStorageClient initialized")
        except Exception as e:
            logger.warn(f"MinioStorageClient initialization error: {e}")
 
    async def upload_file(self, object_key: str, data: Union[bytes, str], mime: str = 'application/octet-stream', overwrite: bool = True) -> Dict[str, Any]:
        try:
            self.client.put_object(Bucket=self.bucket, Key=object_key, Body=data, ContentType=mime)
            url = f"{self.endpoint_url}/{self.bucket}/{object_key}"
            return {"object_key": object_key, "url": url}
        except Exception as e:
            logger.warn(f"MinioStorageClient, upload_file error: {e}")
            return {}

"""

参考资料

https://docs.chainlit.io/data-persistence/custom
https://alexocallaghan.com/configure-boto3-endpoint-url
https://boto3.amazonaws.com/v1/documentation/api/latest/guide/configuration.html#guide-configuration