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

本指南以 lakeFS Python SDK（lakefs_sdk）的 PullsApi 为纲，系统讲解 pull request 数据版本控制的创建、查询、列表、更新与合并五个 API 端点，并深入服务端 Go 实现，帮助你掌握在数据湖上用代码驱动"分支评审-合并"工作流的完整方法。读完本文，你将能用 Python 编写一套可复用的 PR 自动化脚本，并理解其底层权限、校验与 KV 存储机制。

PullsApi 概览

PullsApi 是 lakeFS Python SDK 中负责 pull request（PR）资源的 API 类，所有请求路径均以/api/v1为前缀，围绕仓库下的/pulls资源展开。其生成的源码位于 clients/python/lakefs_sdk/api/pulls_api.py，对应的 OpenAPI 定义在 api/swagger.yml，该接口在规范中被标记为experimental。

五个方法及其 HTTP 映射如下：

每个方法都成对提供xxx_with_http_info变体，返回(响应体, 状态码, 响应头)三元组，便于需要原始 HTTP 细节的场景使用。

客户端初始化与认证

在使用 PullsApi 之前，需要构造lakefs_sdk.Configuration与lakefs_sdk.ApiClient。SDK 支持三种认证方式：HTTP 基本认证（basic_auth）、API Key 认证（cookie_auth）与 Bearer（JWT）认证（jwt_token），三者可任选其一，示例来自文档中的标准初始化模式（clients/python/README.md 有完整说明）。

import time import os import lakefs_sdk from lakefs_sdk.models.pull_request_creation import PullRequestCreation from lakefs_sdk.models.pull_request_creation_response import PullRequestCreationResponse from lakefs_sdk.rest import ApiException from pprint import pprint # 方式一：定义 host（可选，默认 /api/v1） configuration = lakefs_sdk.Configuration( host = "/api/v1" ) # 方式二：HTTP 基本认证 configuration = lakefs_sdk.Configuration( username = os.environ["USERNAME"], password = os.environ["PASSWORD"] ) # 方式三：API Key 认证（cookie_auth），可按需设置前缀 configuration.api_key['cookie_auth'] = os.environ["API_KEY"] # configuration.api_key_prefix['cookie_auth'] = 'Bearer' # 方式四：Bearer（JWT）认证 configuration = lakefs_sdk.Configuration( access_token = os.environ["BEARER_TOKEN"] ) # 进入 ApiClient 上下文，创建 API 实例 with lakefs_sdk.ApiClient(configuration) as api_client: api_instance = lakefs_sdk.PullsApi(api_client) # ... 在此调用 PullsApi 的各个方法

注意：basic_auth、cookie_auth、jwt_token三种授权方式的详细说明见 clients/python/README.md。服务端按basic_auth、cookie_auth、jwt_token顺序尝试认证（见 pulls_api.py 的_auth_settings）。

创建 Pull Request

create_pull_request(repository, pull_request_creation)用于在指定仓库中创建一条 PR，核心入参为PullRequestCreation模型。

请求模型 PullRequestCreation

该模型在 SDK 中由 pull_request_creation.py 定义，其 OpenAPI schema 见 swagger.yml。服务端在 pkg/catalog/catalog.go 中会校验：仓库 ID 与源/目标分支 ID 合法、源目标分支不能相同（否则返回ErrSameBranch）、且两个分支都必须真实存在。

创建示例

with lakefs_sdk.ApiClient(configuration) as api_client: api_instance = lakefs_sdk.PullsApi(api_client) repository = 'my-repo' # str | 仓库名 pull_request_creation = lakefs_sdk.PullRequestCreation( title="Add new data pipeline", description="Merge feature branch into main", source_branch="feature/pipeline", destination_branch="main", ) try: # create pull request api_response = api_instance.create_pull_request(repository, pull_request_creation) print("The response of PullsApi->create_pull_request:\n") pprint(api_response) # PullRequestCreationResponse(id='...') except Exception as e: print("Exception when calling PullsApi->create_pull_request: %s\n" % e)

返回与响应

• 返回类型：PullRequestCreationResponse，仅含一个id字段（PR 的 ID，后续查询、更新、合并都要用到）。
• 请求头：Content-Type: application/json，Accept: application/json。
• 状态码：

服务端在 pkg/api/controller.go 中实现CreatePullRequest：先校验调用者具备pr:WritePullRequest权限（见 pkg/permissions/actions.go 的权限常量定义），再从当前登录用户推导author（user.Committer()），最终由catalog.CreatePullRequest生成 PR ID 并落库。

查询 Pull Request 详情

get_pull_request(repository, pull_request)按 ID 获取单个 PR 的完整信息。

with lakefs_sdk.ApiClient(configuration) as api_client: api_instance = lakefs_sdk.PullsApi(api_client) repository = 'my-repo' # str | 仓库名 pull_request = 'pr-id-from-create-response' # str | pull request id try: # get pull request api_response = api_instance.get_pull_request(repository, pull_request) print("The response of PullsApi->get_pull_request:\n") pprint(api_response) # PullRequest 对象 except Exception as e: print("Exception when calling PullsApi->get_pull_request: %s\n" % e)

返回模型 PullRequest

查询返回 PullRequest，它继承PullRequestBasic（status、title、description），并增加以下字段：

status字段是枚举类型，取值仅限open、closed、merged三种（由 pull_request.py 中的 pydantic validator 强制校验）。OpenAPI 定义见 swagger.yml。

• 请求头：Accept: application/json（无请求体，Content-Type 未定义）。
• 状态码：200（返回PullRequest）、400、401、404、429、0。

底层实现：服务端 GetPullRequest 需要pr:ReadPullRequest权限，返回时将 graveler 层的PullRequest（定义于 pkg/graveler/graveler.go）转换为 API 模型，并补全MergedCommitId与ClosedDate。

分页列出 Pull Request

list_pull_requests(repository, prefix=..., after=..., amount=..., status=...)以游标分页方式列出仓库下符合条件的 PR，返回PullRequestsList（包含pagination与results数组，模型见 pull_requests_list.py，分页结构见 pagination.py）。

with lakefs_sdk.ApiClient(configuration) as api_client: api_instance = lakefs_sdk.PullsApi(api_client) repository = 'my-repo' # str | 仓库名 prefix = 'prefix_example' # str | 返回以该值作为前缀的条目（可选） after = 'after_example' # str | 返回该值之后的条目（可选） amount = 100 # int | 每页返回多少条（可选，默认 100） status = 'all' # str | 按状态过滤（可选，默认 'all'） try: # list pull requests api_response = api_instance.list_pull_requests(repository, prefix=prefix, after=after, amount=amount, status=status) print("The response of PullsApi->list_pull_requests:\n") pprint(api_response) except Exception as e: print("Exception when calling PullsApi->list_pull_requests: %s\n" % e)

查询参数

关于amount与status有两个来自源码的重要细节：

1. amount 的取值范围：SDK 生成代码将其声明为conint(strict=True, le=1000, ge=-1)（见 pulls_api.py），即合法区间为-1到1000。服务端 pkg/catalog/catalog.go 定义ListPullsLimitMax = 1000，越界或为负时会归一化为 1000。
2. status 的过滤语义：OpenAPI 定义中status枚举为[open, closed, all]，默认all（swagger.yml）。服务端 shouldSkipByStatus 的语义是：closed同时包含CLOSED与MERGED两类；open仅匹配OPEN；其余值（如merged）则返回全部。

• 请求头：Accept: application/json（无请求体）。
• 状态码：200（返回 PR 列表）、400、401、404、429、0。

底层实现：服务端 ListPullRequests 需要pr:ListPullRequests权限，其分页游标基于 PR ID 的字典序，配合prefix实现前缀过滤；翻页时用返回的pagination.next_offset作为下一次调用的after。同时注意：列表接口出于权限最小化考虑，不返回 description 字段（见 controller.go 的注释说明），完整描述需调用get_pull_request。

更新 Pull Request

update_pull_request(repository, pull_request, pull_request_basic)通过PATCH方法局部更新 PR 的标题、描述或状态。请求体为 PullRequestBasic，三个字段均为可选，语义为"提供哪个字段就更新哪个字段"：

with lakefs_sdk.ApiClient(configuration) as api_client: api_instance = lakefs_sdk.PullsApi(api_client) repository = 'my-repo' # str | 仓库名 pull_request = 'pr-id' # str | pull request id pull_request_basic = lakefs_sdk.PullRequestBasic( status="closed", description="Update: resolved by discussion", ) try: # update pull request api_instance.update_pull_request(repository, pull_request, pull_request_basic) except Exception as e: print("Exception when calling PullsApi->update_pull_request: %s\n" % e)

• 返回类型：void（无响应体）。
• 请求头：Content-Type: application/json，Accept: application/json。
• 状态码：200（更新成功）、400、401、403、404、429、0。

底层实现：服务端 UpdatePullRequest 需要pr:WritePullRequest权限。graveler 层的 UpdatePullRequest 采用"读取-修改-写回"模式：只对非 nil 字段做变更；当status变为closed或merged（即isPullClosed，见 graveler.go）时，会自动写入ClosedDate为当前时间。最终通过 ref/manager.go 的UpdatePullRequest以 KV 条件更新（SetMsgIf）完成持久化，避免并发覆盖。

合并 Pull Request

merge_pull_request(repository, pull_request)是 PR 工作流的收尾操作，通过PUT触发一次真实的合并：把 PR 的源分支合并到目标分支。

with lakefs_sdk.ApiClient(configuration) as api_client: api_instance = lakefs_sdk.PullsApi(api_client) repository = 'my-repo' # str | 仓库名 pull_request = 'pr-id' # str | pull request id try: # merge pull request api_response = api_instance.merge_pull_request(repository, pull_request) print("The response of PullsApi->merge_pull_request:\n") pprint(api_response) # MergeResult(reference='...') except Exception as e: print("Exception when calling PullsApi->merge_pull_request: %s\n" % e)

返回模型 MergeResult

合并成功返回 MergeResult，仅含reference字段——即合并后生成的 commit ID。

• 请求头：Accept: application/json（无请求体）。
• 状态码：

底层实现：服务端 MergePullRequest 的合并权限要求较特殊——需要同时具备目标分支上的fs:CreateCommit与仓库级的pr:WritePullRequest权限（NodeTypeAnd逻辑）。合并流程为：① 检查 PR 状态必须为OPEN，否则返回 400 "bad pull request status"；② 调用Catalog.Merge将源分支合并进目标分支；③ 合并成功后把 PR 状态更新为merged并写入合并产生的 commit ID（MergedCommitID）。合并失败时 PR 保持 open，可重试。

完整实战：一个 PR 生命周期脚本

综合以上 API，下面给出一个端到端示例：创建 PR → 查询详情 → 分页列出 → 更新状态 → 合并，并打印每个环节的响应。

import os import lakefs_sdk from lakefs_sdk.rest import ApiException from pprint import pprint configuration = lakefs_sdk.Configuration( host="http://localhost:8000/api/v1", # 替换为你的 lakeFS 端点 username=os.environ["LAKEFS_ACCESS_KEY"], password=os.environ["LAKEFS_SECRET_KEY"], ) with lakefs_sdk.ApiClient(configuration) as api_client: pulls = lakefs_sdk.PullsApi(api_client) repo = "my-repo" try: # 1. 创建 PR created = pulls.create_pull_request( repo, lakefs_sdk.PullRequestCreation( title="Add weekly ETL job", description="Introduces the weekly aggregation pipeline", source_branch="feature/weekly-etl", destination_branch="main", ), ) pr_id = created.id print("created PR:", pr_id) # 2. 查询详情 detail = pulls.get_pull_request(repo, pr_id) pprint(detail.to_dict()) # 3. 列出 open 状态的 PR（每页 20 条） listing = pulls.list_pull_requests(repo, amount=20, status="open") print("has_more:", listing.pagination.has_more) # 4. 更新描述 pulls.update_pull_request( repo, pr_id, lakefs_sdk.PullRequestBasic(description="Updated after review"), ) # 5. 合并 merged = pulls.merge_pull_request(repo, pr_id) print("merged commit reference:", merged.reference) # 合并后查询，可见 status=merged 与 merged_commit_id after = pulls.get_pull_request(repo, pr_id) pprint(after.to_dict()) except ApiException as e: print("Exception when calling PullsApi: %s\n" % e)

与服务端实现对照：数据流全景

从 Python SDK 到最终落库，一次 PR 操作经过以下层次（对应仓库中的真实文件）：

1. Python SDK 层：PullsApi方法构造 HTTP 请求，路径形如/repositories/{repository}/pulls，由 clients/python/lakefs_sdk/api/pulls_api.py 生成，经ApiClient.call_api发送。
2. API 规范层：路由与 schema 由 api/swagger.yml 定义，SDK 正是由该 OpenAPI 文件生成的（Python 客户端生成配置见 clients/python/python-static/python-codegen-config.yaml）。
3. HTTP Controller 层：Go 服务端 pkg/api/controller.go 负责鉴权（pr:ReadPullRequest/pr:WritePullRequest/pr:ListPullRequests，见 pkg/permissions/actions.go）、审计日志（LogAction）与参数解析。
4. Catalog 层：pkg/catalog/catalog.go 负责输入校验、ID 生成（graveler.NewRunID()）、状态过滤与分页。
5. Graveler/Ref 层：pkg/graveler/graveler.go 定义 PR 领域模型与更新语义；pkg/graveler/ref/manager.go 通过 KV 存储持久化——PR 本体存在主键分区，同时按"源分支-目标分支"建立二级索引（PullBySrcDstPath），使用SetMsgIf条件写保证并发安全。

值得一提的是，PR 功能的设计初衷记录在仓库的 design/open/pull-request.md：其目标是"把变更请求建模为仓库级资源"，支持评审、讨论与合并前的自动化校验（actions）。当前仓库实现聚焦于 PR 的创建、查询、列表、更新与合并核心流程。

常见问题与注意事项

• 401/403 与权限：若调用报Unauthorized或Forbidden，请检查认证方式是否与服务端配置一致，以及当前用户是否具备对应仓库的pr:ReadPullRequest（读）、pr:WritePullRequest（写/更新/合并）、pr:ListPullRequests（列表）权限。
• 409 Conflict（创建/合并）：创建时目标资源冲突（如同一源-目标分支组合已存在 PR），合并时则代表分支合并冲突——服务端明确提示 409 的响应内容 schema 未来会从空MergeResult迁移到Error格式，注意兼容处理。
• amount 越界：传入的amount超过 1000 或小于 0 时，服务端会归一化为 1000；SDK 侧也通过 pydantic 约束在-1~1000之间，客户端请求即校验。
• status 过滤语义：status=closed同时覆盖closed与merged两类 PR，若只想看已合并的 PR，需要自行结合返回数据的status字段二次过滤。
• 列表不含 description：list_pull_requests出于权限考虑不返回描述字段，需要完整信息请对每个 PR 调用get_pull_request。

延伸阅读

• API 端点总览与认证方式：clients/python/README.md
• PullsApi 生成源码：clients/python/lakefs_sdk/api/pulls_api.py
• 相关数据模型：PullRequest、PullRequestCreation、PullRequestBasic、PullRequestsList、MergeResult、Pagination
• OpenAPI 定义：api/swagger.yml
• 服务端实现：pkg/api/controller.go、pkg/catalog/catalog.go、pkg/graveler/graveler.go、pkg/graveler/ref/manager.go
• 功能设计文档：design/open/pull-request.md

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