Share JSON cache and Linear user helpers

Amp-Thread-ID: https://ampcode.com/threads/T-019e497d-a5a0-77b0-a8bb-fa8abc74113a
Co-authored-by: Amp <amp@ampcode.com>

Author: marcleblanc2

src/src_py_lib/__init__.py

@@ -7,7 +7,12 @@
 from pathlib import Path
 
 from src_py_lib.clients.github import GitHubClient, PullRequest, gh_cli_token, pr_ref_from_url
-from src_py_lib.clients.google_sheets import GoogleSheetsClient, GoogleSheetsError
+from src_py_lib.clients.google_sheets import (
+    GoogleSheetsClient,
+    GoogleSheetsError,
+    gcloud_adc_access_token,
+    quota_project_from_adc,
+)
 from src_py_lib.clients.graphql import (
     GraphQLClient,
     GraphQLError,
@@ -36,6 +41,7 @@
     config_parse_args as parse_args,
 )
 from src_py_lib.utils.http import HTTPClient, HTTPClientError
+from src_py_lib.utils.json_cache import load_json_cache, load_json_subset, save_json_cache
 from src_py_lib.utils.json_types import (
     JSONDict,
     json_dict,
@@ -112,6 +118,7 @@ def _script_name() -> str:
     "error",
     "event",
     "gh_cli_token",
+    "gcloud_adc_access_token",
     "info",
     "introspect_schema",
     "json_dict",
@@ -121,11 +128,15 @@ def _script_name() -> str:
     "json_str",
     "json_strs",
     "linear_client_from_config",
+    "load_json_cache",
+    "load_json_subset",
     "logging",
     "log",
     "log_context",
     "parse_args",
     "pr_ref_from_url",
+    "quota_project_from_adc",
+    "save_json_cache",
     "slack_client_from_config",
     "startup_event",
     "warning",

src/src_py_lib/clients/linear.py

@@ -8,7 +8,7 @@
 from src_py_lib.clients.graphql import GraphQLClient
 from src_py_lib.utils.config import Config, config_field
 from src_py_lib.utils.http import HTTPClient
-from src_py_lib.utils.json_types import JSONDict, JSONValue, json_dict
+from src_py_lib.utils.json_types import JSONDict, JSONValue, json_dict, json_dicts
 
 LINEAR_API_URL = "https://api.linear.app/graphql"
 LINEAR_VALIDATE_QUERY = """
@@ -18,6 +18,31 @@
   }
 }
 """
+LINEAR_USERS_QUERY = """
+query LinearUsers($first: Int!, $after: String) {
+  users(first: $first, after: $after, includeArchived: true) {
+    nodes {
+      id
+      name
+      displayName
+      email
+      teamMemberships(first: 25) {
+        nodes {
+          team {
+            id
+            key
+            name
+          }
+        }
+      }
+    }
+    pageInfo {
+      hasNextPage
+      endCursor
+    }
+  }
+}
+"""
 
 
 class LinearClientConfig(Config):
@@ -61,6 +86,11 @@ def validate(self) -> JSONDict:
             raise RuntimeError("Linear viewer response did not include viewer.email.")
         return viewer
 
+    def list_users(self, *, page_size: int = 100) -> list[JSONDict]:
+        """Return every Linear user with common people-directory fields."""
+        data = self.graphql(LINEAR_USERS_QUERY, page_size=page_size)
+        return json_dicts(json_dict(data.get("users")).get("nodes"))
+
 
 def linear_client_from_config(
     config: LinearClientConfig, *, http: HTTPClient | None = None

src/src_py_lib/utils/json_cache.py

@@ -0,0 +1,42 @@
+"""Small on-disk JSON cache helpers."""
+
+from __future__ import annotations
+
+import json
+from collections.abc import Callable, Mapping
+from pathlib import Path
+from typing import cast
+
+from src_py_lib.utils.json_types import JSONDict
+
+
+def load_json_cache[Entry](
+    path: Path,
+    parse: Callable[[JSONDict], Entry] | None = None,
+) -> dict[str, Entry]:
+    """Load `path` as a string-keyed cache. Missing files return `{}`."""
+    if not path.exists():
+        return {}
+    raw = cast(dict[str, JSONDict], json.loads(path.read_text(encoding="utf-8")))
+    if parse is None:
+        return cast(dict[str, Entry], raw)
+    return {key: parse(value) for key, value in raw.items()}
+
+
+def save_json_cache(path: Path, cache: Mapping[str, object]) -> None:
+    """Write a string-keyed JSON cache with stable formatting."""
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(dict(cache), indent=2, sort_keys=True) + "\n", encoding="utf-8")
+
+
+def load_json_subset[Entry](
+    path: Path,
+    keys: list[str],
+    parse: Callable[[JSONDict], Entry] | None = None,
+) -> dict[str, Entry]:
+    """Load only `keys` that are present in a string-keyed JSON cache."""
+    cache = load_json_cache(path, parse=parse)
+    return {key: cache[key] for key in keys if key in cache}
+
+
+__all__ = ["load_json_cache", "load_json_subset", "save_json_cache"]

tests/test_import.py

@@ -25,13 +25,17 @@ def test_root_public_api_exports_common_entrypoints(self) -> None:
         self.assertIsNotNone(src_py_lib.SlackPacer)
         self.assertIsNotNone(src_py_lib.config_field)
         self.assertIsNotNone(src_py_lib.gh_cli_token)
+        self.assertIsNotNone(src_py_lib.gcloud_adc_access_token)
         self.assertIsNotNone(src_py_lib.info)
         self.assertIsNotNone(src_py_lib.json_dicts)
         self.assertIsNotNone(src_py_lib.json_str)
         self.assertIsNotNone(src_py_lib.log)
         self.assertIsNotNone(src_py_lib.logging)
         self.assertIsNotNone(src_py_lib.linear_client_from_config)
+        self.assertIsNotNone(src_py_lib.load_json_cache)
         self.assertIsNotNone(src_py_lib.parse_args)
+        self.assertIsNotNone(src_py_lib.quota_project_from_adc)
+        self.assertIsNotNone(src_py_lib.save_json_cache)
         self.assertIsNotNone(src_py_lib.slack_client_from_config)
         self.assertIsNotNone(src_py_lib.write_tsv)
 

tests/test_logging_http_clients.py

@@ -1283,6 +1283,59 @@ def test_linear_client_validate_requires_viewer_email(self) -> None:
         with self.assertRaisesRegex(RuntimeError, "viewer.email"):
             client.validate()
 
+    def test_linear_client_list_users_paginates(self) -> None:
+        http = RecordingHTTP(
+            [
+                {
+                    "data": {
+                        "users": {
+                            "nodes": [{"id": "U1", "name": "Alice"}],
+                            "pageInfo": {
+                                "hasNextPage": True,
+                                "endCursor": "cursor-1",
+                            },
+                        }
+                    }
+                },
+                {
+                    "data": {
+                        "users": {
+                            "nodes": [{"id": "U2", "name": "Bob"}],
+                            "pageInfo": {
+                                "hasNextPage": False,
+                                "endCursor": None,
+                            },
+                        }
+                    }
+                },
+            ]
+        )
+
+        users = LinearClient("token", http=http).list_users(page_size=25)
+
+        self.assertEqual([user["id"] for user in users], ["U1", "U2"])
+        first_body = json_dict(http.calls[0]["json_body"])
+        second_body = json_dict(http.calls[1]["json_body"])
+        self.assertEqual(
+            json_dict(first_body.get("variables")),
+            {"first": 25, "after": None},
+        )
+        self.assertEqual(
+            json_dict(second_body.get("variables")),
+            {"first": 25, "after": "cursor-1"},
+        )
+
+    def test_json_cache_helpers_round_trip_and_parse(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            path = Path(tmp) / "nested" / "cache.json"
+
+            src.save_json_cache(path, {"b": {"name": "Bob"}, "a": {"name": "Alice"}})
+            parsed = src.load_json_cache(path, parse=lambda value: str(value.get("name", "")))
+            subset = src.load_json_subset(path, ["a", "missing"], parse=lambda value: value)
+
+        self.assertEqual(parsed, {"a": "Alice", "b": "Bob"})
+        self.assertEqual(subset, {"a": {"name": "Alice"}})
+
     def test_resolve_op_secret_ref_leaves_raw_values_alone(self) -> None:
         self.assertEqual(resolve_op_secret_ref(" raw-secret "), "raw-secret")