taolib.symphony.tracker.linear

taolib.symphony.tracker.linear#

Linear 子包 — Linear API 客户端实现。

Submodules#

Classes#

LinearClient

Linear API 客户端，基于 gql + httpx 异步传输。

Package Contents#

class taolib.symphony.tracker.linear.LinearClient(config: taolib.symphony.tracker.models.TrackerConfig)#

Bases: taolib.symphony.tracker.base.TrackerClient

Linear API 客户端，基于 gql + httpx 异步传输。

参数:

config -- Tracker 配置，包含 endpoint、api_key、project_slug 等。

_client#
_config#
async fetch_candidate_issues() → list[taolib.symphony.tracker.models.Issue]#

获取候选 issue 列表（分页）。

按 project slug + 配置中的 active_states 过滤， 每页取 _PAGE_SIZE 条，自动翻页直到遍历完成。

async fetch_issues_by_states(states: list[str]) → list[taolib.symphony.tracker.models.Issue]#

按指定状态集合过滤 issue（分页）。

主要用于终态清理：启动时扫描 Done / Cancelled 等终态 的工作区残留。

参数:

states -- 状态名称列表。

async fetch_issue_states_by_ids(issue_ids: list[str]) → list[taolib.symphony.tracker.models.Issue]#

根据 ID 列表批量获取 issue 的当前状态（分批）。

将 ID 列表按 _PAGE_SIZE 分批查询，每批独立发起请求， 最终按原始请求顺序排列返回结果。

参数:

issue_ids -- Issue UUID 列表。

async _paginated_query(query: object, variables: dict, field_name: str) → list[dict]#

通用分页查询逻辑。

自动跟踪 endCursor，循环翻页直到 hasNextPage 为 False。

参数:

  • query -- gql 编译后的 AST 查询文档。

  • variables -- GraphQL 变量字典。

  • field_name -- 响应 data 中对应的字段名（如 "issues"）。

返回:

所有页的 node 字典合并列表。

抛出:

LinearMissingCursorError -- hasNextPage=True 但 endCursor 缺失。

async _execute_query(query: object, variables: dict, field_name: str, *, paginated: bool) → tuple[list[dict], dict]#

执行单次 GraphQL 查询并提取结果。

参数:

  • query -- gql AST 查询文档。

  • variables -- GraphQL 变量。

  • field_name -- 响应 data 中的字段名。

  • paginated -- 是否为分页查询（需要提取 pageInfo）。

返回:

(nodes, page_info) 元组。 非分页查询的 page_info 为空字典。

抛出: