PostHog 数据仓库 Factorial（HRIS）连接器 API 对接全解析：连接、分页、版本策略与增量同步决策-平芜编程栈

PostHog 数据仓库 Factorial（HRIS）连接器 API 对接全解析：连接、分页、版本策略与增量同步决策

【免费下载链接】posthog:hedgehog: PostHog is the leading platform for building self-driving products. Our developer tools – AI observability, analytics, session replay, flags, experiments, error tracking, logs, and more – capture all the context agents need to diagnose problems, uncover opportunities, and ship fixes. Steer it all from Slack, web, desktop, or the MCP.项目地址: https://gitcode.com/GitHub_Trending/po/posthog

本文是 PostHog 数据仓库warehouse_sources模块中 Factorial（HRIS）连接器的技术对接清单（API inventory）详解。文章以 api_inventory.md 为骨架，结合 factorial.py、settings.py、source.py 等源码实现，完整展开 Factorial API 的连接方式、游标分页、增量同步策略、17 个同步端点的清单与分区设计、版本生命周期与重钉迁移。读者读完后既能掌握该连接器的完整工作原理，也能了解 PostHog 数据仓库 REST 源接入（REST source）的通用实现范式。

文档定位与整体架构

api_inventory.md是 PostHog 仓库中针对 Factorial 连接器的"源本地笔记"（source-local notes），记录的是以代码为准、经官方文档交叉核对的 API 事实清单。它位于products/warehouse_sources/backend/temporal/data_imports/sources/factorial/目录下，与该目录中的factorial.py（核心实现）、settings.py（端点与字段配置）、source.py（连接器注册与元数据）、canonical_descriptions.py（端点/字段描述）、tests/（测试）共同构成一个完整的 REST 数据源接入单元。

PostHog 数据仓库源接入采用统一骨架：每个外部源通过common/rest_source抽象层（RESTAPIConfig、BasePaginator、rest_api_resource）接入，Factorial 连接器只需声明端点目录、分页器、认证方式和源级元数据，即可被编排框架调度。理解 Factorial 连接器，就等于理解 PostHog 对"带日期版本路径 + 游标分页 + 全量刷新"类 REST API 的标准接入姿势。

Connection：连接层细节

Host 与版本化路径

Factorial 使用单一全局 Host，不存在按账号区分的子域名：

Host：https://api.factorialhr.com
版本：以日期路径段形式携带，基础 URL 为https://api.factorialhr.com/api/<version>
支持的版本标签：2025-04-01、2026-04-01、2026-07-01（默认）

源码中这些常量定义在 factorial.py：

FACTORIAL_HOST = "https://api.factorialhr.com" API_VERSION_2025_04_01 = "2025-04-01" API_VERSION_2026_04_01 = "2026-04-01" API_VERSION_2026_07_01 = "2026-07-01" def base_url(api_version: str) -> str: return f"{FACTORIAL_HOST}/api/{api_version}"

一个关键设计是：资源偶尔会在不同版本间移动分组，但资源路径与响应信封不变。即/resources/<group>/<resource>路径结构和{"meta": ..., "data": [...]}信封在三个版本上完全一致，新版本只是增删响应字段——这些字段差异会被 PostHog 的自动推断 schema（auto-inferred schema）吸收为列的变化，无需代码分支。

标识符序列化（Identifier serialization）

这是本连接器最值得注意的版本差异点：2026-07-01（代号 "Bessel"）将每个资源的 id 序列化为不透明字符串而非整数，因为 id 已超出安全 64 位整数范围。该变化同时影响请求参数、响应和 webhook。

代码采用"无版本分支"的兼容策略（factorial.py 注释与 source.py 一致）：

主键仍是id列：该列类型不做硬编码，交给自动推断（type-agnostic, auto-inferred）；
分页转发不透明游标meta.end_cursor：绝不使用原始记录 id 作为分页参数，因此字符串 id 对分页逻辑无影响。

钉在2025-04-01/2026-04-01的源仍然得到整数 id。

版本生命周期与"静默漂移"陷阱

Factorial 每季度发布一个版本，每个版本服务一年。关键行为：对已退役版本的请求不会被拒绝，而是"使用最旧版本 schema"提供服务——这意味着过期的版本钉（pin）会静默漂移，而不是报错失败。

由此产生的工程结论（api_inventory.md 原文要点）：

2025-04-01已于 2026-04-01 过期，在 source.py 中被标记为 deprecated；
迁移0164_repin_factorial_api_version将源级钉批量重钉到2026-07-01；
由于厂商是"向前回退"而不是"报错"，不存在"版本被拒"状态需要加入get_non_retryable_errors。

源码元数据印证（source.py）：

class FactorialSource(ResumableSource[FactorialSourceConfig, FactorialResumeConfig]): lists_tables_without_credentials = True # static endpoint catalog — safe for public docs supported_versions = (API_VERSION_2025_04_01, API_VERSION_2026_04_01, API_VERSION_2026_07_01) default_version = API_VERSION_2026_07_01 api_docs_url = "https://apidoc.factorialhr.com/docs/api-versioning" deprecated_versions = (VersionDeprecation(version=API_VERSION_2025_04_01, sunset_at=date(2026, 4, 1)),)

测试 test_factorial_source.py 覆盖了版本弃用语义：2025-04-01的 sunset 日期为 2026-04-01，而2026-04-01、2026-07-01及未钉版本均不携带弃用警告。

重钉迁移`0164`：幂等、可逆为空操作

0164_repin_factorial_api_version.py 的实现要点：

仅更新source_type="Factorial"且api_version="2025-04-01"的ExternalDataSource行，将其重钉为2026-07-01；
不动 schema 级ExternalDataSchema.api_version覆盖——那是用户刻意手动钉的，交给弃用警告提示用户自行迁移；
NULL 钉已默认解析为default_version，无需更新；2026-04-01仍在服务期内，保持不动；
只匹配2025-04-01保证幂等：重复执行不匹配任何行；
reverse 为noop：重钉后的行与原生2026-07-01行无法区分，回滚会误伤合法钉；
迁移elidable=True，且无数据/ schema 变换：路径、分页游标、响应信封不变，主键仍为自动推断的id列，所有 schema 都是全量刷新，下次调度直接重建表即可。

认证方式

认证：x-api-key: <key>请求头（API key 方式）。
Factorial 官方也支持 OAuth2，但本连接器未实现——API key 认证对公司/内部集成已完全够用，且授予整个账号的访问权限。
源码实现细节（factorial.py）：通过APIKeyAuth（而非裸 header）注册 key，从而让 key 参与基于值的日志脱敏（log redaction）；同时make_tracked_session(redact_values=(api_key,))进一步脱敏。

"client": { "base_url": base_url(api_version), "auth": { "type": "api_key", "api_key": api_key, "name": "x-api-key", "location": "header", }, "paginator": FactorialCursorPaginator(), "session": make_tracked_session(redact_values=(api_key,)), },

资源路径形态

统一为/resources/<group>/<resource>，例如/resources/employees/employees。端点目录见后文表格。

Pagination：基于记录 id 的游标分页

协议侧约定

游标分页基于记录 id；
参数：limit（默认与上限均为 100）、after_id（向前）、before_id（向后）；
响应信封：{"meta": {...}, "data": [...]}；
meta携带has_next_page、has_previous_page、start_cursor、end_cursor、total、limit；
前向翻页：将after_id = meta.end_cursor传入，直到has_next_page为 false；记录按 id 升序返回；
没有文档化的sort/order参数，排序隐含在 id 游标遍历中。

实现：`FactorialCursorPaginator`

factorial.py 中的FactorialCursorPaginator完整实现了上述协议，核心行为：

init_request：首次请求只注入limit=PAGE_SIZE（PAGE_SIZE = 100），无after_id；
update_state：从响应 JSON 提取meta，仅当has_next_page为真且存在end_cursor且本页有数据时，才记录_after_id = str(end_cursor)并置_has_next_page = True；
空页兜底：即使 API 误报has_next_page=true，只要本页data为空就停止，避免死循环；
update_request：后续请求带上after_id；
恢复支持：get_resume_state/set_resume_state以{"after_id": ...}序列化游标，配合ResumableSourceManager实现断点续传。

测试 test_factorial.py 逐条验证了这些行为：初始状态、fresh 请求只带 limit、非终止页推进 after_id、has_next_page=false停止、缺end_cursor停止、空页即使 API 声称还有下一页也停止、恢复状态往返。

断点续传（Resumable）

factorial_source（factorial.py）与ResumableSourceManager[FactorialResumeConfig]协作：

FactorialResumeConfig只持有一个字段after_id（不透明前向游标，factorial.py）；
可恢复时，将保存的after_id作为initial_paginator_state注入分页器；
save_checkpoint在每页产出后保存，且仅当存在下一页才持久化（Redis TTL 负责清理）；崩溃后重新抓取最后一页而非跳过，合并阶段按主键去重（factorial.py）。

端到端恢复行为测试（test_factorial.py）：fresh 运行每页保存游标、恢复时以保存的游标播种、所有请求都走钉定版本的路径、单页终止不保存状态、不可恢复时不加载状态。

Incremental sync：增量同步决策

这是本连接器策略性最强的部分。核心事实：

Factorial服务端updated_after过滤只对少数资源有文档：project_management/flexible_time_records和project_management/subprojects；
对更高价值的 people / time-off / attendance 流没有文档化该参数（Airbyte 连接器印证了这一点：除shifts外，它都在客户端过滤updated_at）；
按 warehouse-sources 接入规范：一个"仍然遍历每一页的客户端游标"不算增量；
两个updated_after端点在没有真实 API key 的情况下无法用 curl 验证；
结论：当前所有端点一律全量刷新（INCREMENTAL_FIELDS = {}）。

源码依据（settings.py）：

# Full refresh only. Factorial documents a server-side `updated_after` filter on only two of the # endpoints we sync — `project_management/flexible_time_records` and `project_management/subprojects` # — and not on the higher-value people/time-off/attendance streams ... INCREMENTAL_FIELDS: dict[str, list[IncrementalField]] = {}

schema 层随之声明supports_incremental=False、supports_append=False、incremental_fields=[]（source.py），测试也断言"所有 schema 均仅全量刷新"（test_factorial_source.py）。

升级路径：一旦用带未来日期截止（future-date cutoff）的真实账号 curl 验证updated_after确实能收窄结果，就应将flexible_time_records/subprojects提升为增量。

未来方向：Factorial 还提供employee_updates/*变更流资源和 webhooks（api_public/webhook_subscriptions），是未来 webhook 驱动迭代的候选（api_inventory.md 原文注明）。

Synced endpoints：同步端点清单

settings.py中的FACTORIAL_ENDPOINTS（settings.py）定义了 17 个端点，覆盖：人员与组织架构、合同、休假、考勤、报销、薪酬、项目工时、招聘（ATS）。每个列表资源的主键都是id列；分区键只在created_at可靠出现在每一行时设置（事务型记录），查找/配置类资源不分区。

Table	Path	Partition key
employees	`/resources/employees/employees`	created_at
teams	`/resources/teams/teams`	—
team_memberships	`/resources/teams/memberships`	—
locations	`/resources/locations/locations`	—
legal_entities	`/resources/companies/legal_entities`	—
contract_versions	`/resources/contracts/contract_versions`	created_at
leaves	`/resources/timeoff/leaves`	created_at
leave_types	`/resources/timeoff/leave_types`	—
allowances	`/resources/timeoff/allowances`	—
attendance_shifts	`/resources/attendance/shifts`	created_at
expenses	`/resources/expenses/expenses`	created_at
payroll_supplements	`/resources/payroll/supplements`	created_at
flexible_time_records	`/resources/project_management/flexible_time_records`	created_at
projects	`/resources/project_management/projects`	—
candidates	`/resources/ats/candidates`	created_at
job_postings	`/resources/ats/job_postings`	—
applications	`/resources/ats/applications`	created_at

主键与分区的实现细节

FactorialEndpointConfig（settings.py）默认primary_keys=["id"]、should_sync_default=True；partition_key为可空字段；
id在2025-04-01/2026-04-01上序列化为整数，在2026-07-01上为不透明字符串，因此列类型交给推断（见上文标识符序列化）；
get_resource（factorial.py）统一声明data_selector="data"、write_disposition="replace"、table_format="delta"——每个列表端点都把记录包在顶层data键下；
SourceResponse（factorial.py）按分区键设置partition_mode="datetime"、partition_format="week"、sort_mode="asc"（游标按 id 升序，页面顺序稳定向前）；
测试 test_factorial.py 参数化验证：有分区键的端点走 datetime 周分区，无分区键的端点完全跳过分区，所有端点主键均为["id"]。

端点/列描述：`canonical_descriptions.py`

canonical_descriptions.py 为每个端点提供文档来源的描述与关键列注释（如employees的manager_id、terminated_on，contract_versions的salary_amount（单位分）、leaves的approved、applications的candidate_id/ats_job_posting_id等）。未覆盖的列会回退到 LLM 增强，因此部分覆盖是允许的。测试保证描述表的键都是真实端点，避免死数据（test_factorial_source.py）。

凭据校验（Credentials validation）

创建源时需要对 API key 做一次真实探测（factorial.py）：

探测核心employees端点：每个 HRIS 账号都有它，无权限的 key 会在这里 401/403；
请求参数{"limit": 1}，携带x-api-key头，超时 10 秒；
allow_redirects=False作为纵深防御：即使 base URL 是硬编码的，也防止 API key 被转发到重定向后的其他主机；
状态码映射：200 → 有效；401/403 → "Invalid Factorial API key, or it does not have access to your account's data."；其他 → 返回实际状态码；
由于 Factorial API key 拥有整个账号访问权，无需像 OAuth 源那样在创建时做 scope 级校验。

对应测试（test_factorial.py）：200/401/403/500 的状态码映射、探测请求精确断言（URL、headers、params、allow_redirects）、网络异常返回错误消息。

运行时对错误的处理（source.py）：get_non_retryable_errors将401 Client Error、403 Client Error、Unauthorized for url映射为可操作的提示（"key 无效或已吊销，请在 Factorial 账号设置中新建 key 并重新连接"），因为这些凭据问题重试无意义。

Rate limits：限流

POST 在2025-*端点上文档化为200 req/min；
GET 的限流与限流响应头没有公开文档；
当前实现依赖 tracked session 的默认重试机制处理瞬时429/5xx。

Verification status：验证状态

官方口径必须如实呈现：

端点路径、分页行为、updated_after覆盖范围均与官方文档及 Airbyte/Fivetran 连接器流清单交叉核对过；
未对真实账号做 curl 验证（没有可用 API key）；
连接层（host、版本路径、x-api-key、错误 key 返回 401）已通过未认证 curl 得到 401确认——这验证了认证通道与 401 语义，但未能验证业务数据读取路径。

测试覆盖总览

test_factorial.py：分页器状态机（9 个用例）、资源 shape 参数化、分区/主键参数化、端到端断点续传、凭据校验；
test_factorial_source.py：无凭据列表明细（lists_tables_without_credentials=True，公开文档可渲染表格）、schema 全量刷新断言、版本解析与透传（None → 2026-07-01）、版本弃用元数据。

从源码结构可以推断，lists_tables_without_credentials=True意味着get_schemas是纯静态目录（无 I/O），这既让公开文档可以安全渲染"Supported tables"章节，也降低了公开页面的数据泄露风险。

小结：接入 Factorial 的关键决策清单

版本钉是核心状态：Factorial 静默前向回退而非报错，必须依赖迁移 +deprecated_versions元数据主动管理版本，不能等失败暴露；
id 可能是不透明字符串：主键留给推断、分页只用end_cursor，即可无分支兼容2026-07-01；
分页要防死循环：has_next_page可能误报，空页兜底是必须的；
没有验证过的增量就不做增量：客户端过滤不算增量，updated_after未验证前一律全量刷新 +replace写盘；
凭据探测要防御重定向：allow_redirects=False防止 key 外泄；
续传粒度是"每页"：每页保存游标 + 合并去重，崩溃只重拉最后一页。

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考

PostHog 数据仓库 Factorial（HRIS）连接器 API 对接全解析：连接、分页、版本策略与增量同步决策