SUBS = [
# httpx specific
- ('AsyncByteStream', 'SyncByteStream'),
- ('async_auth_flow', 'sync_auth_flow'),
- ('handle_async_request', 'handle_request'),
+ ("AsyncByteStream", "SyncByteStream"),
+ ("async_auth_flow", "sync_auth_flow"),
+ ("handle_async_request", "handle_request"),
# general
- ('AsyncIterator', 'Iterator'),
- ('from anyio import Lock', 'from threading import Lock'),
- ('Async([A-Z][A-Za-z0-9_]*)', r'\2'),
- ('async def', 'def'),
- ('async with', 'with'),
- ('async for', 'for'),
- ('await ', ''),
- ('aclose', 'close'),
- ('aread', 'read'),
- ('__aenter__', '__enter__'),
- ('__aexit__', '__exit__'),
- ('__aiter__', '__iter__'),
- ('@pytest.mark.anyio', ''),
+ ("AsyncIterator", "Iterator"),
+ ("from anyio import Lock", "from threading import Lock"),
+ ("Async([A-Z][A-Za-z0-9_]*)", r"\2"),
+ ("async def", "def"),
+ ("async with", "with"),
+ ("async for", "for"),
+ ("await ", ""),
+ ("aclose", "close"),
+ ("aread", "read"),
+ ("__aenter__", "__enter__"),
+ ("__aexit__", "__exit__"),
+ ("__aiter__", "__iter__"),
+ ("@pytest.mark.anyio", ""),
]
COMPILED_SUBS = [
- (re.compile(r'(^|\b)' + regex + r'($|\b)'), repl)
- for regex, repl in SUBS
+ (re.compile(r"(^|\b)" + regex + r"($|\b)"), repl) for regex, repl in SUBS
]
USED_SUBS = set()
+
def unasync_line(line):
for index, (regex, repl) in enumerate(COMPILED_SUBS):
old_line = line
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}')
+ 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'):
+ if not filename.endswith(".py"):
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)
+ print(in_path, "->", out_path)
if check_only:
unasync_file_check(in_path, out_path)
else:
def main():
- check_only = '--check' in sys.argv
+ check_only = "--check" in sys.argv
unasync_dir("tests/client/async", "tests/client/sync", check_only=check_only)
if len(USED_SUBS) != len(SUBS):
print("These patterns were not used:")
pprint(unused_subs)
- exit(1)
-
+ exit(1)
+
-if __name__ == '__main__':
+if __name__ == "__main__":
main()