sdk/python/discovery2pydoc.py

   1 #!/usr/bin/env python3
   2 # Copyright (C) The Arvados Authors. All rights reserved.
   3 #
   4 # SPDX-License-Identifier: Apache-2.0
   5 """discovery2pydoc - Build skeleton Python from the Arvados discovery document
   6
   7 This tool reads the Arvados discovery document and writes a Python source file
   8 with classes and methods that correspond to the resources that
   9 google-api-python-client builds dynamically. This source does not include any
  10 implementation, but it does include real method signatures and documentation
  11 strings, so it's useful as documentation for tools that read Python source,
  12 including pydoc and pdoc.
  13
  14 If you run this tool with the path to a discovery document, it uses no
  15 dependencies outside the Python standard library. If it needs to read
  16 configuration to find the discovery document dynamically, it'll load the
  17 `arvados` module to do that.
  18 """
  19
  20 import argparse
  21 import inspect
  22 import json
  23 import keyword
  24 import operator
  25 import os
  26 import pathlib
  27 import re
  28 import sys
  29 import urllib.parse
  30 import urllib.request
  31
  32 from typing import (
  33     Any,
  34     Callable,
  35     Mapping,
  36     Optional,
  37     Sequence,
  38 )
  39
  40 LOWERCASE = operator.methodcaller('lower')
  41 NAME_KEY = operator.attrgetter('name')
  42 STDSTREAM_PATH = pathlib.Path('-')
  43 TITLECASE = operator.methodcaller('title')
  44
  45 _TYPE_MAP = {
  46     # Map the API's JavaScript-based type names to Python annotations.
  47     # Some of these may disappear after Arvados issue #19795 is fixed.
  48     'Array': 'list',
  49     'array': 'list',
  50     'boolean': 'bool',
  51     # datetime fields are strings in ISO 8601 format.
  52     'datetime': 'str',
  53     'Hash': 'dict[str, Any]',
  54     'integer': 'int',
  55     'object': 'dict[str, Any]',
  56     'string': 'str',
  57     'text': 'str',
  58 }
  59
  60 def get_type_annotation(name: str) -> str:
  61     return _TYPE_MAP.get(name, name)
  62
  63 def to_docstring(s: str, indent: int) -> str:
  64     prefix = ' ' * indent
  65     s = s.replace('"""', '""\"')
  66     s = re.sub(r'(\n+)', r'\1' + prefix, s)
  67     s = s.strip()
  68     if '\n' in s:
  69         return f'{prefix}"""{s}\n{prefix}"""'
  70     else:
  71         return f'{prefix}"""{s}"""'
  72
  73 def transform_name(s: str, sep: str, fix_part: Callable[[str], str]) -> str:
  74     return sep.join(fix_part(part) for part in s.split('_'))
  75
  76 def classify_name(s: str) -> str:
  77     return transform_name(s, '', TITLECASE)
  78
  79 def humanize_name(s: str) -> str:
  80     return transform_name(s, ' ', LOWERCASE)
  81
  82 class Parameter(inspect.Parameter):
  83     def __init__(self, name: str, spec: Mapping[str, Any]) -> None:
  84         self.api_name = name
  85         self._spec = spec
  86         if keyword.iskeyword(name):
  87             name += '_'
  88         super().__init__(
  89             name,
  90             inspect.Parameter.KEYWORD_ONLY,
  91             annotation=get_type_annotation(self._spec['type']),
  92             # In normal Python the presence of a default tells you whether or
  93             # not an argument is required. In the API the `required` flag tells
  94             # us that, and defaults are specified inconsistently. Don't show
  95             # defaults in the signature: it adds noise and makes things more
  96             # confusing for the reader about what's required and what's
  97             # optional. The docstring can explain in better detail, including
  98             # the default value.
  99             default=inspect.Parameter.empty,
 100         )
 101
 102     def default_value(self) -> object:
 103         try:
 104             src_value: str = self._spec['default']
 105         except KeyError:
 106             return None
 107         if src_value == 'true':
 108             return True
 109         elif src_value == 'false':
 110             return False
 111         elif src_value.isdigit():
 112             return int(src_value)
 113         else:
 114             return src_value
 115
 116     def is_required(self) -> bool:
 117         return self._spec['required']
 118
 119     def doc(self) -> str:
 120         default_value = self.default_value()
 121         if default_value is None:
 122             default_doc = ''
 123         else:
 124             default_doc = f" Default {default_value!r}."
 125         # If there is no description, use a zero-width space to help Markdown
 126         # parsers retain the definition list structure.
 127         description = self._spec['description'] or '\u200b'
 128         return f'''
 129 {self.api_name}: {self.annotation}
 130 : {description}{default_doc}
 131 '''
 132
 133
 134 class Method:
 135     def __init__(self, name: str, spec: Mapping[str, Any]) -> None:
 136         self.name = name
 137         self._spec = spec
 138         self._required_params = []
 139         self._optional_params = []
 140         for param_name, param_spec in spec['parameters'].items():
 141             param = Parameter(param_name, param_spec)
 142             if param.is_required():
 143                 param_list = self._required_params
 144             else:
 145                 param_list = self._optional_params
 146             param_list.append(param)
 147         self._required_params.sort(key=NAME_KEY)
 148         self._optional_params.sort(key=NAME_KEY)
 149
 150     def signature(self) -> inspect.Signature:
 151         parameters = [
 152             inspect.Parameter('self', inspect.Parameter.POSITIONAL_ONLY),
 153             *self._required_params,
 154             *self._optional_params,
 155         ]
 156         try:
 157             returns = get_type_annotation(self._spec['response']['$ref'])
 158         except KeyError:
 159             returns = 'dict[str, Any]'
 160         return inspect.Signature(parameters, return_annotation=returns)
 161
 162     def doc(self) -> str:
 163         doc_lines = [self._spec['description'].splitlines()[0], '\n']
 164         if self._required_params:
 165             doc_lines.append("\nRequired parameters:\n")
 166             doc_lines.extend(param.doc() for param in self._required_params)
 167         if self._optional_params:
 168             doc_lines.append("\nOptional parameters:\n")
 169             doc_lines.extend(param.doc() for param in self._optional_params)
 170         return re.sub(r'\n{3,}', '\n\n', f'''
 171     def {self.name}{self.signature()}:
 172 {to_docstring(''.join(doc_lines), 8)}
 173 ''')
 174
 175
 176 def document_schema(name: str, spec: Mapping[str, Any]) -> str:
 177     lines = [
 178         f"class {name}(TypedDict, total=False):",
 179         to_docstring(spec['description'], 4),
 180     ]
 181     for field_name, field_spec in spec['properties'].items():
 182         field_type = get_type_annotation(field_spec['type'])
 183         try:
 184             subtype = field_spec['items']['$ref']
 185         except KeyError:
 186             pass
 187         else:
 188             field_type += f"[{get_type_annotation(subtype)}]"
 189
 190         field_line = f"    {field_name}: {field_type!r}"
 191         try:
 192             field_line += f" = {field_spec['default']!r}"
 193         except KeyError:
 194             pass
 195         lines.append(field_line)
 196
 197         field_doc: str = field_spec.get('description', '')
 198         if field_spec['type'] == 'datetime':
 199             field_doc += "\n\nString in ISO 8601 datetime format. Pass it to `ciso8601.parse_datetime` to build a `datetime.datetime`."
 200         if field_doc:
 201             lines.append(to_docstring(field_doc, 4))
 202     lines.append('\n')
 203     return '\n'.join(lines)
 204
 205 def document_resource(name: str, spec: Mapping[str, Any]) -> str:
 206     methods = [Method(key, meth_spec) for key, meth_spec in spec['methods'].items()]
 207     return f'''class {classify_name(name)}:
 208     """Methods to query and manipulate Arvados {humanize_name(name)}"""
 209 {''.join(method.doc() for method in sorted(methods, key=NAME_KEY))}
 210 '''
 211
 212 def parse_arguments(arglist: Optional[Sequence[str]]) -> argparse.Namespace:
 213     parser = argparse.ArgumentParser()
 214     parser.add_argument(
 215         '--output-file', '-O',
 216         type=pathlib.Path,
 217         metavar='PATH',
 218         default=STDSTREAM_PATH,
 219         help="""Path to write output. Specify `-` to use stdout (the default)
 220 """)
 221     parser.add_argument(
 222         'discovery_url',
 223         nargs=argparse.OPTIONAL,
 224         metavar='URL',
 225         help="""URL or file path of a discovery document to load.
 226 Specify `-` to use stdin.
 227 If not provided, retrieved dynamically from Arvados client configuration.
 228 """)
 229     args = parser.parse_args(arglist)
 230     if args.discovery_url is None:
 231         from arvados.api import api_kwargs_from_config
 232         discovery_fmt = api_kwargs_from_config('v1')['discoveryServiceUrl']
 233         args.discovery_url = discovery_fmt.format(api='arvados', apiVersion='v1')
 234     elif args.discovery_url == '-':
 235         args.discovery_url = 'file:///dev/stdin'
 236     else:
 237         parts = urllib.parse.urlsplit(args.discovery_url)
 238         if not (parts.scheme or parts.netloc):
 239             args.discovery_url = urllib.parse.urlunsplit(parts._replace(scheme='file'))
 240     if args.output_file == STDSTREAM_PATH:
 241         args.out_file = sys.stdout
 242     else:
 243         args.out_file = args.output_file.open('w')
 244     return args
 245
 246 def main(arglist: Optional[Sequence[str]]=None) -> int:
 247     args = parse_arguments(arglist)
 248     with urllib.request.urlopen(args.discovery_url) as discovery_file:
 249         if not (discovery_file.status is None or 200 <= discovery_file.status < 300):
 250             print(
 251                 f"error getting {args.discovery_url}: server returned {discovery_file.status}",
 252                 file=sys.stderr,
 253             )
 254             return os.EX_IOERR
 255         discovery_document = json.load(discovery_file)
 256     print('''from typing import Any, TypedDict''', file=args.out_file)
 257
 258     schemas = sorted(discovery_document['schemas'].items())
 259     for name, schema_spec in schemas:
 260         print(document_schema(name, schema_spec), file=args.out_file)
 261
 262     resources = sorted(discovery_document['resources'].items())
 263     for name, resource_spec in resources:
 264         print(document_resource(name, resource_spec), file=args.out_file)
 265
 266     print('''class ArvadosAPIClient:''', file=args.out_file)
 267     for name, _ in resources:
 268         class_name = classify_name(name)
 269         method_spec = {
 270             'description': f"Return an instance of `{class_name}` to call methods via this client",
 271             'parameters': {},
 272             'response': {
 273                 '$ref': class_name,
 274             },
 275         }
 276         print(Method(name, method_spec).doc(), file=args.out_file)
 277
 278     return os.EX_OK
 279
 280 if __name__ == '__main__':
 281     sys.exit(main())