unasync.py

#!venv/bin/python
import os
import re
import shutil
import sys
from pathlib import Path

SUBS = [
    ("import asyncio", "import time"),
    ("await asyncio.sleep", "time.sleep"),
    ("async def", "def"),
    ("async with", "with"),
    ("await ", ""),
    ("async for", "for"),
    ("__aiter__", "__iter__"),
    ("AsyncIterator", "Iterator"),
    ("AsyncCacheTransport", "CacheTransport"),
    ("AsyncBaseTransport", "BaseTransport"),
    ("AsyncCacheClient", "CacheClient"),
    ("AsyncClient", "Client"),
    ("AsyncIterable", "Iterable"),
    ("AsyncCacheStream", "CacheStream"),
    ("AsyncByteStream", "ByteStream"),
    ("AsyncHTTPTransport", "HTTPTransport"),
    ("AsyncFileStorage", "FileStorage"),
    ("AsyncCacheConnectionPool", "CacheConnectionPool"),
    ("handle_async_request", "handle_request"),
    ("aread", "read"),
    ("aclose", "close"),
    ("asleep", "sleep"),
    ("AsyncLock", "Lock"),
    (
        "from httpcore._async.interfaces import AsyncRequestInterface",
        "from httpcore._sync.interfaces import RequestInterface",
    ),
    ("from hishel._async._transports", "from hishel._sync._transports"),
    ("AsyncRequestInterface", "RequestInterface"),
    ("__aenter__", "__enter__"),
    ("__aexit__", "__exit__"),
    ("*@pytest.mark.anyio", ""),
    ("*@pytest.mark.asyncio", ""),
    ("*@async_proxy.*", ""),
    (r'*@pytest.mark.parametrize\("anyio_backend", \["asyncio"\]\)', r""),
    ("anysqlite", "sqlite3"),
    ("AsyncManager", "Manager"),
    ("from .async_api ", "from .sync_api "),
    (".aiter_bytes", ".iter_bytes"),
    ("_async", "_sync"),
    ("from async_property import async_property", ""),
    ("*@async_property", "@property"),
    (r"asyncio.run\((.*)\)", r"\2"),
]
COMPILED_SUBS = [(re.compile(r"(^|\b)" + regex + r"($|\b|(?=\())"), repl) for regex, repl in SUBS]

USED_SUBS = set()

IGNORE_DIRS = [
    "__pycache__",
]


def unasync_line(line):
    for index, (regex, repl) in enumerate(COMPILED_SUBS):
        old_line = line
        line = re.sub(regex, repl, line)
        if index not in USED_SUBS:
            if line != old_line:
                USED_SUBS.add(index)
    return line


def unasync_file(in_path, out_path):
    in_path = Path(in_path)
    out_path = Path(out_path)
    out_path.parent.mkdir(parents=True, exist_ok=True)
    with open(in_path, "r") as in_file:
        with open(out_path, "w", newline="") as out_file:
            out_file.write("# " + "*" * 80 + "\n")
            out_file.write(
                "# *"
                + "WARNING: This file is automatically generated by unasync.py.".center(78)
                + "*\n"
            )
            out_file.write("# *" + "DO NOT MANUALLY EDIT".center(78) + "*\n")
            out_file.write(
                "# *"
                + f"Source File: {in_path.relative_to(Path(__file__).parent)}".center(78)
                + "*\n"
            )
            out_file.write("# " + "*" * 80 + "\n\n")
            last_line = None
            for line in in_file.readlines():
                line = unasync_line(line)
                if last_line and last_line.strip().startswith("@") and not line.strip():
                    pass
                else:
                    out_file.write(line)
                last_line = line


def unasync_file_check(in_path, out_path):
    with open(in_path, "r") as in_file:
        with open(out_path, "r") as out_file:
            for in_line, out_line in zip(in_file.readlines(), out_file.readlines()):
                expected = unasync_line(in_line)
                if out_line != expected:
                    print(f"unasync mismatch between {in_path!r} and {out_path!r}")
                    print(f"Async code:         {in_line!r}")
                    print(f"Expected sync code: {expected!r}")
                    print(f"Actual sync code:   {out_line!r}")
                    sys.exit(1)


def unasync_dir(in_dir, out_dir, check_only=False):
    for dirpath, dirnames, filenames in os.walk(in_dir):
        for filename in filenames:
            if not filename.endswith(".py") and any(f in dirpath for f in IGNORE_DIRS):
                continue
            rel_dir = os.path.relpath(dirpath, in_dir)
            in_path = os.path.normpath(os.path.join(in_dir, rel_dir, filename))
            out_path = os.path.normpath(os.path.join(out_dir, rel_dir, filename))
            # print(in_path, "->", out_path)
            if check_only:
                if not filename.endswith(".py"):
                    continue
                unasync_file_check(in_path, out_path)
            else:
                if (
                    Path(out_path).exists()
                    and Path(out_path).stat().st_mtime > Path(in_path).stat().st_mtime
                ):
                    continue
                if not filename.endswith(".py"):
                    Path(out_path).parent.mkdir(parents=True, exist_ok=True)
                    shutil.copy(in_path, out_path)
                else:
                    unasync_file(in_path, out_path)


def main():
    check_only = "--check" in sys.argv
    async_dirs = Path(__file__).parent.glob("patent_client/**/_async")

    for dir in async_dirs:
        sync_dir = dir.with_name("_sync")
        sync_dir.mkdir(parents=True, exist_ok=True)
        unasync_dir(dir, sync_dir, check_only)

    # if len(USED_SUBS) != len(SUBS):
    #    unused_subs = [SUBS[i] for i in range(len(SUBS)) if i not in USED_SUBS]

    # print("This SUBS was not used")
    # pprint(unused_subs)
    # exit(1)


if __name__ == "__main__":
    main()