mirror of
https://github.com/zulip/zulip.git
synced 2026-06-21 21:32:29 +08:00
c0712431df
This is required for the upcoming type behavior of the "anchor" parameter. This change is the minimal work required to have our OpenAPI code not fail when checking a union-type value of this form. We'll likely want to, in the future, do something nicer, but it'd require more extensive infrastructure for parsing of OpenAPI data that it's worth with our current approach (we may want to switch to using a library).
348 lines
13 KiB
Python
348 lines
13 KiB
Python
import re
|
|
import json
|
|
import inspect
|
|
|
|
from django.conf import settings
|
|
|
|
from markdown.extensions import Extension
|
|
from markdown.preprocessors import Preprocessor
|
|
from typing import Any, Dict, Optional, List, Tuple
|
|
import markdown
|
|
|
|
import zerver.openapi.python_examples
|
|
from zerver.lib.openapi import get_openapi_fixture, openapi_spec
|
|
|
|
MACRO_REGEXP = re.compile(r'\{generate_code_example(\(\s*(.+?)\s*\))*\|\s*(.+?)\s*\|\s*(.+?)\s*(\(\s*(.+)\s*\))?\}')
|
|
CODE_EXAMPLE_REGEX = re.compile(r'\# \{code_example\|\s*(.+?)\s*\}')
|
|
|
|
PYTHON_CLIENT_CONFIG = """
|
|
#!/usr/bin/env python3
|
|
|
|
import zulip
|
|
|
|
# Pass the path to your zuliprc file here.
|
|
client = zulip.Client(config_file="~/zuliprc")
|
|
|
|
"""
|
|
|
|
PYTHON_CLIENT_ADMIN_CONFIG = """
|
|
#!/usr/bin/env python
|
|
|
|
import zulip
|
|
|
|
# The user for this zuliprc file must be an organization administrator
|
|
client = zulip.Client(config_file="~/zuliprc-admin")
|
|
|
|
"""
|
|
|
|
DEFAULT_AUTH_EMAIL = "BOT_EMAIL_ADDRESS"
|
|
DEFAULT_AUTH_API_KEY = "BOT_API_KEY"
|
|
DEFAULT_EXAMPLE = {
|
|
"integer": 1,
|
|
"string": "demo",
|
|
"boolean": False,
|
|
}
|
|
|
|
def parse_language_and_options(input_str: Optional[str]) -> Tuple[str, Dict[str, Any]]:
|
|
if not input_str:
|
|
return ("", {})
|
|
language_and_options = re.match(r"(?P<language>\w+)(,\s*(?P<options>[\"\'\w\d\[\],= ]+))?", input_str)
|
|
assert(language_and_options is not None)
|
|
kwargs_pattern = re.compile(r"(?P<key>\w+)\s*=\s*(?P<value>[\'\"\w\d]+|\[[\'\",\w\d ]+\])")
|
|
language = language_and_options.group("language")
|
|
assert(language is not None)
|
|
if language_and_options.group("options"):
|
|
_options = kwargs_pattern.finditer(language_and_options.group("options"))
|
|
options = {}
|
|
for m in _options:
|
|
options[m.group("key")] = json.loads(m.group("value").replace("'", '"'))
|
|
return (language, options)
|
|
return (language, {})
|
|
|
|
def extract_python_code_example(source: List[str], snippet: List[str]) -> List[str]:
|
|
start = -1
|
|
end = -1
|
|
for line in source:
|
|
match = CODE_EXAMPLE_REGEX.search(line)
|
|
if match:
|
|
if match.group(1) == 'start':
|
|
start = source.index(line)
|
|
elif match.group(1) == 'end':
|
|
end = source.index(line)
|
|
break
|
|
|
|
if (start == -1 and end == -1):
|
|
return snippet
|
|
|
|
snippet.extend(source[start + 1: end])
|
|
snippet.append(' print(result)')
|
|
snippet.append('\n')
|
|
source = source[end + 1:]
|
|
return extract_python_code_example(source, snippet)
|
|
|
|
def render_python_code_example(function: str, admin_config: Optional[bool]=False,
|
|
**kwargs: Any) -> List[str]:
|
|
method = zerver.openapi.python_examples.TEST_FUNCTIONS[function]
|
|
function_source_lines = inspect.getsourcelines(method)[0]
|
|
|
|
if admin_config:
|
|
config = PYTHON_CLIENT_ADMIN_CONFIG.splitlines()
|
|
else:
|
|
config = PYTHON_CLIENT_CONFIG.splitlines()
|
|
|
|
snippet = extract_python_code_example(function_source_lines, [])
|
|
|
|
code_example = []
|
|
code_example.append('```python')
|
|
code_example.extend(config)
|
|
|
|
for line in snippet:
|
|
# Remove one level of indentation and strip newlines
|
|
code_example.append(line[4:].rstrip())
|
|
|
|
code_example.append('```')
|
|
|
|
return code_example
|
|
|
|
def curl_method_arguments(endpoint: str, method: str,
|
|
api_url: str) -> List[str]:
|
|
# We also include the -sS verbosity arguments here.
|
|
method = method.upper()
|
|
url = "{}/v1{}".format(api_url, endpoint)
|
|
valid_methods = ["GET", "POST", "DELETE", "PUT", "PATCH", "OPTIONS"]
|
|
if method == "GET":
|
|
# Then we need to make sure that each -d option translates to becoming
|
|
# a GET parameter (in the URL) and not a POST parameter (in the body).
|
|
# TODO: remove the -X part by updating the linting rule. It's redundant.
|
|
return ["-sSX", "GET", "-G", url]
|
|
elif method in valid_methods:
|
|
return ["-sSX", method, url]
|
|
else:
|
|
msg = "The request method {} is not one of {}".format(method,
|
|
valid_methods)
|
|
raise ValueError(msg)
|
|
|
|
def get_openapi_param_example_value_as_string(endpoint: str, method: str, param: Dict[str, Any],
|
|
curl_argument: bool=False) -> str:
|
|
if "type" in param["schema"]:
|
|
param_type = param["schema"]["type"]
|
|
else:
|
|
# Hack: Ideally, we'd extract a common function for handling
|
|
# oneOf values in types and do something with the resulting
|
|
# union type. But for this logic's purpose, it's good enough
|
|
# to just check the first parameter.
|
|
param_type = param["schema"]["oneOf"][0]["type"]
|
|
param_name = param["name"]
|
|
if param_type in ["object", "array"]:
|
|
example_value = param.get("example", None)
|
|
if not example_value:
|
|
msg = """All array and object type request parameters must have
|
|
concrete examples. The openAPI documentation for {}/{} is missing an example
|
|
value for the {} parameter. Without this we cannot automatically generate a
|
|
cURL example.""".format(endpoint, method, param_name)
|
|
raise ValueError(msg)
|
|
ordered_ex_val_str = json.dumps(example_value, sort_keys=True)
|
|
if curl_argument:
|
|
return " --data-urlencode {}='{}'".format(param_name, ordered_ex_val_str)
|
|
return ordered_ex_val_str # nocoverage
|
|
else:
|
|
example_value = param.get("example", DEFAULT_EXAMPLE[param_type])
|
|
if type(example_value) == bool:
|
|
example_value = str(example_value).lower()
|
|
if param["schema"].get("format", "") == "json":
|
|
example_value = json.dumps(example_value)
|
|
if curl_argument:
|
|
return " -d '{}={}'".format(param_name, example_value)
|
|
return example_value
|
|
|
|
def generate_curl_example(endpoint: str, method: str,
|
|
api_url: str,
|
|
auth_email: str=DEFAULT_AUTH_EMAIL,
|
|
auth_api_key: str=DEFAULT_AUTH_API_KEY,
|
|
exclude: Optional[List[str]]=None,
|
|
include: Optional[List[str]]=None) -> List[str]:
|
|
if exclude is not None and include is not None:
|
|
raise AssertionError("exclude and include cannot be set at the same time.")
|
|
|
|
lines = ["```curl"]
|
|
operation = endpoint + ":" + method.lower()
|
|
operation_entry = openapi_spec.spec()['paths'][endpoint][method.lower()]
|
|
global_security = openapi_spec.spec()['security']
|
|
|
|
operation_params = operation_entry.get("parameters", [])
|
|
operation_request_body = operation_entry.get("requestBody", None)
|
|
operation_security = operation_entry.get("security", None)
|
|
|
|
if settings.RUNNING_OPENAPI_CURL_TEST: # nocoverage
|
|
from zerver.openapi.curl_param_value_generators import patch_openapi_example_values
|
|
operation_params, operation_request_body = patch_openapi_example_values(operation, operation_params,
|
|
operation_request_body)
|
|
|
|
format_dict = {}
|
|
for param in operation_params:
|
|
if param["in"] != "path":
|
|
continue
|
|
example_value = get_openapi_param_example_value_as_string(endpoint, method, param)
|
|
format_dict[param["name"]] = example_value
|
|
example_endpoint = endpoint.format_map(format_dict)
|
|
|
|
curl_first_line_parts = ["curl"] + curl_method_arguments(example_endpoint, method,
|
|
api_url)
|
|
lines.append(" ".join(curl_first_line_parts))
|
|
|
|
insecure_operations = ['/dev_fetch_api_key:post']
|
|
if operation_security is None:
|
|
if global_security == [{'basicAuth': []}]:
|
|
authentication_required = True
|
|
else:
|
|
raise AssertionError("Unhandled global securityScheme. Please update the code to handle this scheme.")
|
|
elif operation_security == []:
|
|
if operation in insecure_operations:
|
|
authentication_required = False
|
|
else:
|
|
raise AssertionError("Unknown operation without a securityScheme. Please update insecure_operations.")
|
|
else:
|
|
raise AssertionError("Unhandled securityScheme. Please update the code to handle this scheme.")
|
|
|
|
if authentication_required:
|
|
lines.append(" -u %s:%s" % (auth_email, auth_api_key))
|
|
|
|
for param in operation_params:
|
|
if param["in"] == "path":
|
|
continue
|
|
param_name = param["name"]
|
|
|
|
if include is not None and param_name not in include:
|
|
continue
|
|
|
|
if exclude is not None and param_name in exclude:
|
|
continue
|
|
|
|
example_value = get_openapi_param_example_value_as_string(endpoint, method, param,
|
|
curl_argument=True)
|
|
lines.append(example_value)
|
|
|
|
if "requestBody" in operation_entry:
|
|
properties = operation_entry["requestBody"]["content"]["multipart/form-data"]["schema"]["properties"]
|
|
for key, property in properties.items():
|
|
lines.append(' -F "{}=@{}"'.format(key, property["example"]))
|
|
|
|
for i in range(1, len(lines)-1):
|
|
lines[i] = lines[i] + " \\"
|
|
|
|
lines.append("```")
|
|
|
|
return lines
|
|
|
|
def render_curl_example(function: str, api_url: str,
|
|
exclude: Optional[List[str]]=None,
|
|
include: Optional[List[str]]=None) -> List[str]:
|
|
""" A simple wrapper around generate_curl_example. """
|
|
parts = function.split(":")
|
|
endpoint = parts[0]
|
|
method = parts[1]
|
|
kwargs = dict() # type: Dict[str, Any]
|
|
if len(parts) > 2:
|
|
kwargs["auth_email"] = parts[2]
|
|
if len(parts) > 3:
|
|
kwargs["auth_api_key"] = parts[3]
|
|
kwargs["api_url"] = api_url
|
|
kwargs["exclude"] = exclude
|
|
kwargs["include"] = include
|
|
return generate_curl_example(endpoint, method, **kwargs)
|
|
|
|
SUPPORTED_LANGUAGES = {
|
|
'python': {
|
|
'client_config': PYTHON_CLIENT_CONFIG,
|
|
'admin_config': PYTHON_CLIENT_ADMIN_CONFIG,
|
|
'render': render_python_code_example,
|
|
},
|
|
'curl': {
|
|
'render': render_curl_example
|
|
}
|
|
} # type: Dict[str, Any]
|
|
|
|
class APICodeExamplesGenerator(Extension):
|
|
def __init__(self, api_url: Optional[str]) -> None:
|
|
self.config = {
|
|
'api_url': [
|
|
api_url,
|
|
'API URL to use when rendering curl examples'
|
|
]
|
|
}
|
|
|
|
def extendMarkdown(self, md: markdown.Markdown, md_globals: Dict[str, Any]) -> None:
|
|
md.preprocessors.add(
|
|
'generate_code_example', APICodeExamplesPreprocessor(md, self.getConfigs()), '_begin'
|
|
)
|
|
|
|
class APICodeExamplesPreprocessor(Preprocessor):
|
|
def __init__(self, md: markdown.Markdown, config: Dict[str, Any]) -> None:
|
|
super(APICodeExamplesPreprocessor, self).__init__(md)
|
|
self.api_url = config['api_url']
|
|
|
|
def run(self, lines: List[str]) -> List[str]:
|
|
done = False
|
|
while not done:
|
|
for line in lines:
|
|
loc = lines.index(line)
|
|
match = MACRO_REGEXP.search(line)
|
|
|
|
if match:
|
|
language, options = parse_language_and_options(match.group(2))
|
|
function = match.group(3)
|
|
key = match.group(4)
|
|
argument = match.group(6)
|
|
if self.api_url is None:
|
|
raise AssertionError("Cannot render curl API examples without API URL set.")
|
|
options['api_url'] = self.api_url
|
|
|
|
if key == 'fixture':
|
|
if argument:
|
|
text = self.render_fixture(function, name=argument)
|
|
else:
|
|
text = self.render_fixture(function)
|
|
elif key == 'example':
|
|
if argument == 'admin_config=True':
|
|
text = SUPPORTED_LANGUAGES[language]['render'](function, admin_config=True)
|
|
else:
|
|
text = SUPPORTED_LANGUAGES[language]['render'](function, **options)
|
|
|
|
# The line that contains the directive to include the macro
|
|
# may be preceded or followed by text or tags, in that case
|
|
# we need to make sure that any preceding or following text
|
|
# stays the same.
|
|
line_split = MACRO_REGEXP.split(line, maxsplit=0)
|
|
preceding = line_split[0]
|
|
following = line_split[-1]
|
|
text = [preceding] + text + [following]
|
|
lines = lines[:loc] + text + lines[loc+1:]
|
|
break
|
|
else:
|
|
done = True
|
|
return lines
|
|
|
|
def render_fixture(self, function: str, name: Optional[str]=None) -> List[str]:
|
|
fixture = []
|
|
|
|
# We assume that if the function we're rendering starts with a slash
|
|
# it's a path in the endpoint and therefore it uses the new OpenAPI
|
|
# format.
|
|
if function.startswith('/'):
|
|
path, method = function.rsplit(':', 1)
|
|
fixture_dict = get_openapi_fixture(path, method, name)
|
|
else:
|
|
fixture_dict = zerver.openapi.python_examples.FIXTURES[function]
|
|
|
|
fixture_json = json.dumps(fixture_dict, indent=4, sort_keys=True,
|
|
separators=(',', ': '))
|
|
|
|
fixture.append('```')
|
|
fixture.extend(fixture_json.splitlines())
|
|
fixture.append('```')
|
|
|
|
return fixture
|
|
|
|
def makeExtension(*args: Any, **kwargs: str) -> APICodeExamplesGenerator:
|
|
return APICodeExamplesGenerator(*args, **kwargs)
|