Phad RustFS SDK

Phad RustFS SDK 是一个用于访问 RustFS 文件服务的 Python SDK，通过 Nacos 服务发现机制获取服务地址，提供文件上传、文件列表查询、文件信息获取等功能。

功能特性

• 基于 Nacos 的服务发现
• 支持文件上传
• 支持文件列表查询
• 支持单个文件信息获取
• 支持文件 URL 获取
• 支持文件删除
• 完善的异常处理
• 简单易用的 API 设计

安装

pip install phad-rustfs-sdk

快速开始

基础配置

from phad_rustfs_sdk import PhadRustFSClient, PhadRustFSConfig

# 创建配置
config = PhadRustFSConfig(
    nacos_server="10.84.4.141:8848",
    service_name="llm-file",
    namespace="llm_dev",
    username="phadagent",
    password="bB0!bO7@aA2)gC2?fH"
)

# 创建客户端
client = PhadRustFSClient(config)

获取文件列表

# 测试基本功能
try:
    result = client.get_file_list([1770261007132])
    print("测试成功！")
    print(f"结果: {result}")
except Exception as e:
    print(f"测试失败: {e}")

上传文件

# 上传文件
result = client.upload_file("test.pdf", 1, "123", "true", "my-bucket", "linghu", 1)
print(result)
if result["success"]:
    print(f"文件ID: {result['data'].get('fileId')}")
    print(f"文件URL: {result['data'].get('url')}")

API 文档

PhadRustFSConfig

配置类，用于配置 SDK 的连接参数。

参数:

• nacos_server (str): Nacos 服务器地址，格式为 "host:port"
• service_name (str): 服务名称，默认为 "llm-file"
• namespace (str): Nacos 命名空间，默认为 "public"
• username (str, optional): Nacos 用户名
• password (str, optional): Nacos 密码
• default_service_url (str, optional): 默认服务地址
• timeout (int): 请求超时时间（秒），默认为 30

PhadRustFSClient

客户端类，提供文件服务相关的操作方法。

方法:

get_file_list(file_ids: List[int]) -> Dict[str, Any]

获取文件列表。

参数:

• file_ids: 文件 ID 列表

返回:

• 包含文件列表的字典，格式为:

  {
      "success": bool,
      "data": {
          "code": int,
          "msg": str,
          "data": [
              {
                  "fileId": int,
                  "name": str,
                  "url": str,
                  "fileSize": int,
                  ...
              }
          ]
      }
  }
  

get_file_info(file_id: int) -> Dict[str, Any]

获取单个文件信息。

参数:

• file_id: 文件 ID

返回:

• 包含文件信息的字典

get_file_url(file_id: int) -> str

获取文件 URL。

参数:

• file_id: 文件 ID

返回:

• 文件 URL

refresh_service_address() -> str

刷新服务地址（重新从 Nacos 获取）。

返回:

• 新的服务地址

upload_file(file_path: str, source_type: int = 1, from_data_id: Optional[str] = None, rename: bool = True, bucket_name: Optional[str] = None, folder_name: Optional[str] = None, storage_type: int = 1) -> Dict[str, Any]

上传文件到远程服务。

参数:

• file_path: 文件路径
• source_type: 文件来源：1-系统, 2-web (默认: 1)
• from_data_id: 数据来源记录ID (可选)
• rename: 重命名标记：true-重命名, false-不重命名 (默认: true)
• bucket_name: 自定义文件桶名称 (可选)
• folder_name: 自定义文件夹名称 (可选)
• storage_type: 存储类型：1-内网, 2-外网 (默认: 1)

返回:

• 包含上传结果的字典，格式为:

  {
      "success": bool,
      "data": {
          "fileId": int,
          "url": str,
          ...
      }
  }
  

delete_file(file_id: int) -> Dict[str, Any]

删除文件。

参数:

• file_id: 文件 ID

返回:

• 包含删除结果的字典，格式为:

  {
      "success": bool,
      "data": {
          "code": int,
          "msg": str,
          ...
      }
  }
  

详细使用示例

示例1: 获取文件列表

from phad_rustfs_sdk import PhadRustFSClient, PhadRustFSConfig

# 创建配置
config = PhadRustFSConfig(
    nacos_server="10.84.4.141:8848",
    service_name="llm-file",
    namespace="llm_dev",
    username="phadagent",
    password="bB0!bO7@aA2)gC2?fH"
)

# 创建客户端
client = PhadRustFSClient(config)

# 测试基本功能
try:
    result = client.get_file_list([1770261007132])
    print("测试成功！")
    print(f"结果: {result}")
except Exception as e:
    print(f"测试失败: {e}")

示例2: 上传文件

from phad_rustfs_sdk import PhadRustFSClient, PhadRustFSConfig

config = PhadRustFSConfig(
    nacos_server="10.84.4.141:8848",
    service_name="llm-file",
    namespace="llm_dev",
    username="phadagent",
    password="bB0!bO7@aA2)gC2?fH"
)

client = PhadRustFSClient(config)

# 上传文件
result = client.upload_file("test.pdf", 1, "123", "true", "my-bucket", "linghu", 1)
print(result)
if result["success"]:
    print(f"文件ID: {result['data'].get('fileId')}")
    print(f"文件URL: {result['data'].get('url')}")

示例3: 删除文件

from phad_rustfs_sdk import PhadRustFSClient, PhadRustFSConfig

config = PhadRustFSConfig(
    nacos_server="10.84.4.141:8848",
    service_name="llm-file",
    namespace="llm_dev",
    username="phadagent",
    password="bB0!bO7@aA2)gC2?fH"
)

client = PhadRustFSClient(config)

# 删除文件
file_id = 1770261007132  # 替换为实际的文件ID
result = client.delete_file(file_id)
print(result)
if result["success"]:
    print("文件删除成功！")
else:
    print("文件删除失败！")

异常处理

SDK 提供了完善的异常处理机制：

from phad_rustfs_sdk import (
    PhadRustFSClient, 
    PhadRustFSConfig,
    APIError,
    AuthenticationError,
    ServiceNotFoundError,
    FileNotFoundError
)

config = PhadRustFSConfig(...)
client = PhadRustFSClient(config)

try:
    result = client.get_file_list([1770261007132])
except AuthenticationError as e:
    print(f"认证失败: {e}")
except ServiceNotFoundError as e:
    print(f"服务未找到: {e}")
except FileNotFoundError as e:
    print(f"文件未找到: {e}")
except APIError as e:
    print(f"API错误: {e}")
    if e.status_code:
        print(f"状态码: {e.status_code}")
    if e.response_data:
        print(f"响应数据: {e.response_data}")

异常类型

• PhadRustFSError: 基础异常类
• AuthenticationError: Nacos 认证失败异常
• ServiceNotFoundError: 服务未找到异常
• FileNotFoundError: 文件未找到异常
• APIError: API 调用失败异常

开发

安装开发依赖

pip install -e ".[dev]"

运行测试

pytest

代码格式化

black phad_rustfs_sdk/

类型检查

mypy phad_rustfs_sdk/

许可证

MIT License

贡献

欢迎提交 Issue 和 Pull Request！