zulip/zerver/lib/bugdown/api_code_examples.py

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)