Linear

pip install httpx geniffy

uv add httpx geniffy

import httpx
from geniffy import Geniffy, NotFoundError

geniffy = Geniffy()                               # reads GENIFFY_API_KEY
LABELS = {"channel": "linear"}
ISSUES = """
query Issues($filter: IssueFilter, $after: String) {
  issues(first: 50, after: $after, filter: $filter, orderBy: updatedAt, includeArchived: true) {
    nodes {
      id identifier title description updatedAt trashed
      state { name }
      comments(first: 100) { nodes { body createdAt user { name } } }
    }
    pageInfo { hasNextPage endCursor }
  }
}
"""


def linear(api_key: str) -> httpx.Client:
    return httpx.Client(base_url="https://api.linear.app", timeout=30, headers={"Authorization": api_key})


def space_for(team: str) -> str:
    return "linear_" + team.lower()


def thread(issue: dict) -> str:
    """An issue as one note: its state and description, then each person's comment with who wrote it and when."""
    parts = [f"{issue['identifier']}, {issue['state']['name']}.", issue.get("description") or issue["title"]]
    for comment in issue["comments"]["nodes"]:
        if comment["user"] and comment["body"].strip():    # people, not bots or integrations
            parts.append(f"From {comment['user']['name']}, {comment['createdAt'][:10]}:\n{comment['body']}")
    return "\n\n".join(parts)


def sync(team: str, api_key: str, since: str | None = None) -> str | None:
    """Bring a Linear team's issues into its own space; team is its key, such as ENG. Returns the time to pass
    next time: pass None the first time and every issue is read; after that, only issues updated since."""
    mem, newest, after, seen = geniffy.space(space_for(team)), since, None, set()
    where = {"team": {"key": {"eq": team}}}
    if since:
        where["updatedAt"] = {"gt": since}
    with linear(api_key) as api:
        while True:
            out = api.post("/graphql", json={"query": ISSUES, "variables": {"filter": where, "after": after}})
            body = out.raise_for_status().json()
            if body.get("errors"):
                raise RuntimeError(body["errors"][0]["message"])
            page = body["data"]["issues"]
            for issue in page["nodes"]:
                newest, ref = max(newest or "", issue["updatedAt"]), f"linear:{issue['id']}"
                if issue["trashed"]:
                    try:
                        mem.sources.delete(external_id=ref)
                    except NotFoundError:
                        pass
                    continue
                mem.memories.add(thread(issue), title=f"{issue['identifier']} {issue['title']}",
                                 said_at=issue["updatedAt"], external_id=ref, labels=LABELS)
                seen.add(ref)
            if not page["pageInfo"]["hasNextPage"]:
                break
            after = page["pageInfo"]["endCursor"]
    if since is None:                             # everything was read: what is no longer there goes
        mem.sources.delete_labelled(LABELS, keep=seen)
    return newest

known = geniffy.space(space_for("ENG")).context("Why did we move off Redis?")