Skip to content

[stubsabot] Refactor suggest_typeshed_* functions #16308

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
donbarbos wants to merge 2 commits into
base: main
Choose a base branch
from
Open

[stubsabot] Refactor suggest_typeshed_* functions #16308

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
c895565
[stubsabot] Refactor suggest_typeshed_* functions
donbarbos Aug 27, 2026
c82c651
rename update to action
donbarbos Aug 27, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
128 changes: 57 additions & 71 deletions scripts/stubsabot.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,14 @@
from packaging.specifiers import Specifier
from termcolor import colored

from ts_utils.metadata import ObsoleteMetadata, StubMetadata, read_metadata, update_metadata
from ts_utils.metadata import (
ObsoleteMetadata,
StubMetadata,
StubtestSettings,
read_metadata,
read_stubtest_settings,
update_metadata,
)
from ts_utils.paths import PYRIGHT_CONFIG, STUBS_PATH, distribution_path
from ts_utils.stubs import third_party_stubs

Expand Down Expand Up @@ -824,15 +831,13 @@ def normalize(name: str) -> str:
BRANCH_PREFIX = "stubsabot"


def get_update_pr_body(update: Update, metadata: Mapping[str, Any]) -> str:
def get_update_pr_body(update: Update, settings: StubtestSettings) -> str:
body = "\n".join(f"{k}: {v}" for k, v in update.links.items())

if update.diff_analysis is not None:
body += f"\n\n{update.diff_analysis}"

stubtest_settings: dict[str, Any] = metadata.get("tool", {}).get("stubtest", {})
stubtest_will_run = not stubtest_settings.get("skip", False)
if stubtest_will_run:
if not settings.skip:
body += textwrap.dedent("""

If stubtest fails for this PR:
Expand Down Expand Up @@ -866,66 +871,55 @@ def remove_stubs(distribution: str) -> None:
f.writelines(lines)


async def suggest_typeshed_update(update: Update, session: aiohttp.ClientSession, action_level: ActionLevel) -> None:
def get_commit_message(action: Update | Obsolete | Remove) -> tuple[str, str]:
"""Return git commit message title and body."""
match action:
case Update():
stubtest_settings = read_stubtest_settings(action.distribution)
return (
f"[stubsabot] Bump {action.distribution} to {action.new_version}",
get_update_pr_body(action, stubtest_settings),
)
case Obsolete():
return (
f"[stubsabot] Mark {action.distribution} as obsolete since {action.obsolete_since_version}",
"\n".join(f"{k}: {v}" for k, v in action.links.items()),
)
case Remove():
return (
f"[stubsabot] Remove {action.distribution} as {action.reason}",
"\n".join(f"{k}: {v}" for k, v in action.links.items()),
)


def run_action(action: Update | Obsolete | Remove) -> None:
match action:
case Update():
update_metadata(action.distribution, version=action.new_version)
case Obsolete():
obsolete_t = cast(dict[str, object], tomlkit.inline_table())
obsolete_t.update({"version": action.obsolete_since_version, "date": action.obsolete_since_date.date().isoformat()})
update_metadata(action.distribution, obsolete_since=obsolete_t)
case Remove():
remove_stubs(action.distribution)


async def process_typeshed_change(
action: Update | Obsolete | Remove, session: aiohttp.ClientSession, action_level: ActionLevel
) -> None:
if action_level <= ActionLevel.nothing:
return
title = f"[stubsabot] Bump {update.distribution} to {update.new_version}"
async with _repo_lock:
branch_name = f"{BRANCH_PREFIX}/{normalize(update.distribution)}"
subprocess.check_call(["git", "checkout", "-B", branch_name, "origin/main"])
meta = update_metadata(update.distribution, version=update.new_version)
body = get_update_pr_body(update, meta)
subprocess.check_call(["git", "commit", "--all", "-m", f"{title}\n\n{body}"])
if action_level <= ActionLevel.local:
return
if not latest_commit_is_different_to_last_commit_on_origin(branch_name):
print(f"No pushing to origin required: origin/{branch_name} exists and requires no changes!")
return
somewhat_safe_force_push(branch_name)
if action_level <= ActionLevel.fork:
return

await create_or_update_pull_request(title=title, body=body, branch_name=branch_name, session=session)


async def suggest_typeshed_obsolete(obsolete: Obsolete, session: aiohttp.ClientSession, action_level: ActionLevel) -> None:
if action_level <= ActionLevel.nothing:
return
title = f"[stubsabot] Mark {obsolete.distribution} as obsolete since {obsolete.obsolete_since_version}"
async with _repo_lock:
branch_name = f"{BRANCH_PREFIX}/{normalize(obsolete.distribution)}"
subprocess.check_call(["git", "checkout", "-B", branch_name, "origin/main"])
obsolete_t = cast(dict[str, object], tomlkit.inline_table())
obsolete_t.update({"version": obsolete.obsolete_since_version, "date": obsolete.obsolete_since_date.date().isoformat()})
update_metadata(obsolete.distribution, obsolete_since=obsolete_t)
body = "\n".join(f"{k}: {v}" for k, v in obsolete.links.items())
subprocess.check_call(["git", "commit", "--all", "-m", f"{title}\n\n{body}"])
if action_level <= ActionLevel.local:
return
if not latest_commit_is_different_to_last_commit_on_origin(branch_name):
print(f"No PR required: origin/{branch_name} exists and requires no changes!")
return
somewhat_safe_force_push(branch_name)
if action_level <= ActionLevel.fork:
return

await create_or_update_pull_request(title=title, body=body, branch_name=branch_name, session=session)


async def suggest_typeshed_remove(remove: Remove, session: aiohttp.ClientSession, action_level: ActionLevel) -> None:
if action_level <= ActionLevel.nothing:
return
title = f"[stubsabot] Remove {remove.distribution} as {remove.reason}"
async with _repo_lock:
branch_name = f"{BRANCH_PREFIX}/{normalize(remove.distribution)}"
branch_name = f"{BRANCH_PREFIX}/{normalize(action.distribution)}"
subprocess.check_call(["git", "checkout", "-B", branch_name, "origin/main"])
remove_stubs(remove.distribution)
body = "\n".join(f"{k}: {v}" for k, v in remove.links.items())
run_action(action)
title, body = get_commit_message(action)
subprocess.check_call(["git", "commit", "--all", "-m", f"{title}\n\n{body}"])
if action_level <= ActionLevel.local:
return
if not latest_commit_is_different_to_last_commit_on_origin(branch_name):
print(f"No pushing to origin required: origin/{branch_name} exists and requires no changes!")
print(f"No pushing to origin required: 'origin/{branch_name}' exists and requires no changes!")
return
somewhat_safe_force_push(branch_name)
if action_level <= ActionLevel.fork:
Expand Down Expand Up @@ -995,13 +989,13 @@ async def main() -> int:

action_count = 0
for task in asyncio.as_completed(tasks):
update = await task
print(f"{update.distribution}... ", end="")
print(update)
action = await task
print(f"{action.distribution}... ", end="")
print(action)

if isinstance(update, NoUpdate):
if isinstance(action, NoUpdate):
continue
if isinstance(update, Error):
if isinstance(action, Error):
error = True
continue

Expand All @@ -1011,16 +1005,8 @@ async def main() -> int:
action_count += 1

try:
if isinstance(update, Update):
await suggest_typeshed_update(update, session, action_level=args.action_level)
continue
if isinstance(update, Obsolete):
await suggest_typeshed_obsolete(update, session, action_level=args.action_level)
continue
# Redundant, but keeping for extra runtime validation
if isinstance(update, Remove): # pyright: ignore[reportUnnecessaryIsInstance]
await suggest_typeshed_remove(update, session, action_level=args.action_level)
continue
await process_typeshed_change(action, session, action_level=args.action_level)
continue
except RemoteConflictError as e:
print(colored(f"... but ran into {type(e).__qualname__}: {e}", "red"))
continue
Expand Down