lakeFS Python SDK StagingApi 详解：两阶段对象上传与暂存链接机制-开发者社区

数据工程
数据湖
大数据
对象存储
后端

【免费下载链接】lakeFS

lakeFS - Data version control for your data lake | Git for data

项目地址：https://gitcode.com/gh_mirrors/la/lakeFS

导读

StagingApi 是 lakeFS Python SDK（lakefs_sdk）中负责"暂存（staging）"写入路径的核心 API：它通过get_physical_address与link_physical_address两个方法，把"上传数据对象到对象存储"和"把对象登记到 lakeFS 仓库"两个动作解耦，从而在不经过 lakeFS 服务器中转数据的情况下安全写入对象。阅读完本文，你将掌握 lakeFS 暂存写入的完整工作流、StagingLocation/StagingMetadata/ObjectStats三个关键模型的字段语义、presign与 ETag 条件写入的实战用法，以及服务端与 SDK 中的对应实现细节。

一、StagingApi 概览：两个方法完成一次暂存写入

StagingApi 的全部端点都挂在/api/v1之下，共两个方法，对应两条 HTTP 请求路径：

方法	HTTP 请求	描述
`get_physical_address`	`GET /repositories/{repository}/branches/{branch}/staging/backing`	生成一个客户端可以直接向其上传对象的地址
`link_physical_address`	`PUT /repositories/{repository}/branches/{branch}/staging/backing`	把该物理地址上的对象与仓库内的一个路径关联，产生一次未提交（uncommitted）变更

方法签名定义在 clients/python/lakefs_sdk/api/staging_api.py 中：

def get_physical_address(self, repository, branch, path, presign=None, **kwargs) -> StagingLocation def link_physical_address(self, repository, branch, path, staging_metadata, if_none_match=None, if_match=None, **kwargs) -> ObjectStats

调用示例、参数表与返回类型详见 clients/python/docs/StagingApi.md；服务端的行为契约定义在 api/swagger.yml 的/repositories/{repository}/branches/{branch}/staging/backing路径下（operationId 分别为getPhysicalAddress与linkPhysicalAddress）。

1.1 两种认证方式

两个方法都支持三种认证，SDK 生成的客户端在 staging_api.py 中固定声明_auth_settings = ['basic_auth', 'cookie_auth', 'jwt_token']：

basic_auth：HTTP Basic 认证，配置username与password；
cookie_auth：API Key（Cookie），通过configuration.api_key['cookie_auth']设置，需要前缀时可用configuration.api_key_prefix['cookie_auth'] = 'Bearer'；
jwt_token：Bearer JWT，通过Configuration(access_token=...)配置。

1.2 为什么需要"暂存"两步

lakeFS 的对象存储（blockstore）直接面向 S3、GCS、Azure Blob 等后端。两步设计的好处是：

数据不经过 lakeFS 服务器：客户端直接向对象存储上传数据（或直接引用外部已存在的对象地址），避免服务器吞吐成为瓶颈；
元数据与数据分离：上传完成后再通过link_physical_address把对象的 checksum、大小、路径等元数据登记进仓库，形成未提交变更，随后可用Commit固化；
支持外部地址：link_physical_address允许关联"仓库存储命名空间之外"的地址，为数据导入、跨仓库引用等场景提供可能。

二、get_physical_address：生成上传地址

2.1 请求参数

参数	类型	必填	说明
`repository`	`str`	是	仓库名称（路径参数）
`branch`	`str`	是	分支名称（路径参数）
`path`	`str`	是	相对分支的对象路径（查询参数）
`presign`	`bool`	否	是否生成预签名 URL

2.2 返回模型 StagingLocation

返回类型为StagingLocation，定义在 clients/python/lakefs_sdk/models/staging_location.py，共三个字段：

字段	类型	说明
`physical_address`	`str`	对象在后端对象存储上的原生 URI（如`s3://...`、`gs://...`），或者当`presign=true`时为可供 HTTP GET/PUT 使用的 URL
`presigned_url`	`str`	仅当请求传`presign=true`时返回，是可直接用于上传的预签名 URL
`presigned_url_expiry`	`int`	若存在且非零，表示`physical_address`作为预签名 URL 的过期 Unix Epoch 时间；该时间可能短于预签名 URL 本身的生命周期（例如认证 token 即将过期时）

2.3 完整调用示例

import time import os import lakefs_sdk from lakefs_sdk.models.staging_location import StagingLocation from lakefs_sdk.rest import ApiException from pprint import pprint # 配置认证：Basic configuration = lakefs_sdk.Configuration( username=os.environ["USERNAME"], password=os.environ["PASSWORD"] ) # 或配置 JWT # configuration = lakefs_sdk.Configuration(access_token=os.environ["BEARER_TOKEN"]) with lakefs_sdk.ApiClient(configuration) as api_client: api_instance = lakefs_sdk.StagingApi(api_client) repository = 'repo_name' branch = 'main' path = 'data/part-0001.parquet' # 相对分支的路径 presign = True api_response = api_instance.get_physical_address(repository, branch, path, presign=presign) pprint(api_response)

HTTP 请求头：Accept: application/json，无请求体。状态码：200（成功，返回暂存物理地址）、400、401、404、429，其余为500内部错误。

2.4 服务端实现

在 pkg/api/controller.go 的GetPhysicalAddress处理器中，服务端完成以下工作：

鉴权：要求WriteObjectAction权限（controller.go）；
通过Catalog.GetAddressWithSignature为对象生成带签名的物理地址；
通过BlockAdapter.ResolveNamespace把相对地址解析为完整对象存储 URI；
若presign=true，调用BlockAdapter.GetPreSignedURL(..., block.PreSignModeWrite, "")生成写入型预签名 URL，并填充PresignedUrl与PresignedUrlExpiry字段（controller.go）。

三、link_physical_address：把物理地址与路径关联

3.1 请求参数

参数	类型	必填	说明
`repository`	`str`	是	仓库名称
`branch`	`str`	是	分支名称
`path`	`str`	是	相对分支的对象路径
`staging_metadata`	`StagingMetadata`	是	上传对象的信息（JSON 请求体）
`if_none_match`	`str`	否	设为`"*"`表示"仅当该路径尚无对象时才允许写入"，其它值不被支持
`if_match`	`str`	否	设为对象的 ETag，表示"仅当对象当前 ETag 与给定值匹配时才允许操作"

if_none_match与if_match在 SDK 生成代码中分别映射为 HTTP 头If-None-Match与If-Match（见 staging_api.py），二者提供原子化的写入条件，用于避免并发覆盖或重复写入。

3.2 请求体模型 StagingMetadata

StagingMetadata定义在 clients/python/lakefs_sdk/models/staging_metadata.py，字段如下：

字段	类型	必填	说明
`staging`	`StagingLocation`	是	上传地址（通常直接回传`get_physical_address`的结果）
`checksum`	`str`	是	对象内容在后端存储上的唯一标识（通常是 ETag）
`size_bytes`	`int`	是	对象字节大小
`user_metadata`	`dict[str, str]`	否	用户自定义元数据
`content_type`	`str`	否	对象媒体类型
`mtime`	`int`	否	Unix Epoch 秒数；服务端可能忽略
`force`	`bool`	否	默认`False`，是否强制覆盖写入

3.3 返回模型 ObjectStats

link_physical_address成功返回ObjectStats（200 状态码），定义在 clients/python/lakefs_sdk/models/object_stats.py。关键字段：

字段	类型	说明
`path`	`str`	对象路径
`path_type`	`str`	枚举：`common_prefix`或`object`
`physical_address`	`str`	对象在后端对象存储上的位置，原生 URI（`s3://...`、`gs://...`等）或`presign=true`时的 HTTP URL
`physical_address_expiry`	`int`	预签名 URL 过期时间（可选）
`checksum`	`str`	对象校验和
`size_bytes`	`int`	字节数
`mtime`	`int`	Unix Epoch 秒数
`metadata`	`dict[str, str]`	对象元数据
`content_type`	`str`	对象媒体类型

3.4 完整调用示例

import os import lakefs_sdk from lakefs_sdk.models.object_stats import ObjectStats from lakefs_sdk.models.staging_metadata import StagingMetadata from lakefs_sdk.rest import ApiException from pprint import pprint configuration = lakefs_sdk.Configuration( username=os.environ["USERNAME"], password=os.environ["PASSWORD"] ) with lakefs_sdk.ApiClient(configuration) as api_client: api_instance = lakefs_sdk.StagingApi(api_client) repository = 'repo_name' branch = 'main' path = 'data/part-0001.parquet' # 第一步：取得上传地址（假定此前已完成上传并拿到 ETag） staging_location = lakefs_sdk.StagingLocation( physical_address="s3://.../...", presigned_url="https://...", ) staging_metadata = lakefs_sdk.StagingMetadata( staging=staging_location, checksum="2e9ec317e197e02e4264d128c2e7e681", # 后端返回的 ETag size_bytes=1048576, user_metadata={"team": "data-eng"}, content_type="application/octet-stream", mtime=None, force=False, ) if_none_match = '*' # 仅当路径尚无对象时允许写入 if_match = None # 或传入对象当前 ETag 做条件写入 api_response = api_instance.link_physical_address( repository, branch, path, staging_metadata, if_none_match=if_none_match, if_match=if_match, ) pprint(api_response)

HTTP 请求头：Content-Type: application/json，Accept: application/json。状态码语义：

状态码	含义
`200`	成功，返回对象元数据（`ObjectStats`）
`400`	校验错误（如 checksum 为空）
`401`	未认证
`403`	无权限
`404`	资源不存在
`409`	与某次提交冲突，返回`StagingLocation`供重试
`412`	前置条件失败（`If-None-Match`/`If-Match`不满足）
`429`	请求过多
`501`	未实现
`0`/ 默认	内部服务器错误

3.5 服务端实现

controller.go 的LinkPhysicalAddress处理器揭示了完整流程：

鉴权要求WriteObjectAction（controller.go）；
校验请求体中的 blockstore 类型与仓库存储配置一致；
若addressType == catalog.AddressTypeRelative（地址位于仓库存储命名空间内），调用Catalog.VerifyLinkAddress校验该地址确由 lakeFS 签发（controller.go）；命名空间之外的地址则直接接受；
使用httputil.StripQuotesAndSpaces清洗 ETag，若为空则返回 400（controller.go）；
通过catalog.NewDBEntryBuilder()构造对象条目（路径、物理地址、大小、checksum、ContentType、元数据），并用apiutil.BuildConditionFromParams将if_match/if_none_match转换为 graveler 条件（controller.go）；
Catalog.CreateEntry(..., graveler.WithForce(...), graveler.WithCondition(...))写入未提交条目（controller.go）。

这里的CreateEntry走的是 graveler 层的 staging 管理：pkg/graveler/staging/目录下的实现（如 pkg/graveler/staging/manager_test.go）即负责未提交（staged）数据的读写，最终这些暂存条目会在 Commit 时被固化。

四、完整实战：预签名两段式上传

结合presign=true与两个 StagingApi 方法，可以实现不经过 lakeFS 服务器的直传流程（参考官方端到端测试 esti/presign_test.go 中的preSignGetPhysicalAddress用例）：

获取预签名上传地址：GET .../staging/backing?path=foo/uploaded&presign=true，得到StagingLocation（含presigned_url）；
直传数据：向presigned_url发起 HTTPPUT，携带对象内容与Content-Type，从响应头Etag取得校验和；
链接对象：PUT .../staging/backing，请求体StagingMetadata中staging回传第一步的结果、checksum填 ETag、size_bytes填实际字节数；
读取验证：GET Object拉回对象内容，确认与上传内容一致。

Python 代码骨架：

import requests # 1. 获取预签名地址 loc = api_instance.get_physical_address(repo, branch, "foo/uploaded", presign=True) # 2. 直传对象到预签名 URL resp = requests.put(loc.presigned_url, data=obj_content, headers={"Content-Type": "application/octet-stream"}) etag = resp.headers.get("Etag") # 3. 链接物理地址与路径 stats = api_instance.link_physical_address( repo, branch, "foo/uploaded", staging_metadata=lakefs_sdk.StagingMetadata( staging=loc, checksum=etag, size_bytes=len(obj_content), content_type="application/octet-stream", ), )

4.1 高阶客户端中的实际用法

在更上层的 Python 封装clients/python-wrapper中，预签名上传正是依赖 StagingApi 完成的。见 clients/python-wrapper/lakefs/object.py 的_upload_presign实现：

staging_location = self._client.sdk_client.staging_api.get_physical_address( self._obj.repo, self._obj.ref, self._obj.path, True) url = staging_location.presigned_url # ... 直接向 url 发起 PUT，提取 ETag ... staging_metadata = StagingMetadata(staging=staging_location, size_bytes=size_bytes, checksum=etag, user_metadata=self.metadata, content_type=self.content_type, mtime=None) if_none_match = "*" if self._mode.startswith("x") else None return self._client.sdk_client.staging_api.link_physical_address( self._obj.repo, self._obj.ref, self._obj.path, staging_metadata=staging_metadata, if_none_match=if_none_match)

注意两个细节：写入模式为x（排他创建）时自动设置if_none_match="*"，服务器返回 412 时映射为ObjectExistsException；Azure blockstore 场景下还会额外添加x-ms-blob-type: BlockBlob头（object.py）。对应的单测见 clients/python-wrapper/tests/utests/test_object.py。

Go 侧参考实现位于 pkg/api/helpers/upload.go 的uploadObject：它同样先调用getPhysicalAddress（Presign=true），用http.MethodPut直传，再以LinkPhysicalAddressJSONRequestBody关联元数据；若响应为 409（提交冲突），返回ErrConflict供上层重试。

五、条件写入与冲突处理最佳实践

排他创建（不存在才写）：传if_none_match='*'，服务端仅在路径无对象时写入，返回 412 表示对象已存在；这是实现"对象级原子创建"的推荐手段。
乐观并发更新（ETag 匹配才写）：传if_match=当前ETag，仅当对象 ETag 与给定值一致时允许覆盖，避免"读-改-写"之间的并发丢失更新。
提交冲突（409）：链接发生在某次提交进行时可能返回 409 并携带新的StagingLocation，客户端应重新执行"获取地址 → 上传 → 链接"流程（Go 客户端在 upload.go 中实现了带重试的上传循环）。
外部地址链接：link_physical_address接受仓库存储命名空间之外的物理地址，但命名空间内的地址会被校验签名，防止伪造 lakeFS 地址。

六、小结

StagingApi 是 lakeFS 写入路径的关键拼图：get_physical_address解决"往哪传"，link_physical_address解决"传完怎么登记"。配合presign与 ETag 条件头，可以在保证原子性与并发安全的前提下实现高吞吐的对象直传。若要进一步探索：

API 契约：查看 api/swagger.yml 中staging标签下的getPhysicalAddress/linkPhysicalAddress定义；
服务端实现：阅读 pkg/api/controller.go；
端到端验证：运行或阅读 esti/presign_test.go；
上层封装：参考 clients/python-wrapper/lakefs/object.py 与 pkg/api/helpers/upload.go。