Cerebral Python SDK

Tilde Python SDK

Python SDK 波浪号 数据版本控制API。

安装

pip install tilde-sdk

或与 紫外线:

uv add tilde-sdk

需要Python 3.11+。

快速开始

import tilde

repo = tilde.repository("my-org/my-repo")

# Run commands in an interactive sandbox
with repo.shell(image="python:3.12") as sh:
    sh.run("pip install pandas")
    result = sh.run("python train.py")
    print(result.stdout.text())

    # Stream output line by line
    result = sh.run("cat /sandbox/results.csv")
    for line in result.stdout.iter_lines():
        print(line)

\[！重要\] 默认情况下为事务性。 在沙盒中进行的所有文件系统修改 发生在事务会话的上下文中。如果中途有任何事情失败或 中止，更改不生效。只有成功的沙盒更改是 以原子方式提交到存储库，因此您的数据始终位于 一致的状态。看 会话 了解更多详情。

认证

SDK按以下顺序解析凭据（第一场比赛获胜）：

1. 显式参数 — Client(api_key=...) 或 tilde.configure(api_key=...).
2. 环境变量 — TILDE_API_KEY.
3. CLI配置文件 — ~/.tilde/config.yaml （作者： tilde auth login).
4. 沙盒元数据端点 — TILDE_SANDBOX_CREDENTIALS_URI （自动检测

仅当没有配置静态密钥时；由沙盒运行时注入代码 跑步 *里面* 蒂尔德沙盒）。

静态密钥总是胜过沙盒自动检测，因此设置了凭据 呼叫者的故意行为永远不会被无声地覆盖。

使用CLI保存的凭据

之后 tilde auth login，SDK会自动获取凭据——否 需要env-var或代码配置：

import tilde

repo = tilde.repository("my-org/my-repo")

环境变量

适用于未运行CLI的CI/CD、代理工作流和Docker容器：

export TILDE_API_KEY="your-api-key"
export TILDE_ENDPOINT_URL="https://tilde.run"  # optional

模块级配置

import tilde

tilde.configure(api_key="your-api-key")
repo = tilde.repository("my-org/my-repo")

显式客户端

当您需要多种配置或完全控制时，请使用显式客户端 在HTTP生命周期中：

from tilde import Client

with Client(api_key="your-api-key") as client:
    repo = client.repository("my-org/my-repo")

缺少API密钥在构造时不是错误；一 ConfigurationError 在发出第一个请求时提出。

配置参考

沙盒执行

针对存储库数据运行代码的最快方法。沙盒执行 在将数据装载为卷的隔离容器中——每次更改都是 被捕获为交易。

交互式 shell

使用 repo.shell() 要在单个沙箱中运行多个命令：

with repo.shell(image="python:3.12") as sh:
    sh.run("pip install pandas")
    result = sh.run("python train.py")
    print(result.stdout.text())
    print(result.stderr.text())

shell是一个上下文管理器——在干净退出时自动提交更改， 除此之外，它们还会倒退。每 sh.run() 培养独立的孩子 进程：shell状态(cd、导出变量、别名） 不 持续存在 电话。一次调用中的链命令(sh.run("cd /foo && ls"))或通过 env= / cwd= 到 run().

shell.run() 返回a RunResult 和 stdout, stderr，以及 exit_code. 通过 check=True 提高 CommandError 非零出口。

一枪处决

对于不需要交互式会话的单个命令：

result = repo.execute("python train.py", image="python:3.12")
print(result.stdout.text())

# check=False to handle errors yourself
result = repo.execute("might-fail", image="python:3.12", check=False)
if result.exit_code != 0:
    print(result.stdout.text())

输出流

RunResult.stdout 和 RunResult.stderr 是 OutputStream 实例：

仓库

import tilde

# Shorthand: the only top-level shortcut in the SDK
repo = tilde.repository("my-org/my-repo")

# Lazy-loaded properties
print(repo.id, repo.description, repo.visibility)

# Update
repo.update(description="New description", visibility="public")

# Delete (soft delete)
repo.delete()

通过其组织创建存储库：

org = tilde.organizations.get("my-org")
repo = org.repositories.create(
    "my-repo",
    description="My dataset",
    session_max_duration_days=7,
    retention_days=90,
)

for r in org.repositories.list():
    print(r.name, r.description)

提交

# Newest first, auto-paginating
for commit in repo.commits.list():
    print(commit.id, commit.committer, commit.message)

# Cap results
for commit in repo.commits.list(amount=10):
    print(commit.id)

# Look up by ID
commit = repo.commits.get("a1b2c3d4e5f6")

# Diff introduced by a commit
for change in commit.diff():
    print(change.status, change.path)

# Revert
revert = commit.revert(message="undo")

分页

每 .list() 返回a PaginatedIterator 它懒洋洋地翻着几页。二 关键字参数调优迭代：

for entry in commit.objects.list(prefix="data/", page_size=500):
    process(entry)

会话

会话提供对对象的直接事务访问。更喜欢沙盒 执行以运行代码；会话用于细粒度对象操作 从你自己的过程。

# Context manager — rolls back on error
with repo.session() as session:
    session.objects.put("data/report.csv", b"content")
    session.objects.delete("data/old.csv")
    session.commit("update CSV files")

# Explicit control
session = repo.session()
print(session.session_id)
session.objects.put("data/file.csv", b"content")
session.commit("modifying data")
# or: session.rollback()

从另一个线程、进程或计算机恢复会话：

session = repo.attach(session_id)
session.objects.put("data/file2.csv", b"more content")
session.commit("finishing work")

对象

写作

import pathlib

with repo.session() as session:
    # From bytes
    session.objects.put("data/hello.txt", b"Hello, Tilde!")

    # From a file
    with open("dataset.parquet", "rb") as f:
        session.objects.put("data/dataset.parquet", f)

    # From a Path (opened/closed automatically)
    session.objects.put("data/model.bin", pathlib.Path("model.bin"))

    # Server-side copy (no download/upload)
    session.objects.copy("data/hello.txt", "data/hello-backup.txt")

    # Delete
    session.objects.delete("data/old.csv")
    session.objects.delete_many(["data/a.csv", "data/b.csv"])

    session.commit("upload files")

≥64MB的文件自动使用多部分上传；较小的文件使用单个 主持PUT。无需更改代码。

阅读

可以从已提交的快照中读取对象（通过 Commit)或从内部 会话（包括未提交的阶段性更改）：

# From a commit
commit = next(iter(repo.commits.list()))
with commit.objects.get("data/hello.txt") as f:
    print(f.read().decode())

# Within a session
with repo.session() as session:
    with session.objects.get("data/file.csv") as f:
        data = f.read()

    # Metadata only
    meta = session.objects.head("data/file.csv")
    print(meta.etag, meta.content_type, meta.content_length)

流媒体和字节范围

大型对象：禁用缓存并分块流式传输：

with commit.objects.get("data/large.bin", cache=False) as f:
    for chunk in f.iter_bytes(chunk_size=1024 * 1024):
        output.write(chunk)

字节范围适用于文件头、日志尾或Parquet等列格式：

# First 4 bytes
with commit.objects.get("data/file.parquet", byte_range=(0, 3)) as f:
    magic = f.read()

# From offset 1024 to end
with commit.objects.get("data/file.parquet", byte_range=(1024, None)) as f:
    tail = f.read()
    print(f.content_range)   # "bytes 1024-49151/49152"
    print(f.content_length)  # bytes returned

列表

for entry in commit.objects.list(prefix="data/", delimiter="/"):
    print(entry.path, entry.type)  # "object" or "prefix"

未承诺的变更

session = repo.session()
session.objects.put("data/new.csv", b"content")
for entry in session.uncommitted():
    print(entry.path)

组织

import tilde

# Create, list, get, delete at the module level
org = tilde.organizations.create("my-org", display_name="My Org")
for o in tilde.organizations.list():
    print(o.name, o.display_name)
org = tilde.organizations.get("my-org")
tilde.organizations.delete("my-org")

一 Organization 将每个组织范围的子资源作为属性公开：

org = tilde.organizations.get("my-org")
org.repositories
org.members
org.agents
org.roles
org.groups
org.policies
org.connectors

成员

for m in org.members.list():
    print(m.username, m.email)

org.members.create("alice")          # add by username
org.members.delete("user-uuid")      # remove by user_id

群组

groups = org.groups

group = groups.create("engineers", description="Engineering team")
group = groups.get(group.id)                                # includes members + attachments
group.members.create(subject_type="user", subject_id="user-uuid")
group.members.delete(subject_type="user", subject_id="user-uuid")
group.update(name="engineering", description="Updated")
group.delete()

# Effective groups for a principal
for g in groups.effective(principal_type="user", principal_id="user-uuid"):
    print(g.group_name, g.source)

政策

policies = org.policies

# Validate before creating
result = policies.validate("GetObject()")
print(result.valid, result.errors)

policy = policies.create(
    name="read-only",
    policy_text="ListRepositories()\nGetRepository()\nGetObject()\n",
    description="Read-only access",
)

# Generate from natural language
text = policies.generate("Allow read-only access to all repositories")

# Attach / detach
policy.attach(principal_type="group", principal_id="group-uuid")
policy.detach(principal_type="group", principal_id="group-uuid")

# Effective policies for a principal
for ep in policies.effective(principal_type="user", principal_id="user-uuid"):
    print(ep.policy_name, ep.source)

角色和代理

通过API密钥验证的非人类身份：

• 角色 (trk- 前缀）——CI/CD管道，无元数据的自动化。
• 代理 (tak- 前缀）——带有元数据的自动化工具和人工智能助手

以及可选的内联策略。

# Roles
role = org.roles.create("ci-deployer", description="CI/CD pipeline")
for role in org.roles.list():
    print(role.name)

# Agents
agent = org.agents.create(
    "data-pipeline",
    description="Nightly pipeline",
    metadata={"env": "prod"},
)
agent.update(metadata={"env": "staging"})
agent.update(inline_policy="ListRepositories()\nGetObject()\n")

API密钥

agent = org.agents.get("data-pipeline")

# Create a key (the full token is only shown once)
created = agent.api_keys.create("pipeline-key")
print(created.token)              # tak-...

for key in agent.api_keys.list():
    print(key.name, key.token_hint)

key = agent.api_keys.get(key_id)
key.revoke()

相同 api_keys 集合存在于 Role 实例（前缀为令牌 trk-).

秘密

加密的键值对作为环境变量注入沙盒。

# Repository-scoped
repo.secrets.create("API_KEY", "sk-abc123...")
secret = repo.secrets.get("API_KEY")
print(secret.value)
repo.secrets.delete("API_KEY")

# Agent-scoped (override repo secrets with the same key)
agent = org.agents.get("data-pipeline")
agent.secrets.create("OPENAI_KEY", "sk-abc123...")

沙盒启动时的优先级（从高到低）：

1. env= 在沙盒请求中传递
2. 代理机密（如果以代理身份运行）
3. 存储库机密

连接器和进口

connectors = org.connectors

# S3
conn = connectors.create(
    name="production-s3",
    type="s3",
    source_uri="s3://my-bucket/datasets/",
    config={
        "access_key_id": "AKIA...",
        "secret_access_key": "...",
        "region": "us-west-2",
    },
)

# Attach to a repo
repo.connectors.attach(conn.id)
for c in repo.connectors.list():
    print(c.name, c.type)

从连接器导入：

import time

job = repo.imports.create_from_connector(
    connector_id=conn.id,
    destination_path="imported/",
    source_prefix="datasets/",
    commit_message="Import production datasets",
)

while job.status not in ("completed", "failed"):
    time.sleep(2)
    job.refresh()
    print(job.status, job.objects_imported)

if job.status == "completed":
    print(f"Import done! Commit: {job.commit_id}")

跨存储库将副本数据从一个Tilde存储库导入到另一个：

job = repo.imports.create_from_repository(
    repo_path="other-org/source-data",
    destination_path="external/",
    source_prefix="datasets/train/",
    commit_message="Import training data",
)

代理审批工作流

当存储库需要批准代理提交时， session.commit() 默认情况下会阻止，直到有人批准或回滚。批准URL为 作为a发射 UserWarning.

with repo.session() as session:
    session.objects.put("data/results.csv", b"col1,col2\na,b\n")
    session.commit("add results")  # blocks until approved

非阻塞：

result = session.commit("add results", block_for_approval=False)
# result is None; session stays open for review

结构化结果（无阻塞，无警告）：

result = session.commit_result("add results")
if result.status == "committed":
    print(result.commit_id)
elif result.status == "approval_required":
    print(result.web_url)

低级沙盒和触发器

repo.shell() 和 repo.execute() 满足大多数需求。用于直接控制 异步生命周期、触发器和委托，使用 repo.sandboxes 和 repo.sandbox_triggers。请参阅 完整文档 了解详情。

错误处理

所有SDK异常都继承自 TildeError:

TildeError                           # base for all SDK errors
├── ConfigurationError               # missing API key, bad endpoint
├── TransportError                   # network failures, DNS, timeouts
├── SerializationError               # invalid JSON in response
├── SandboxError                     # sandbox lifecycle failure
├── CommandError                     # non-zero exit (repo.execute / shell.run(check=True))
└── APIError                         # base for HTTP API errors
    ├── BadRequestError              # 400
    ├── AuthenticationError          # 401
    ├── ForbiddenError               # 403
    ├── NotFoundError                # 404
    ├── ConflictError                # 409
    ├── GoneError                    # 410
    ├── PreconditionFailedError      # 412
    ├── LockedError                  # 423
    └── ServerError                  # 5xx

APIError 携带 status_code, message, code, request_id, method, url，以及 response_text.

from tilde import NotFoundError, TildeError

try:
    with repo.session() as session:
        with session.objects.get("nonexistent") as f:
            f.read()
except NotFoundError as e:
    print(f"Not found: {e.message} (request_id={e.request_id})")
except TildeError as e:
    print(f"SDK error: {e}")

文档

完整文档可在 .

发展

# Install dev dependencies
uv sync --all-extras

# Run tests
uv run pytest

# Lint and format
uv run ruff check src/ tests/
uv run ruff format src/ tests/

# Type check
uv run mypy src/tilde/

# Build
uv build

许可证

Apache 2.0