#!/usr/bin/env python # Licensed to Elasticsearch B.V under one or more agreements. # Elasticsearch B.V licenses this file to you under the Apache 2.0 License. # See the LICENSE file in the project root for more information import os import json import re import urllib3 from itertools import chain from functools import lru_cache import black from click.testing import CliRunner from pathlib import Path from jinja2 import Environment, FileSystemLoader, TemplateNotFound http = urllib3.PoolManager() # line to look for in the original source file SEPARATOR = " # AUTO-GENERATED-API-DEFINITIONS #" # global substitutions for python keywords SUBSTITUTIONS = {"type": "doc_type", "from": "from_"} # api path(s) BRANCH_NAME = "7.x" CODE_ROOT = Path(__file__).absolute().parent.parent BASE_PATH = ( CODE_ROOT.parent / "elasticsearch" / "rest-api-spec" / "src" / "main" / "resources" / "rest-api-spec" / "api" ) XPACK_PATH = ( CODE_ROOT.parent / "elasticsearch" / "x-pack" / "plugin" / "src" / "test" / "resources" / "rest-api-spec" / "api" ) jinja_env = Environment( loader=FileSystemLoader([CODE_ROOT / "utils" / "templates"]), trim_blocks=True, lstrip_blocks=True, ) def blacken(filename): runner = CliRunner() result = runner.invoke(black.main, [str(filename)]) assert result.exit_code == 0, result.output @lru_cache() def is_valid_url(url): return http.request("HEAD", url).status == 200 class Module: def __init__(self, namespace): self.namespace = namespace self._apis = [] self.parse_orig() def add(self, api): self._apis.append(api) def parse_orig(self): self.orders = [] self.header = "class C:" fname = CODE_ROOT / "elasticsearch" / "client" / f"{self.namespace}.py" if os.path.exists(fname): with open(fname) as f: content = f.read() header_lines = [] for line in content.split("\n"): header_lines.append(line) if line == SEPARATOR: break # no separator found else: header_lines = [] for line in content.split("\n"): header_lines.append(line) if line.startswith("class"): break self.header = "\n".join(header_lines) self.orders = re.findall( r'\n def ([a-z_]+)\(', content, re.MULTILINE ) def _position(self, api): try: return self.orders.index(api.name) except ValueError: return len(self.orders) def sort(self): self._apis.sort(key=self._position) def dump(self): self.sort() fname = CODE_ROOT / "elasticsearch" / "client" / f"{self.namespace}.py" with open(fname, "w") as f: f.write(self.header) for api in self._apis: f.write(api.to_python()) blacken(fname) class API: def __init__(self, namespace, name, definition): self.namespace = namespace self.name = name # overwrite the dict to maintain key order definition["params"] = { SUBSTITUTIONS.get(p, p): v for p, v in definition.get("params", {}).items() } self._def = definition self.description = "" self.doc_url = "" if isinstance(definition["documentation"], str): self.doc_url = definition["documentation"] else: # set as attribute so it may be overriden by Module.add self.description = definition["documentation"].get("description", "") self.doc_url = definition["documentation"].get("url", "") # Filter out bad URL refs like 'TODO' # and serve all docs over HTTPS. if self.doc_url: if not self.doc_url.startswith("http"): self.doc_url = "" if self.doc_url.startswith("http://"): self.doc_url = self.doc_url.replace("http://", "https://") # Try setting doc refs like 'current' and 'master' to our branches ref. if BRANCH_NAME is not None: revised_url = re.sub("/elasticsearch/reference/[^/]+/", f"/elasticsearch/reference/{BRANCH_NAME}/", self.doc_url) if is_valid_url(revised_url): self.doc_url = revised_url else: print(f"URL {revised_url!r}, falling back on {self.doc_url!r}") @property def all_parts(self): parts = {} for url in self._def["url"]["paths"]: parts.update(url.get("parts", {})) for p in parts: parts[p]["required"] = all( p in url.get("parts", {}) for url in self._def["url"]["paths"] ) # This piece of logic corresponds to calling # client.tasks.get() w/o a task_id which was erroneously # allowed in the 7.1 client library. This functionality # is deprecated and will be removed in 8.x. if self.namespace == "tasks" and self.name == "get": parts["task_id"]["required"] = False for k, sub in SUBSTITUTIONS.items(): if k in parts: parts[sub] = parts.pop(k) dynamic, components = self.url_parts def ind(item): try: return components.index(item[0]) except ValueError: return len(components) parts = dict(sorted(parts.items(), key=ind)) return parts @property def params(self): parts = self.all_parts params = self._def.get("params", {}) return chain( ((p, parts[p]) for p in parts if parts[p]["required"]), (("body", self.body),) if self.body else (), ((p, parts[p]) for p in parts if not parts[p]["required"] and p not in params), sorted(params.items(), key=lambda x: (x[0] not in parts, x[0])), ) @property def body(self): b = self._def.get("body", {}) if b: b.setdefault("required", False) return b @property def query_params(self): return ( k for k in sorted(self._def.get("params", {}).keys()) if k not in self.all_parts ) @property def path(self): return max( (path for path in self._def["url"]["paths"]), key=lambda p: len(re.findall(r"\{([^}]+)\}", p["path"])), ) @property def method(self): # To adhere to the HTTP RFC we shouldn't send # bodies in GET requests. default_method = self.path["methods"][0] if self.body and default_method == "GET" and "POST" in self.path["methods"]: return "POST" return default_method @property def url_parts(self): path = self.path["path"] dynamic = "{" in path if not dynamic: return dynamic, path parts = [] for part in path.split("/"): if not part: continue if part[0] == "{": part = part[1:-1] parts.append(SUBSTITUTIONS.get(part, part)) else: parts.append(f"'{part}'") return dynamic, parts @property def required_parts(self): parts = self.all_parts required = [p for p in parts if parts[p]["required"]] if self.body.get("required"): required.append("body") return required def to_python(self): try: t = jinja_env.get_template(f"overrides/{self.namespace}/{self.name}") except TemplateNotFound: t = jinja_env.get_template("base") return t.render( api=self, substitutions={v: k for k, v in SUBSTITUTIONS.items()} ) def read_modules(): modules = {} for path in (BASE_PATH, XPACK_PATH): for f in sorted(os.listdir(path)): name, ext = f.rsplit(".", 1) if ext != "json" or name == "_common": continue with open(path / f) as api_def: api = json.load(api_def)[name] namespace = "__init__" if "." in name: namespace, name = name.rsplit(".", 1) # The data_frame API has been changed to transform. if namespace == "data_frame_transform_deprecated": continue if namespace not in modules: modules[namespace] = Module(namespace) modules[namespace].add(API(namespace, name, api)) return modules def dump_modules(modules): for mod in modules.values(): mod.dump() if __name__ == "__main__": dump_modules(read_modules())