abdds 0.2.0

Async Baidu Disk SDK - 百度网盘异步 Python SDK

abdds

Async Baidu Disk SDK - 简单、有效、报错清晰的百度网盘异步 Python SDK。

基于 百度网盘开放平台 API 封装，核心依赖 httpx + aiofiles，全面异步化。

安装

pip install abdds

要求 Python 3.9+。

快速开始

1. 获取应用凭证

在 百度网盘开放平台 创建应用，获取 client_id、client_secret 和 app_name。

2. 初始化客户端

from pathlib import Path
from abdds import AsyncBaiduPanClient

async with AsyncBaiduPanClient(
    client_id="your_client_id",
    client_secret="your_client_secret",
    app_name="your_app_name",
) as client:
    if not client.is_authenticated:
        print(f"请在浏览器打开: {client.auth_url}")
        code = input("请输入授权码: ")
        await client.fetch_token(code)

Token 会自动保存到 ~/.baidupan/ 目录，下次启动无需重新授权。

3. 使用 API

# 查询空间配额
q = await client.quota()
print(f"已用: {q.used / 1024**3:.1f} GB / 总计: {q.total / 1024**3:.1f} GB")

# 上传文件 - 支持 Path / bytes / Generator / AsyncGenerator
result = await client.upload(Path("local_file.txt"))
result = await client.upload(Path("data.bin"), remote_path="/backup/data.bin")
result = await client.upload(b"hello world", remote_path="/hello.txt")

# 获取文件元信息
metas = await client.file_metas([result.fs_id])
dlink = metas[0].dlink

# 下载文件到本地路径
await client.download_to_file(dlink, Path("downloaded.txt"))

# 流式下载（适合大文件或自定义处理）
async for chunk in client.download_stream(dlink):
    process(chunk)

# 带断点续传的流式下载
async for chunk in client.download_stream_with_range(dlink, range=(1024, None)):
    process(chunk)

# 列出文件
items = await client.list_files("/apps/your_app_name/")
for item in items:
    print(f"{'[DIR]' if item.is_dir else '     '} {item.filename} ({item.size} bytes)")

# 删除文件
await client.delete(["/apps/your_app_name/old_file.txt"])

或使用异步工厂方法：

client = await AsyncBaiduPanClient.create(client_id, client_secret, app_name)
q = await client.quota()
await client.close()

API 参考

AsyncBaiduPanClient

方法	说明
await upload(source, remote_path="")	异步上传数据。source 支持 Path / bytes / Generator[bytes] / AsyncGenerator[bytes]；remote_path 为远程路径字符串
await download_to_file(dlink, local_path, *, chunk_size=1048576)	下载文件到本地路径，返回 Path
download_stream(dlink, *, chunk_size=1048576)	异步流式下载，返回 AsyncGenerator[bytes]
download_stream_with_range(dlink, range, *, chunk_size=1048576, max_retries=5)	带断点续传的异步流式下载，range 为 (start, end) 元组
await quota()	获取空间配额，返回 ApiQuotaInfo
await file_metas(fsids)	获取文件元信息（含 dlink），返回 list[ApiFileMeta]
await list_files(path="/")	列出目录文件，返回 list[ApiFileListItem]
await delete(paths)	批量删除文件
await fetch_token(code)	通过授权码获取 Token
await refresh_token()	刷新 Token
is_authenticated	是否已认证（属性）
await close()	关闭客户端

异常体系

BaiduPanError
├── BaiduPanNetworkError    # 网络通信异常
├── BaiduPanAPIError        # API 业务异常 (errno, errmsg, request_id)
└── TokenExpiredError       # Token 过期或无效

所有异常均继承自 BaiduPanError，方便统一捕获。

特性

• 全面异步 — 基于 httpx + aiofiles，原生 async/await
• 自动重试 — 网络请求自动重试（可配置次数）
• Token 管理 — 自动持久化、自动刷新、文件权限限制
• 分片上传 — 大文件自动分片上传（4MB 分片）
• 流式下载 — 支持断点续传、Range 分片、指数重试
• 类型安全 — 完整类型标注，支持 PEP 561 (py.typed)
• 报错清晰 — 异常包含 errno、errmsg、request_id、url 等上下文

与同步版本 bdds 的差异

特性	bdds (同步)	abdds (异步)
HTTP 客户端	requests	httpx
文件 I/O	内置 open	aiofiles
上下文管理器	with / __enter__	async with / __aenter__
上传数据源	Path / bytes / Generator	Path / bytes / Generator / AsyncGenerator
下载返回	Generator[bytes]	AsyncGenerator[bytes]
Token 回调	同步函数	异步函数
测试 mock	responses	respx

开发

# 安装开发依赖
pip install -e ".[dev]"

# 运行测试
python -m pytest -v

# 构建
python -m build

License

MIT