upstream
upstream/examples/try_fixes.py

try_fixes.py

The same, with the async client and every fork running at once.

Python · 32 linesused in Forking and snapshots
import asyncioimport osimport sys from upstream import AsyncUpstream TEST = "cd app && git apply fix.patch && uv run pytest -q"  async def attempt(sb, patch):    await sb.write_file("/workspace/app/fix.patch", patch)    r = await sb.exec(["sh", "-c", TEST])    return r.exit_code == 0  async def main(repo, patches):    async with AsyncUpstream(api_key=os.environ["UPSTREAM_API_KEY"]) as client:        base = await client.sandbox(template="python")        await base.exec(["sh", "-c", f"git clone {repo} app && cd app && uv sync"])         forks = [await base.fork() for _ in patches]        passed = await asyncio.gather(*(attempt(sb, p) for sb, p in zip(forks, patches)))         for sb, ok in zip(forks, passed):            print(sb.sandbox_id, "pass" if ok else "fail")            await sb.destroy()        await base.destroy()  if __name__ == "__main__":    repo, *patch_files = sys.argv[1:]    asyncio.run(main(repo, [open(p).read() for p in patch_files]))