From 21b8dfb663451ebf71e1b1acb8333fc58e3b949e Mon Sep 17 00:00:00 2001 From: Isaac Jessup Date: Thu, 28 Sep 2023 08:33:08 -0400 Subject: [PATCH] operations: List... Adjusted the Reduce class to handle a list of operations rather than a dict of path and methods. A string or pattern can be used in the list to match against operationIds, or a tuple with a string or pattern as the first element representing the path and a list of strings representing HTTP verbs as the second element will match against the path/methods. Also, in any case that a path is not culled, the non-operation keys are kept intact to maintain defaults and overrides. The MSGraph test was updated to accommodate the annotation change as well as to handle a few validation fixes. --- aiopenapi3/extra.py | 88 +++++++++++++++++++++------------------------ tests/extra_test.py | 76 +++++++++++++++++++-------------------- 2 files changed, 77 insertions(+), 87 deletions(-) diff --git a/aiopenapi3/extra.py b/aiopenapi3/extra.py index f826885d..e451a756 100644 --- a/aiopenapi3/extra.py +++ b/aiopenapi3/extra.py @@ -1,5 +1,4 @@ -from re import Pattern -from typing import Dict, List, Union +from typing import List, Union, Optional, Tuple import logging import re @@ -13,55 +12,50 @@ class Reduce(Document, Init): log = logging.getLogger("aiopenapi3.extra.Reduce") - def __init__(self, operations: Dict[Union[str, Pattern], List[Union[str, Pattern]]]) -> None: - """ - :param operations: paths/methods to reduce to - """ - self.operations: List[Union[str, Pattern]] = operations + def __init__( + self, + operations: List[ + Union[Tuple[Union[re.Pattern, str], Optional[List[Union[re.Pattern, str]]]], Union[re.Pattern, str]] + ], + ) -> None: + self.operations = operations super().__init__() def _reduced_paths(self, ctx: "Document.Context") -> dict: - reduced_paths = {} - for path_key, path_value in ctx.document["paths"].items(): - # Extracting Non-Operation Objects - non_op_objects = { - key: val - for key, val in path_value.items() - if key in {"summary", "description", "servers", "parameters"} - } - - for operation_key, operation_value in path_value.items(): - if operation_key in non_op_objects: # Skip if the key is a Non-Operation Object - continue - - if not isinstance(operation_value, dict): - continue - - for pattern, operation_patterns in self.operations.items(): - # If pattern is None, look for operationId in operation_patterns - if pattern is None and operation_patterns is not None: - operation_id = operation_value.get("operationId", "") - if any( - op_pattern == operation_id - or (isinstance(op_pattern, Pattern) and re.match(op_pattern, operation_id)) - for op_pattern in operation_patterns - ): - reduced_paths.setdefault(path_key, {}).update(non_op_objects) - reduced_paths[path_key][operation_key] = operation_value - - else: - if ( - (isinstance(pattern, str) and pattern == path_key) - or (isinstance(pattern, Pattern) and re.match(pattern, path_key)) - ) and any( - op_pattern == operation_key - or (isinstance(op_pattern, Pattern) and re.match(op_pattern, operation_key)) - for op_pattern in operation_patterns or [operation_key] + reduced = {} + keep_keys = {"summary", "description", "servers", "parameters"} + for operation in self.operations: + if isinstance(operation, (str, re.Pattern)): + for path_key, path_value in ctx.document["paths"].items(): + for operation_key, operation_value in path_value.items(): + if "operationId" in operation_value and ( + (isinstance(operation, str) and operation == operation_value["operationId"]) + or ( + isinstance(operation, re.Pattern) + and re.match(operation, operation_value["operationId"]) + ) ): - reduced_paths.setdefault(path_key, {}).update(non_op_objects) - reduced_paths[path_key][operation_key] = operation_value - - return reduced_paths + if path_key not in reduced: + reduced[path_key] = {k: v for k, v in path_value.items() if k in keep_keys} + reduced[path_key][operation_key] = operation_value + else: + pattern, operation_patterns = operation + for path_key in ctx.document["paths"].keys(): + if (isinstance(pattern, str) and pattern == path_key) or ( + isinstance(pattern, re.Pattern) and re.match(pattern, path_key) + ): + reduced[path_key] = { + k: v + for k, v in ctx.document["paths"][path_key].items() + if k in keep_keys + or not operation_patterns + or any( + (isinstance(op_pattern, str) and op_pattern == k) + or (isinstance(op_pattern, re.Pattern) and re.match(op_pattern, k)) + for op_pattern in operation_patterns + ) + } + return reduced def parsed(self, ctx: "Document.Context") -> "Document.Context": """Parse the given context.""" diff --git a/tests/extra_test.py b/tests/extra_test.py index 13ee21d5..2d225465 100644 --- a/tests/extra_test.py +++ b/tests/extra_test.py @@ -11,7 +11,8 @@ from aiopenapi3 import OpenAPI from aiopenapi3.loader import FileSystemLoader -from aiopenapi3.extra import Cull, Reduce +from aiopenapi3.extra import Cull, Init, Reduce, Document +from typing import Dict class PetStoreReduced(Reduce): @@ -22,20 +23,37 @@ def __init__(self): class MSGraph: def __init__(self): super().__init__( - operations={ - "/me/profile": None, - re.compile(r"/me/sendMail.*"): None, - } + operations=[ + ("/me/profile", None), + (re.compile(r"/me/sendMail.*"), None), + "accessReviewDecisions.accessReviewDecision.ListAccessReviewDecision", + re.compile(r"drives.drive.items.driveItem.permissions.permission*"), + ] ) - def parsed(self, ctx): + @staticmethod + def _remove_parameter(document, path, parameter_name): + if document["paths"].get(path, {}).get("parameters"): + document["paths"][path]["parameters"] = [ + p for p in document["paths"][path]["parameters"] if p.get("name", "") != parameter_name + ] + + @staticmethod + def _drop_required(schema: Dict, requirement: str) -> None: + if "required" in schema: + schema["required"] = [i for i in schema["required"] if i != requirement] + if not schema["required"]: + del schema["required"] + + def parsed(self, ctx: "Document.Context") -> "Document.Context": # Drop massive unnecessary discriminator del ctx.document["components"]["schemas"]["microsoft.graph.entity"]["discriminator"] - # Run standard reduction process ctx = super().parsed(ctx) - - # Fix invalids + # Remove superfluous parameters + self._remove_parameter(ctx.document, "/applications(appId='{appId}')", "uniqueName") + self._remove_parameter(ctx.document, "/applications(uniqueName='{uniqueName}')", "appId") + # Fix parameter names for operation in ctx.document.get("paths", {}).values(): for details in operation.values(): # Check if parameters exist for this operation @@ -46,43 +64,21 @@ def parsed(self, ctx): # Check if description matches the desired format if description.strip() == "Usage: on='{on}'": parameter["name"] = "on" - - # Drop requirement for @odata.type since it's not actually required - for schema in ctx.document["components"]["schemas"].values(): - if "required" in schema: - schema["required"] = [i for i in schema["required"] if i != "@odata.type"] - if not schema["required"]: - del schema["required"] + if "content" in parameter.keys(): + parameter["schema"] = parameter["content"].get("application/json", {}).get("schema", {}) + del parameter["content"] + # Drop requirement for @odata.type since it's not actually enforced + for schema in ctx.document.get("components", {}).get("schemas", {}).values(): if isinstance(schema, dict): + self._drop_required(schema, "@odata.type") for s in schema.get("allOf", []): - if "required" in s: - s["required"] = [i for i in s["required"] if i != "@odata.type"] - if not s["required"]: - del s["required"] - - ctx.document.setdefault("security", []).append({"token": []}) - ctx.document.setdefault("components", {}).setdefault("securitySchemes", {}).setdefault( - "token", - {"type": "http", "scheme": "bearer", "bearerFormat": "JWT"}, - ) - - # Rebuild Tags - ctx.document["tags"] = [ - {"name": tag} - for tag in set( - tag - for operations in ctx.document.get("paths", {}).values() - for details in operations.values() - if isinstance(details, dict) - for tag in details.get("tags", []) - ) - ] - + self._drop_required(s, "@odata.type") return ctx class MSGraphCulled(MSGraph, Cull): - pass + def paths(self, ctx: "Init.Context") -> "Init.Context": + return ctx class MSGraphReduced(MSGraph, Reduce):