Workflow to automatically sync typeshed (#13845)

Resolves #13812

Author: hauntsaninja

.github/workflows/sync_typeshed.yml

@@ -0,0 +1,31 @@
+name: Sync typeshed
+
+on:
+  workflow_dispatch:
+  schedule:
+    - cron: "0 0 1,15 * *"
+
+permissions:
+  contents: write
+  pull-requests: write
+
+jobs:
+  sync_typeshed:
+    name: Sync typeshed
+    if: github.repository == 'python/mypy'
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+        # TODO: use whatever solution ends up working for
+        # https://github.com/python/typeshed/issues/8434
+      - uses: actions/setup-python@v4
+        with:
+          python-version: "3.10"
+      - name: git config
+        run: |
+          git config --global user.name mypybot
+          git config --global user.email '<>'
+      - name: Sync typeshed
+        run: |
+          python -m pip install requests==2.28.1
+          GITHUB_TOKEN=${{ secrets.GITHUB_TOKEN }} python misc/sync-typeshed.py --make-pr

misc/sync-typeshed.py

@@ -10,16 +10,21 @@
 from __future__ import annotations
 
 import argparse
+import functools
 import os
+import re
 import shutil
 import subprocess
 import sys
 import tempfile
 import textwrap
+from collections.abc import Mapping
+
+import requests
 
 
 def check_state() -> None:
-    if not os.path.isfile("README.md"):
+    if not os.path.isfile("README.md") and not os.path.isdir("mypy"):
         sys.exit("error: The current working directory must be the mypy repository root")
     out = subprocess.check_output(["git", "status", "-s", os.path.join("mypy", "typeshed")])
     if out:
@@ -37,6 +42,7 @@ def update_typeshed(typeshed_dir: str, commit: str | None) -> str:
     if commit:
         subprocess.run(["git", "checkout", commit], check=True, cwd=typeshed_dir)
     commit = git_head_commit(typeshed_dir)
+
     stdlib_dir = os.path.join("mypy", "typeshed", "stdlib")
     # Remove existing stubs.
     shutil.rmtree(stdlib_dir)
@@ -60,6 +66,69 @@ def git_head_commit(repo: str) -> str:
     return commit.strip()
 
 
+@functools.cache
+def get_github_api_headers() -> Mapping[str, str]:
+    headers = {"Accept": "application/vnd.github.v3+json"}
+    secret = os.environ.get("GITHUB_TOKEN")
+    if secret is not None:
+        headers["Authorization"] = (
+            f"token {secret}" if secret.startswith("ghp") else f"Bearer {secret}"
+        )
+    return headers
+
+
+@functools.cache
+def get_origin_owner() -> str:
+    output = subprocess.check_output(["git", "remote", "get-url", "origin"], text=True).strip()
+    match = re.match(
+        r"(git@github.com:|https://github.com/)(?P<owner>[^/]+)/(?P<repo>[^/\s]+)", output
+    )
+    assert match is not None, f"Couldn't identify origin's owner: {output!r}"
+    assert (
+        match.group("repo").removesuffix(".git") == "mypy"
+    ), f'Unexpected repo: {match.group("repo")!r}'
+    return match.group("owner")
+
+
+def create_or_update_pull_request(*, title: str, body: str, branch_name: str) -> None:
+    fork_owner = get_origin_owner()
+
+    with requests.post(
+        "https://api.github.com/repos/python/mypy/pulls",
+        json={
+            "title": title,
+            "body": body,
+            "head": f"{fork_owner}:{branch_name}",
+            "base": "master",
+        },
+        headers=get_github_api_headers(),
+    ) as response:
+        resp_json = response.json()
+        if response.status_code == 422 and any(
+            "A pull request already exists" in e.get("message", "")
+            for e in resp_json.get("errors", [])
+        ):
+            # Find the existing PR
+            with requests.get(
+                "https://api.github.com/repos/python/mypy/pulls",
+                params={"state": "open", "head": f"{fork_owner}:{branch_name}", "base": "master"},
+                headers=get_github_api_headers(),
+            ) as response:
+                response.raise_for_status()
+                resp_json = response.json()
+                assert len(resp_json) >= 1
+                pr_number = resp_json[0]["number"]
+            # Update the PR's title and body
+            with requests.patch(
+                f"https://api.github.com/repos/python/mypy/pulls/{pr_number}",
+                json={"title": title, "body": body},
+                headers=get_github_api_headers(),
+            ) as response:
+                response.raise_for_status()
+            return
+        response.raise_for_status()
+
+
 def main() -> None:
     parser = argparse.ArgumentParser()
     parser.add_argument(
@@ -72,12 +141,21 @@ def main() -> None:
         default=None,
         help="Location of typeshed (default to a temporary repository clone)",
     )
+    parser.add_argument(
+        "--make-pr",
+        action="store_true",
+        help="Whether to make a PR with the changes (default to no)",
+    )
     args = parser.parse_args()
+
     check_state()
-    print("Update contents of mypy/typeshed from typeshed? [yN] ", end="")
-    answer = input()
-    if answer.lower() != "y":
-        sys.exit("Aborting")
+
+    if args.make_pr:
+        if os.environ.get("GITHUB_TOKEN") is None:
+            raise ValueError("GITHUB_TOKEN environment variable must be set")
+
+    branch_name = "mypybot/sync-typeshed"
+    subprocess.run(["git", "checkout", "-B", branch_name, "origin/master"], check=True)
 
     if not args.typeshed_dir:
         # Clone typeshed repo if no directory given.
@@ -95,19 +173,34 @@ def main() -> None:
 
     # Create a commit
     message = textwrap.dedent(
-        """\
+        f"""\
         Sync typeshed
 
         Source commit:
         https://github.com/python/typeshed/commit/{commit}
-        """.format(
-            commit=commit
-        )
+        """
     )
     subprocess.run(["git", "add", "--all", os.path.join("mypy", "typeshed")], check=True)
     subprocess.run(["git", "commit", "-m", message], check=True)
     print("Created typeshed sync commit.")
 
+    # Currently just LiteralString reverts
+    commits_to_cherry_pick = ["780534b13722b7b0422178c049a1cbbf4ea4255b"]
+    for commit in commits_to_cherry_pick:
+        subprocess.run(["git", "cherry-pick", commit], check=True)
+        print(f"Cherry-picked {commit}.")
+
+    if args.make_pr:
+        subprocess.run(["git", "push", "--force", "origin", branch_name], check=True)
+        print("Pushed commit.")
+
+        warning = "Note that you will need to close and re-open the PR in order to trigger CI."
+
+        create_or_update_pull_request(
+            title="Sync typeshed", body=message + "\n" + warning, branch_name=branch_name
+        )
+        print("Created PR.")
+
 
 if __name__ == "__main__":
     main()

tox.ini

@@ -29,7 +29,7 @@ commands =
 description = type check ourselves
 commands =
     python -m mypy --config-file mypy_self_check.ini -p mypy -p mypyc
-    python -m mypy --config-file mypy_self_check.ini misc --exclude misc/fix_annotate.py --exclude misc/async_matrix.py
+    python -m mypy --config-file mypy_self_check.ini misc --exclude misc/fix_annotate.py --exclude misc/async_matrix.py --exclude misc/sync-typeshed.py
 
 [testenv:docs]
 description = invoke sphinx-build to build the HTML docs