1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
|
#!venv/bin/python
import os
import re
import sys
from pprint import pprint
SUBS = [
('from .._backends.auto import AutoBackend', 'from .._backends.sync import SyncBackend'),
('import trio as concurrency', 'from tests import concurrency'),
('AsyncIterator', 'Iterator'),
('Async([A-Z][A-Za-z0-9_]*)', r'\2'),
('async def', 'def'),
('async with', 'with'),
('async for', 'for'),
('await ', ''),
('handle_async_request', 'handle_request'),
('aclose', 'close'),
('aiter_stream', 'iter_stream'),
('aread', 'read'),
('asynccontextmanager', 'contextmanager'),
('__aenter__', '__enter__'),
('__aexit__', '__exit__'),
('__aiter__', '__iter__'),
('@pytest.mark.anyio', ''),
('@pytest.mark.trio', ''),
('AutoBackend', 'SyncBackend'),
]
COMPILED_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
line = re.sub(regex, repl, line)
if old_line != line:
USED_SUBS.add(index)
return line
def unasync_file(in_path, out_path):
with open(in_path, "r") as in_file:
with open(out_path, "w", newline="") as out_file:
for line in in_file.readlines():
line = unasync_line(line)
out_file.write(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'):
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:
unasync_file_check(in_path, out_path)
else:
unasync_file(in_path, out_path)
def main():
check_only = '--check' in sys.argv
unasync_dir("httpcore/_async", "httpcore/_sync", check_only=check_only)
unasync_dir("tests/_async", "tests/_sync", check_only=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("These patterns were not used:")
pprint(unused_subs)
exit(1)
if __name__ == '__main__':
main()
|
