tigris-python-sdk

Tigris Python SDK

通过 boto3 和 tigris-boto3-ext 扩展包在 Python 中使用 Tigris。标准 S3 调用保持不变；该扩展通过在已激活的 boto3 事件上注入请求头，添加 Tigris 专属功能——快照、存储桶分叉、原位重命名对象，以及用于批量获取的 Bundle API。

如需了解通过 django-storages 上传到 Django 的方式，请参阅 ./resources/django.md。

快速入门

# 1. Install CLI & authenticate
npm install -g @tigrisdata/cli
tigris login

# 2. Create bucket and access key
tigris buckets create my-bucket
tigris access-keys create "my-bucket-key"
# ⚠ Save the Secret Access Key — shown only once
tigris access-keys assign tid_xxx --bucket my-bucket --role Editor

# 3. Install Python deps
pip install boto3 tigris-boto3-ext

# .env
AWS_ENDPOINT_URL_S3=https://t3.storage.dev
AWS_ACCESS_KEY_ID=tid_xxx
AWS_SECRET_ACCESS_KEY=tsec_yyy
AWS_REGION=auto

import boto3
from tigris_boto3_ext import (
    TigrisSnapshotEnabled, create_snapshot, get_snapshot_version,
)

s3 = boto3.client("s3")  # picks up env vars

# Create a snapshot-enabled bucket (header injection scoped to the block)
with TigrisSnapshotEnabled(s3):
    s3.create_bucket(Bucket="my-bucket")

# Take a snapshot
result = create_snapshot(s3, "my-bucket", snapshot_name="daily-backup")
version = get_snapshot_version(result)

详细设置步骤请参阅 CLI 入门。

---

CLI 入门

第 1 步：安装 CLI

npm install -g @tigrisdata/cli
tigris --version

t3 是 tigris 的别名——所有命令都可以使用任一名称。

第 2 步：进行身份验证

tigris login    # opens browser for OAuth
tigris whoami   # verify

用于 CI/CD：

tigris configure --access-key <key> --access-secret <secret>

第 3 步：创建存储桶

tigris buckets create my-bucket

存储桶默认设为私有并在全球范围内可用。使用 --public 允许匿名读取，使用 --locations 固定区域。

第 4 步：创建访问密钥

tigris access-keys create "my-bucket-key"

输出访问密钥 ID（tid_xxx）和秘密访问密钥（tsec_yyy）。秘密访问密钥仅显示一次。

第 5 步：为存储桶分配密钥

tigris access-keys assign tid_xxx --bucket my-bucket --role Editor

角色	权限	适用场景
Editor	读取 + 写入 + 删除	上传或删除对象的应用代码
ReadOnly	只读	只下载或列出对象的应用

第 6 步：配置环境

# .env (add to .gitignore)
AWS_ENDPOINT_URL_S3=https://t3.storage.dev
AWS_ACCESS_KEY_ID=tid_xxx
AWS_SECRET_ACCESS_KEY=tsec_yyy
AWS_REGION=auto

boto3 原生读取 AWS_ENDPOINT_URL_S3（boto3 ≥ 1.34）——设置该变量后，无需客户端覆盖设置。

第 7 步：安装 Python 依赖项

pip install boto3 tigris-boto3-ext

需要 Python 3.9+ 和 boto3 ≥ 1.26.0。

---

boto3 客户端设置

import boto3

# Reads AWS_ENDPOINT_URL_S3, AWS_ACCESS_KEY_ID, etc. from env
s3 = boto3.client("s3")

# Or configure explicitly
s3 = boto3.client(
    "s3",
    endpoint_url="https://t3.storage.dev",
    aws_access_key_id="tid_xxx",
    aws_secret_access_key="tsec_yyy",
    region_name="auto",
)

tigris-boto3-ext 适用于任何 boto3 S3 客户端——它会在 before-sign.s3.* 上注册处理程序，在激活时注入 Tigris 请求头。未处于已激活作用域内的调用会原样通过。

---

使用模式

tigris-boto3-ext 提供了三种限定 Tigris 专属行为作用域的方式。选择最适合调用位置的一种：

模式	最适用场景
上下文管理器	作用域明确的一次性代码块
装饰器	始终需要 Tigris 功能的函数
辅助函数	无需作用域设置的直接调用

上下文管理器

from tigris_boto3_ext import (
    TigrisSnapshotEnabled, TigrisSnapshot, TigrisFork, TigrisRename,
)

# Enable snapshots on bucket creation
with TigrisSnapshotEnabled(s3):
    s3.create_bucket(Bucket="my-bucket")

# List snapshots for a bucket
with TigrisSnapshot(s3, "my-bucket"):
    snapshots = s3.list_buckets()

# Read from a specific snapshot version
with TigrisSnapshot(s3, "my-bucket", snapshot_version="12345"):
    obj = s3.get_object(Bucket="my-bucket", Key="file.txt")

# Fork a bucket (current state or from a snapshot)
with TigrisFork(s3, "source-bucket", snapshot_version="12345"):
    s3.create_bucket(Bucket="forked-bucket")

# Rename in place — keep scope tight
with TigrisRename(s3):
    s3.copy_object(
        Bucket="my-bucket",
        CopySource="my-bucket/old.txt",
        Key="new.txt",
    )

装饰器

from tigris_boto3_ext import snapshot_enabled, with_snapshot, forked_from, with_rename

@snapshot_enabled
def make_snapshot_bucket(s3, name):
    return s3.create_bucket(Bucket=name)

@with_snapshot("my-bucket", snapshot_version="12345")
def read_old_config(s3):
    return s3.get_object(Bucket="my-bucket", Key="config.json")

@forked_from("source-bucket", snapshot_version="12345")
def make_fork(s3, name):
    return s3.create_bucket(Bucket=name)

@with_rename
def rename(s3, bucket, old, new):
    return s3.copy_object(Bucket=bucket, CopySource=f"{bucket}/{old}", Key=new)

辅助函数

from tigris_boto3_ext import (
    create_snapshot_bucket,
    create_snapshot,
    list_snapshots,
    create_fork,
    get_object_from_snapshot,
    get_snapshot_version,
    list_objects_from_snapshot,
    head_object_from_snapshot,
    has_snapshot_enabled,
    get_bucket_info,
    rename_object,
)

create_snapshot_bucket(s3, "my-bucket")

result = create_snapshot(s3, "my-bucket", snapshot_name="backup-1")
version = get_snapshot_version(result)

snapshots = list_snapshots(s3, "my-bucket")
create_fork(s3, "fork-bucket", "my-bucket", snapshot_version=version)

obj = get_object_from_snapshot(s3, "my-bucket", "file.txt", version)
objects = list_objects_from_snapshot(s3, "my-bucket", version, Prefix="data/")
metadata = head_object_from_snapshot(s3, "my-bucket", "file.txt", version)

rename_object(s3, "my-bucket", "old.txt", "new.txt")

# Inspect bucket
if has_snapshot_enabled(s3, "my-bucket"):
    info = get_bucket_info(s3, "my-bucket")
    # info: { snapshot_enabled, fork_source_bucket, fork_source_snapshot }

---

快照

存储桶在某个时间点的副本。Tigris 只存储差异，因此快照成本很低。

# 1. Enable snapshots when creating the bucket
create_snapshot_bucket(s3, "production-data")

# 2. Take a named snapshot
result = create_snapshot(s3, "production-data", snapshot_name="daily-2026-05-07")
version = get_snapshot_version(result)

# 3. Read from the snapshot — historical view
with TigrisSnapshot(s3, "production-data", snapshot_version=version):
    obj = s3.get_object(Bucket="production-data", Key="config.json")

# 4. List all snapshots
snaps = list_snapshots(s3, "production-data")
for b in snaps.get("Buckets", []):
    print(b["Name"])

必须在创建存储桶时启用快照；无法事后启用。

存储桶分叉

分叉是根据源存储桶的实时状态或快照创建的独立存储桶。分叉发生差异前会共享存储，因此几乎不产生额外成本。

# Fork from current state
with TigrisFork(s3, "source-bucket"):
    s3.create_bucket(Bucket="forked-bucket")

# Fork from a snapshot — common for testing against prod data
create_fork(s3, "test-bucket", "production-data", snapshot_version=version)

# Inspect lineage
info = get_bucket_info(s3, "test-bucket")
print(info["fork_source_bucket"], info["fork_source_snapshot"])

常见用途：基于生产数据创建测试/预发布环境、从事故前的快照进行时光回溯式调试，以及在机器学习工作流中为每项实验创建独立工作区。

原位重命名对象

Tigris 通过 copy_object 和 X-Tigris-Rename: true 请求头实现重命名——只更改键，不会重写数据。

from tigris_boto3_ext import TigrisRename, rename_object

# Helper form (recommended)
rename_object(s3, "my-bucket", "old-name.txt", "new-name.txt")

# Context manager — every copy_object inside the block becomes a rename
with TigrisRename(s3):
    s3.copy_object(
        Bucket="my-bucket",
        CopySource="my-bucket/old.txt",
        Key="new.txt",
    )

请让 TigrisRename / with_rename 作用域保持精简。在代码块中执行的任何 copy_object 都会变成重命名——其他无关复制操作将无法按预期工作。

条件写入

Tigris 支持标准的 S3 前置条件标头——If-Match、If-None-Match、If-Modified-Since、If-Unmodified-Since——用于乐观并发控制和仅在对象不存在时创建的语义。这些都是纯 boto3 参数；tigris-boto3-ext 不会在此基础上添加任何内容，因为 boto3 已经会传递这些参数。

from botocore.exceptions import ClientError

# Create-only (fail if the key already exists)
try:
    s3.put_object(
        Bucket="my-bucket",
        Key="config.json",
        Body=b'{"version": 1}',
        IfNoneMatch="*",
    )
except ClientError as e:
    if e.response["Error"]["Code"] == "PreconditionFailed":
        print("Object already exists")
    else:
        raise

# Optimistic concurrency — read, modify, write only if unchanged
head = s3.head_object(Bucket="my-bucket", Key="config.json")
etag = head["ETag"]

new_body = b'{"version": 2}'
try:
    s3.put_object(
        Bucket="my-bucket",
        Key="config.json",
        Body=new_body,
        IfMatch=etag,
    )
except ClientError as e:
    if e.response["Error"]["Code"] == "PreconditionFailed":
        # Someone else wrote the object since we read it — retry the read-modify-write
        ...
    else:
        raise

boto3 参数	行为	常见用途
IfMatch=<etag>	仅当当前 ETag 匹配时才写入	乐观并发控制 / 避免更新丢失
IfNoneMatch="*"	仅当对象不存在时才写入	仅创建一次 / 幂等写入
IfNoneMatch=<etag>	仅当当前 ETag 不匹配时才写入	跳过相同内容的重复上传
IfModifiedSince	仅当对象在给定的 RFC 1123 时间戳之后修改时才继续	条件读取 / 缓存验证
IfUnmodifiedSince	仅当对象在给定的 RFC 1123 时间戳之后未修改时才继续	有时间限制的乐观并发控制

前置条件失败时会返回 ClientError，其 Error.Code == "PreconditionFailed"（HTTP 412）。请捕获这个特定代码——不要吞掉所有 ClientError，因为相同的异常类型还涵盖 NoSuchKey、限流等情况。

IfMatch / IfNoneMatch 可用于 put_object、copy_object、get_object、head_object 和 delete_object。

Bundle API

在单个流式 tar 归档中获取多个对象。专为机器学习训练设计，因为逐个对象往返请求的延迟占主导。

import tarfile
from tigris_boto3_ext import bundle_objects, BundleError, BUNDLE_ON_ERROR_FAIL

keys = [f"dataset/train/img_{i:05d}.jpg" for i in range(1000)]
response = bundle_objects(s3, "my-dataset-bucket", keys)

with tarfile.open(fileobj=response, mode="r|") as tar:
    for member in tar:
        if member.name == "__bundle_errors.json":
            continue  # error manifest — skip
        f = tar.extractfile(member)
        if f is not None:
            data = f.read()
            # feed to training pipeline

错误模式：

• 默认：缺失或失败的对象会在 tar 内的 __bundle_errors.json 中报告；允许部分成功。
• on_error=BUNDLE_ON_ERROR_FAIL：任何失败都会引发 BundleError，其中包含 status_code 和 body。适用于推理场景，因为此时每个键都必须存在。

try:
    response = bundle_objects(s3, "my-bucket", keys, on_error=BUNDLE_ON_ERROR_FAIL)
except BundleError as e:
    print(f"Bundle failed (HTTP {e.status_code}): {e.body}")

始终以流式模式打开 tar（mode="r|"）。不要将整个归档加载到内存中。

---

常见工作流

备份与恢复

import boto3
from tigris_boto3_ext import (
    create_snapshot_bucket, create_snapshot, create_fork, get_snapshot_version,
)

s3 = boto3.client("s3")

create_snapshot_bucket(s3, "production-data")

# Daily snapshot
result = create_snapshot(s3, "production-data", snapshot_name="daily-backup")
version = get_snapshot_version(result)

# Restore: fork the snapshot into a recovered bucket
create_fork(s3, "restored-data", "production-data", snapshot_version=version)

使用生产数据进行测试

result = create_snapshot(s3, "production-data", snapshot_name="for-tests")
version = get_snapshot_version(result)
create_fork(s3, "test-data", "production-data", snapshot_version=version)

try:
    # Writes to test-data don't touch production
    s3.put_object(Bucket="test-data", Key="experiment.txt", Body=b"...")
finally:
    s3.delete_bucket(Bucket="test-data")

时间回溯查询

from tigris_boto3_ext import get_object_from_snapshot, list_objects_from_snapshot

old = get_object_from_snapshot(s3, "my-bucket", "config.json", "12345")
config = old["Body"].read()

old_logs = list_objects_from_snapshot(
    s3, "my-bucket", "12345", Prefix="logs/2025/12/",
)
for obj in old_logs.get("Contents", []):
    print(obj["Key"])

---

工作原理

当上下文/装饰器处于活动状态时，tigris-boto3-ext 会在 boto3 的 before-sign.s3.* 事件上注册处理程序，以注入 Tigris 标头：

标头	用途
X-Tigris-Enable-Snapshot: true	在创建存储桶时启用快照
X-Tigris-Snapshot: true; name=<name>	创建命名快照
X-Tigris-Snapshot: <bucket>	列出存储桶的快照
X-Tigris-Snapshot-Version: <version>	从特定快照版本读取
X-Tigris-Fork-Source-Bucket: <bucket>	从此存储桶分叉
X-Tigris-Fork-Source-Bucket-Snapshot: <ver>	从特定快照分叉
X-Tigris-Rename: true	将 CopyObject 转换为原地重命名

HeadBucket 响应包含 X-Tigris-Enable-Snapshot、X-Tigris-Fork-Source-Bucket 和 X-Tigris-Fork-Source-Bucket-Snapshot——可通过 get_bucket_info() 和 has_snapshot_enabled() 获取。

---

重要规则

始终：

• 创建存储桶时启用快照——之后无法补启用
• 在调用 create_snapshot 后立即捕获 get_snapshot_version(result)——读取或分叉时需要它
• 以流式方式处理 Bundle API 响应（mode="r|"）——切勿将整个 tar 加载到内存中
• 遍历 tar 成员时跳过 __bundle_errors.json
• 严格控制 TigrisRename / with_rename 的作用范围——其中每个 copy_object 都会变成重命名操作

切勿：

• 将 region_name 硬编码为真实的 AWS 区域——请使用 auto
• 测试时直接使用生产存储桶，尤其是在分叉无需成本的情况下
• 忘记设置 AWS_ENDPOINT_URL_S3——否则 boto3 会访问 AWS S3，而不是 Tigris

---

已知问题

问题	原因与解决方法
快照调用失败，返回 404/409	存储桶未启用快照。使用 create_snapshot_bucket 重新创建。
copy_object 重命名了不相关的文件	TigrisRename 代码块范围太宽。仅将重命名调用置于其中。
遍历 Bundle 时卡住	tar 以非流式模式打开。请使用 mode="r|"，而不是 "r"。
意外引发 BundleError	传入了 BUNDLE_ON_ERROR_FAIL。默认模式会在 tar 中报告错误。
调用访问的是 AWS，而不是 Tigris	未设置 AWS_ENDPOINT_URL_S3。请导出该变量，或显式传入 endpoint_url=。
分叉未启用快照	分叉不会继承快照功能——请在分叉上单独启用。

---

CLI 速查

t3 是 tigris 的别名。

# Auth
tigris login
tigris whoami

# Buckets
tigris buckets create <name> [--public] [--locations <region>]
tigris buckets list
tigris buckets delete <name>

# Access keys
tigris access-keys create "<name>"
tigris access-keys assign <tid_xxx> --bucket <name> --role Editor

# Snapshots & forks (also doable from Python via this skill)
tigris buckets snapshot <bucket> --name <snapshot-name>
tigris buckets snapshots <bucket>
tigris buckets fork <new> --source <bucket> [--snapshot <version>]

---

相关技能

• file-storage — Tigris 的通用设置，以及 JS/TS @tigrisdata/storage SDK
• tigris-snapshots-forking — CLI 端的快照/分叉工作流
• tigris-agent-kit — Agent 存储工作流（用于沙箱和检查点的分叉）
• tigris-sdk-guide — 选择适用于不同语言的 SDK

官方文档

• 软件包：https://pypi.org/project/tigris-boto3-ext/
• 仓库：https://github.com/tigrisdata/tigris-boto3-ext
• Tigris 文档：https://www.tigrisdata.com/docs/