344 lines
12 KiB
Python
344 lines
12 KiB
Python
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
|
# not use this file except in compliance with the License. You may obtain
|
|
# a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
|
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
|
# License for the specific language governing permissions and limitations
|
|
# under the License.
|
|
#
|
|
import logging
|
|
import inspect
|
|
import re
|
|
from multiprocessing import Process
|
|
from pathlib import Path
|
|
|
|
from ruamel.yaml.scalarstring import LiteralScalarString
|
|
|
|
from codegenerator.common.schema import SpecSchema
|
|
from codegenerator.common.schema import PathSchema
|
|
from codegenerator.common.schema import ParameterSchema
|
|
from codegenerator.openapi import base
|
|
from codegenerator.openapi.base import OpenStackServerSourceBase
|
|
from codegenerator.openapi.utils import merge_api_ref_doc
|
|
from codegenerator.openapi.placement_schemas import resource_class
|
|
from codegenerator.openapi.placement_schemas import trait
|
|
from codegenerator.openapi.placement_schemas import reshaper
|
|
|
|
|
|
class PlacementGenerator(OpenStackServerSourceBase):
|
|
URL_TAG_MAP = {"/versions": "version"}
|
|
|
|
RESOURCE_MODULES = [reshaper, resource_class, trait]
|
|
|
|
VERSIONED_METHODS: dict = {}
|
|
|
|
def _api_ver_major(self, ver):
|
|
return ver.major
|
|
|
|
def _api_ver_minor(self, ver):
|
|
return ver.minor
|
|
|
|
def _api_ver(self, ver):
|
|
return (ver.major, ver.minor)
|
|
|
|
def _generate(self, target_dir, args):
|
|
from oslo_config import cfg
|
|
from oslo_config import fixture as config_fixture
|
|
|
|
from placement import microversion
|
|
from placement import handler
|
|
from placement import conf
|
|
|
|
self.api_version = microversion.max_version_string()
|
|
self.min_api_version = microversion.min_version_string()
|
|
self.VERSIONED_METHODS = microversion.VERSIONED_METHODS
|
|
|
|
config = cfg.ConfigOpts()
|
|
conf_fixture = self.useFixture(config_fixture.Config(config))
|
|
conf.register_opts(conf_fixture.conf)
|
|
handler = handler.PlacementHandler(config=conf_fixture.conf)
|
|
|
|
self.router = handler._map
|
|
|
|
work_dir = Path(target_dir)
|
|
work_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
impl_path = Path(
|
|
work_dir, "openapi_specs", "placement", f"v{self.api_version}.yaml"
|
|
)
|
|
impl_path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
openapi_spec = self.load_openapi(impl_path)
|
|
if not openapi_spec:
|
|
openapi_spec = SpecSchema(
|
|
info={
|
|
"title": "OpenStack Placement API",
|
|
"description": LiteralScalarString(
|
|
"Placement API provided by Placement service"
|
|
),
|
|
"version": self.api_version,
|
|
},
|
|
openapi="3.1.0",
|
|
security=[{"ApiKeyAuth": []}],
|
|
components={
|
|
"securitySchemes": {
|
|
"ApiKeyAuth": {
|
|
"type": "apiKey",
|
|
"in": "header",
|
|
"name": "X-Auth-Token",
|
|
}
|
|
}
|
|
},
|
|
)
|
|
|
|
for route in self.router.matchlist:
|
|
self._process_route(route, openapi_spec)
|
|
|
|
self._sanitize_param_ver_info(openapi_spec, self.min_api_version)
|
|
|
|
if args.api_ref_src:
|
|
merge_api_ref_doc(
|
|
openapi_spec, args.api_ref_src, allow_strip_version=False
|
|
)
|
|
|
|
self.dump_openapi(openapi_spec, impl_path, args.validate)
|
|
|
|
lnk = Path(impl_path.parent, "v1.yaml")
|
|
lnk.unlink(missing_ok=True)
|
|
lnk.symlink_to(impl_path.name)
|
|
|
|
return impl_path
|
|
|
|
def generate(self, target_dir, args):
|
|
proc = Process(target=self._generate, args=[target_dir, args])
|
|
proc.start()
|
|
proc.join()
|
|
if proc.exitcode != 0:
|
|
raise RuntimeError("Error generating Placement OpenAPI schema")
|
|
|
|
def _process_route(
|
|
self, route, openapi_spec, ver_prefix=None, framework=None
|
|
):
|
|
# Placement exposes "action" as controller in route defaults, all others - "controller"
|
|
if not ("controller" in route.defaults or "action" in route.defaults):
|
|
return
|
|
if "action" in route.defaults and "_methods" in route.defaults:
|
|
# placement 405 handler
|
|
return
|
|
# Path can be "/servers/{id}", but can be
|
|
# "/volumes/:volume_id/types/:(id)" - process
|
|
# according to the routes lib logic
|
|
path = ver_prefix if ver_prefix else ""
|
|
operation_spec = None
|
|
for part in route.routelist:
|
|
if isinstance(part, dict):
|
|
path += "{" + part["name"] + "}"
|
|
else:
|
|
path += part
|
|
|
|
if path == "":
|
|
# placement has "" path - see weird explanation in the placement source code
|
|
return
|
|
|
|
method = (
|
|
route.conditions.get("method", "GET")[0]
|
|
if route.conditions
|
|
else "GET"
|
|
)
|
|
|
|
controller = route.defaults.get("controller")
|
|
action = route.defaults.get("action")
|
|
logging.info(
|
|
"Path: %s; method: %s; operation: %s", path, method, action
|
|
)
|
|
|
|
versioned_methods = self.VERSIONED_METHODS
|
|
controller_actions = {}
|
|
if not controller and action and hasattr(action, "func"):
|
|
# Placement base app
|
|
framework = "placement"
|
|
controller = action
|
|
contr = action
|
|
action = None
|
|
|
|
closurevars = inspect.getclosurevars(controller.func)
|
|
nonlocals = closurevars.nonlocals
|
|
qualified_name: str | None = nonlocals.get("qualified_name")
|
|
if qualified_name:
|
|
action = qualified_name
|
|
|
|
else:
|
|
raise RuntimeError(f"Unsupported controller {controller}")
|
|
|
|
# Get Path elements
|
|
path_elements: list[str] = list(filter(None, path.split("/")))
|
|
if path_elements and base.VERSION_RE.match(path_elements[0]):
|
|
path_elements.pop(0)
|
|
operation_tags = self._get_tags_for_url(path)
|
|
|
|
# Build path parameters (/foo/{foo_id}/bar/{id} => $foo_id, $foo_bar_id)
|
|
# Since for same path we are here multiple times check presence of
|
|
# parameter before adding new params
|
|
path_params: list[ParameterSchema] = []
|
|
path_resource_names: list[str] = [
|
|
x.replace("-", "_")
|
|
for x in filter(lambda x: not x.startswith("{"), path_elements)
|
|
]
|
|
for path_element in path_elements:
|
|
if "{" in path_element:
|
|
param_name = path_element.strip("{}")
|
|
global_param_name = (
|
|
"_".join(path_resource_names) + f"_{param_name}"
|
|
)
|
|
|
|
param_ref_name = self._get_param_ref(
|
|
openapi_spec,
|
|
global_param_name,
|
|
param_name,
|
|
param_location="path",
|
|
path=path,
|
|
)
|
|
# Ensure reference to the param is in the path_params
|
|
if param_ref_name not in [k.ref for k in list(path_params)]:
|
|
path_params.append(ParameterSchema(ref=param_ref_name))
|
|
# Cleanup path_resource_names
|
|
# if len(path_resource_names) > 0 and VERSION_RE.match(path_resource_names[0]):
|
|
# # We should not have version prefix in the path_resource_names
|
|
# path_resource_names.pop(0)
|
|
if len(path_resource_names) == 0:
|
|
path_resource_names.append("root")
|
|
elif path_elements[-1].startswith("{"):
|
|
rn = path_resource_names[-1]
|
|
if rn.endswith("ies"):
|
|
rn = rn.replace("ies", "y")
|
|
elif rn.endswith("sses"):
|
|
rn = rn[:-2]
|
|
elif rn.endswith("statuses"):
|
|
rn = rn[:-2]
|
|
else:
|
|
rn = rn.rstrip("s")
|
|
path_resource_names[-1] = rn
|
|
|
|
# Set operationId
|
|
operation_id = re.sub(
|
|
r"^(/?v[0-9.]*/)",
|
|
"",
|
|
"/".join([x.strip("{}") for x in path_elements])
|
|
+ f":{method.lower()}", # noqa
|
|
)
|
|
|
|
if action in versioned_methods:
|
|
# Normal REST operation with version bounds
|
|
(start_version, end_version) = (None, None)
|
|
|
|
for versioned_method in sorted(
|
|
versioned_methods[action], key=lambda v: v[0]
|
|
):
|
|
start_version = versioned_method[0]
|
|
# end_version = versioned_method[1]
|
|
func = versioned_method[2]
|
|
|
|
# Placement defaults to max version when unset. Since for us
|
|
# setting of max_mv means deprecation ignore that.
|
|
if str(end_version) == str(self.api_version):
|
|
end_version = None
|
|
|
|
# Get the path/op spec only when we have
|
|
# something to fill in
|
|
path_spec = openapi_spec.paths.setdefault(
|
|
path, PathSchema(parameters=path_params)
|
|
)
|
|
operation_spec = getattr(path_spec, method.lower())
|
|
if not operation_spec.operationId:
|
|
operation_spec.operationId = operation_id
|
|
operation_spec.tags.extend(operation_tags)
|
|
operation_spec.tags = list(set(operation_spec.tags))
|
|
|
|
self.process_operation(
|
|
func,
|
|
openapi_spec,
|
|
operation_spec,
|
|
path_resource_names,
|
|
controller=controller,
|
|
method=method,
|
|
operation_name=action.split(".")[-1],
|
|
start_version=start_version,
|
|
end_version=end_version,
|
|
)
|
|
# NOTE(gtema): this looks weird, but
|
|
# placement has different MV functions for single operation.
|
|
# Would it have single function with schema decorators it
|
|
# would be easy to respect MV information. In current form
|
|
# there is no sense in processing them individually - it is
|
|
# single method in the end. Just getting jsonschema is tricky.
|
|
break
|
|
else:
|
|
if callable(controller.func):
|
|
func = controller.func
|
|
|
|
# Get the path/op spec only when we have
|
|
# something to fill in
|
|
path_spec = openapi_spec.paths.setdefault(
|
|
path, PathSchema(parameters=path_params)
|
|
)
|
|
operation_spec = getattr(path_spec, method.lower())
|
|
if not operation_spec.operationId:
|
|
operation_spec.operationId = operation_id
|
|
if operation_tags:
|
|
operation_spec.tags.extend(operation_tags)
|
|
operation_spec.tags = list(set(operation_spec.tags))
|
|
|
|
self.process_operation(
|
|
func,
|
|
openapi_spec,
|
|
operation_spec,
|
|
path_resource_names,
|
|
controller=controller,
|
|
operation_name=func.__name__,
|
|
method=method,
|
|
path=path,
|
|
)
|
|
|
|
return operation_spec
|
|
|
|
def _post_process_operation_hook(
|
|
self, openapi_spec, operation_spec, path: str | None = None
|
|
):
|
|
"""Hook to allow service specific generator to modify details"""
|
|
for resource_mod in self.RESOURCE_MODULES:
|
|
hook = getattr(resource_mod, "_post_process_operation_hook", None)
|
|
if hook:
|
|
hook(openapi_spec, operation_spec, path=path)
|
|
|
|
def _get_schema_ref(
|
|
self,
|
|
openapi_spec,
|
|
name,
|
|
description=None,
|
|
schema_def=None,
|
|
action_name=None,
|
|
):
|
|
# Invoke modularized schema _get_schema_ref
|
|
for resource_mod in self.RESOURCE_MODULES:
|
|
hook = getattr(resource_mod, "_get_schema_ref", None)
|
|
if hook:
|
|
(ref, mime_type, matched) = hook(
|
|
openapi_spec, name, description, schema_def, action_name
|
|
)
|
|
if matched:
|
|
return (ref, mime_type)
|
|
|
|
# Default
|
|
(ref, mime_type) = super()._get_schema_ref(
|
|
openapi_spec,
|
|
name,
|
|
description,
|
|
schema_def=schema_def,
|
|
action_name=action_name,
|
|
)
|
|
return (ref, mime_type)
|