8dfc4c1
#!/usr/bin/env python3
"""Regenerate the query/body flags in view/app/src/services/api/ops.ts.
The frontend calls the generated client through the `Apis` proxy, which has to
know where the query object and the request body sit in each generated
signature (`(pathParams..., params?, data?, options?)`). The mapping is derived
from the generated sources in packages/apiclient so it never drifts again:
query -> a parameter literally named `params`
body -> one parameter past the path params and the query object
path -> parameters before `params` in the matching `get<Name>Url`
"""
import glob
import json
import re
import sys
CLIENT = 'packages/apiclient'
OPS = 'view/app/src/services/api/ops.ts'
def paren_block(src, start):
i = src.index('(', start)
depth = 0
for j in range(i, len(src)):
if src[j] == '(':
depth += 1
elif src[j] == ')':
depth -= 1
if depth == 0:
return src[i + 1:j]
raise ValueError('unbalanced parenthesis')
def split_top(block):
out, depth, cur = [], 0, ''
for ch in block:
if ch in '([{<':
depth += 1
elif ch in ')]}>':
depth -= 1
if ch == ',' and depth == 0:
out.append(cur)
cur = ''
else:
cur += ch
if cur.strip():
out.append(cur)
return [n.strip() for n in out if n.strip()]
def param_name(arg):
m = re.match(r'(\w+)\??\s*:', arg)
return m.group(1) if m else None
def scan_signatures():
info = {}
for path in glob.glob(f'{CLIENT}/*/*.ts'):
if path.endswith('.schemas.ts') or path.endswith('index.ts'):
continue
src = open(path, encoding='utf-8').read()
m = re.search(r'export const (\w+) = async \(', src)
if not m:
continue
args = [a for a in split_top(paren_block(src, m.start())) if param_name(a) != 'options']
url = re.search(r'export const (get\w+Url) = \(', src)
url_args = [a for a in split_top(paren_block(src, url.start())) if param_name(a) != 'options'] if url else []
has_query = any(param_name(a) == 'params' for a in args) or any(param_name(a) == 'params' for a in url_args)
path_count = len([a for a in url_args if param_name(a) != 'params'])
rest = args[path_count:]
body_count = len(rest) - (1 if has_query else 0)
if body_count not in (0, 1):
raise SystemExit(f'unexpected signature in {path}: {args} / {url_args}')
body_first = has_query and body_count == 1 and param_name(rest[0]) != 'params'
info[m.group(1)] = {
'path': path_count,
'query': has_query,
'body': body_count == 1,
'bodyFirst': body_first,
}
return info
def main():
info = scan_signatures()
src = open(OPS, encoding='utf-8').read()
chunks = re.split(r"(?m)^(?= ')", src)
patched, missing, mismatch = [], [], []
for chunk in chunks:
m = re.search(r"^ '([\w.\-]+)':\s*\{", chunk, re.M)
if not m:
patched.append(chunk)
continue
key = m.group(1)
fn = re.search(r'fn:\s*api\.(\w+)', chunk)
info_for = info.get(fn.group(1)) if fn else None
if not info_for:
missing.append(key)
patched.append(chunk)
continue
declared = re.search(r'path:\s*(\d+)', chunk)
if declared and int(declared.group(1)) != info_for['path']:
mismatch.append((key, declared.group(1), info_for['path']))
chunk = re.sub(r'path:\s*\d+', f"path: {info_for['path']}", chunk, count=1)
chunk = re.sub(r'query:\s*\w+', f"query: {'true' if info_for['query'] else 'false'}", chunk, count=1)
chunk = re.sub(r'body:\s*\w+', f"body: {'true' if info_for['body'] else 'false'}", chunk, count=1)
chunk = re.sub(r',?\s*bodyFirst:\s*\w+,?', '', chunk)
if info_for['bodyFirst']:
end = chunk.rfind('}')
head = chunk[:end].rstrip()
sep = ' ' if head.endswith(',') else ', '
chunk = f'{head}{sep}bodyFirst: true {chunk[end:]}'
patched.append(chunk)
if mismatch:
print(json.dumps(mismatch, indent=1))
raise SystemExit('path arity disagreed with the generated client')
open(OPS, 'w', encoding='utf-8').write(''.join(patched))
stats = {'query': sum(v['query'] for v in info.values()), 'body': sum(v['body'] for v in info.values())}
print(f'patched {len(info)} operations (query {stats["query"]}, body {stats["body"]}), missing {missing}')
if __name__ == '__main__':
sys.exit(main())